api: serve PDFs — extraction in the child under the pool, full text or map, errors
Route "pdf" through clean.Kind in doFetch/cleanFetched: extract under the pool (keyed by caller class + document hash, one per caller), map the five pdftext errors plus a pool panic to 422, ErrBusy to 503+Retry-After 10, ErrCallerBusy to 429, and answer with the whole text or the map. Adds RetryAfter to fetch.Error and the matching header. No cache, pages or grep — those are task 07. Co-Authored-By: ornith-1.5-35b-a3b <ornith-1.5-35b-a3b@llama.cpp.invalid>
This commit is contained in:
@@ -9,3 +9,4 @@ One row per task session that ends in a commit. Newest last.
|
||||
| p5a/03-dcache | 2026-09-25 | done | 1 | gate: ok | evict() tie-break | Added the pdf_doc/pdf_page tables, `PDFEntry`/`PDFKey`/`PutPDF`/`GetPDF`/`TouchPDF` (one-tx write deleting old pages first, page-ordered read, 304 touch moves only fetched_at), PDFH/PDFM counters, `Size()` including pdf_doc nbytes. One deviation: extended the eviction union with `pdf_doc` and added `rowid ASC` to the ORDER BY so same-second puts evict in insertion order (deterministic whole-document eviction); without it TestPDFEviction is order-dependent. Left Stats() signature unchanged for task 07. Given test byte-identical. |
|
||||
| p5a/04-pdfpool | 2026-09-25 | done | 1 | gate: ok | none | Implemented the generic scheduler. `Pool` holds one mutex, a `jobs` map (key→in-flight job), a `callerJobs` map (caller→their job), and a FIFO `waiters` line. `Do` applies the six rules in order: join (counted in `Waiting`, no slot, allowed while the caller holds another job, `fn` may be nil), caller-busy, then slot/queue/refuse. A job's result is broadcast to every sharer via a `done` channel; results are not cached (key forgotten on finish). `fn` runs under `context.WithoutCancel` so the flight outlives a caller that leaves. Panic in `fn` is recovered into an error, freed the slot and caller. One implementation finding (not a plan deviation): queued jobs must not be counted in `Running()`, so `startLocked` only registers a job and a separate `takeSlotLocked`/`grant` increments the slot count — abandoning a queued starter (`abandon`) releases sharers without touching the slot. Given test byte-identical. |
|
||||
| p5a/05-pdfdoc | 2026-09-25 | done | 1 | gate: ok | none | Implemented the render package with no I/O on a `*pdftext.Doc`. `Marker`/`Full` print each extracted page as `--- page N ---`, text trimmed, joined by a blank line. `Render` dispatches grep → page range → whole. Whole fits in OutCap runes, else a map: title (or `Untitled PDF`)/pages/`~tokens` line with `pages 1-E extracted` only when fewer extracted than `Pages`, a cuttable `## Outline` (two-space indent per level below 1, cut before the header/instruction), the instruction line, and whole leading pages within OutCap/4 (at least page 1, cut to fit). Range parses `a-b`/`a`/`a-`, wraps out-of-range in `ErrBadPages` with `pages out of range (1-E)`, clamps `b`, caps at 100 pages and OutCap runes, and notes the rest as `pages="k-b"`. Grep is RE2 with smart-case `(?i)` prefix, 200-byte cap, per-line matches with previous/next page context in `p. P-`/`p. P:` form, first 100 shown, total counted; RE2 keeps the `(a+)+$` test linear. Every length is in runes. Given test byte-identical; make gate ok. |
|
||||
| p5a/06-api-pdf | 2026-09-26 | done | resumed | gate: ok (7 of 10 PDF tests; 3 are task 07's) | none | Served PDFs through `/v1/fetch` (clean) and `/v2/scrape`. Added `RetryAfter int` to `fetch.Error` and a `Retry-After` header line guarded by `> 0`. Added `PDFRunner`/`PDFPool`/`PDFTTL`/`PDFMaxQuestionChunks` Server fields (PDFPool built once via `sync.Once` only when nil, so the test's direct override is respected) and `pdfPoolOnce`. Routed `clean.Kind` in `doFetch`/`cleanFetched` (html/json/text switched, "other"→415, "pdf"→extractPDF); `doFetch`'s switch routes "text" explicitly so a PDF never runs `clean.Text` first. New `internal/api/pdf.go`: `DefaultPDFTTL`/`DefaultPDFMaxQuestionChunks` (task 07), `runner()`/`pool()` defaults, `extractPDF` (pool key `callerClass|SHA256`, two-value assert, `fmt.Errorf("pdf: bad result %T")`), `pdfError` (422 for the five pdftext errors + panic default, 503+RetryAfter 10 for ErrBusy, 429 for ErrCallerBusy), `renderPDF` (`pdfdoc.Render`, whole text or map). `doFetch` returns the 4th `doc *pdftext.Doc`; digest uses `pdfdoc.Full(doc)` (computed once) as content, clean uses `renderPDF.Content`, envelope `pdf` object has pages/title/outline/extracted/first/last/map/grep_hits/note, raw mode→400. Scrape markdown=`renderPDF.Content`, html="", `metadata.numPages`=doc.Pages. 7 of 10 given tests pass under -race -count=3; the 3 failures (`TestPDFPagesAndGrep`, `TestPDFCache`, `TestPDFQuestionTooLong`) are task 07 tests that reference fields not in this task, exactly as the task doc (lines 145, 157-158) predicts; nothing else fails. Given files byte-identical. |
|
||||
|
||||
+110
-39
@@ -10,10 +10,12 @@ import (
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
"unicode/utf8"
|
||||
|
||||
"git.wntrmute.dev/kyle/sift/internal/auth"
|
||||
"git.wntrmute.dev/kyle/sift/internal/clean"
|
||||
@@ -23,6 +25,9 @@ import (
|
||||
"git.wntrmute.dev/kyle/sift/internal/fleetctx"
|
||||
"git.wntrmute.dev/kyle/sift/internal/guard"
|
||||
"git.wntrmute.dev/kyle/sift/internal/metrics"
|
||||
"git.wntrmute.dev/kyle/sift/internal/pdfdoc"
|
||||
"git.wntrmute.dev/kyle/sift/internal/pdfpool"
|
||||
"git.wntrmute.dev/kyle/sift/internal/pdftext"
|
||||
"git.wntrmute.dev/kyle/sift/internal/recover"
|
||||
"git.wntrmute.dev/kyle/sift/internal/search"
|
||||
"git.wntrmute.dev/kyle/sift/internal/selfip"
|
||||
@@ -44,8 +49,13 @@ type Server struct {
|
||||
OutCap int
|
||||
SelfIP *selfip.Discoverer
|
||||
Digest *digest.Engine
|
||||
Search *search.Engine
|
||||
Ctx *fleetctx.Client
|
||||
|
||||
PDFRunner *pdftext.Runner
|
||||
PDFPool *pdfpool.Pool
|
||||
PDFTTL time.Duration
|
||||
PDFMaxQuestionChunks int
|
||||
Search *search.Engine
|
||||
Ctx *fleetctx.Client
|
||||
// PermitRestricted is context.permit_restricted: those callers never
|
||||
// have a fetch host written to the log (host=-). Independent of Honcho.
|
||||
PermitRestricted []string
|
||||
@@ -56,6 +66,8 @@ type Server struct {
|
||||
per map[string]chan struct{}
|
||||
|
||||
reqN, blockN, scrapeN atomic.Uint64
|
||||
|
||||
pdfPoolOnce sync.Once
|
||||
}
|
||||
|
||||
func New(version string, g *guard.Guard, f *fetch.Fetcher, keys *auth.Store, log *slog.Logger) *Server {
|
||||
@@ -300,6 +312,10 @@ type v1Req struct {
|
||||
Context *bool `json:"context"`
|
||||
JQ string `json:"jq"`
|
||||
Recover bool `json:"recover"`
|
||||
|
||||
// Pages narrows a PDF digest to "2-3"; Grep matches lines in it. Optional; task-07.
|
||||
Pages string `json:"pages"`
|
||||
Grep string `json:"grep"`
|
||||
}
|
||||
|
||||
func (s *Server) caller(w http.ResponseWriter, r *http.Request) *auth.Caller {
|
||||
@@ -357,12 +373,23 @@ func (s *Server) scrape(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
}
|
||||
s.scrapeN.Add(1)
|
||||
res, cl, ferr := s.doFetch(r, c, req.URL, "", onlyMain)
|
||||
res, cl, doc, ferr := s.doFetch(r, c, req.URL, "", onlyMain)
|
||||
if ferr != nil {
|
||||
s.writeFetchErr(w, c, req.URL, ferr)
|
||||
return
|
||||
}
|
||||
md, htmlOut := cl.Markdown, ""
|
||||
pages := 0
|
||||
if doc != nil {
|
||||
rp, err := s.renderPDF(doc)
|
||||
if err != nil {
|
||||
s.writeFetchErr(w, c, req.URL, err)
|
||||
return
|
||||
}
|
||||
md = rp.Content
|
||||
htmlOut = ""
|
||||
pages = doc.Pages
|
||||
}
|
||||
if wantHTML {
|
||||
htmlOut = cl.HTML
|
||||
}
|
||||
@@ -371,7 +398,7 @@ func (s *Server) scrape(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
rawTok := len(res.Body) / 4
|
||||
retTok := len(md) / 4
|
||||
s.Log.Info("scrape", "caller", c.Name, "host", s.logHost(c, res.FinalURL), "status", res.Status, "bytes", len(res.Body))
|
||||
s.Log.Info("scrape", "caller", c.Name, "host", s.logHost(c, res.FinalURL), "status", res.Status, "bytes", len(res.Body), "pdf_pages", pages)
|
||||
writeJSON(w, 200, map[string]any{
|
||||
"success": true,
|
||||
"data": map[string]any{
|
||||
@@ -382,6 +409,7 @@ func (s *Server) scrape(w http.ResponseWriter, r *http.Request) {
|
||||
"sourceURL": res.FinalURL,
|
||||
"statusCode": res.Status,
|
||||
"contentType": res.ContentType,
|
||||
"numPages": pages,
|
||||
"sift": map[string]any{
|
||||
"hash": res.SHA256(),
|
||||
"fetchedAt": res.FetchedAt.Format(time.RFC3339),
|
||||
@@ -429,12 +457,12 @@ func (s *Server) v1fetch(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
t0 := time.Now()
|
||||
onlyMain := true
|
||||
res, cl, ferr := s.doFetch(r, c, req.URL, req.JQ, onlyMain)
|
||||
res, cl, doc, ferr := s.doFetch(r, c, req.URL, req.JQ, onlyMain)
|
||||
var snap *recover.Result
|
||||
if ferr != nil && req.Recover && recoverWorthy(ferr) {
|
||||
sr, rerr := recover.Try(r.Context(), s.Fetch, req.URL)
|
||||
if rerr == nil && sr != nil && sr.Res != nil {
|
||||
res, cl, ferr = s.cleanFetched(sr.Res, req.JQ, onlyMain)
|
||||
res, cl, doc, ferr = s.cleanFetched(sr.Res, req.JQ, onlyMain)
|
||||
snap = sr
|
||||
}
|
||||
}
|
||||
@@ -444,16 +472,50 @@ func (s *Server) v1fetch(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
fetchMS := time.Since(t0).Milliseconds()
|
||||
|
||||
content := cl.Markdown
|
||||
if cl.Kind == "json" {
|
||||
content = cl.HTML
|
||||
}
|
||||
if mode == "raw" {
|
||||
content = string(res.Body)
|
||||
if len(content) > s.OutCap {
|
||||
content = clean.TruncateUTF8(content, s.OutCap)
|
||||
cl.Trunc = true
|
||||
cl.Warning = "truncated to output cap"
|
||||
var docLen, pdfPages int
|
||||
var content string
|
||||
var pdfMap map[string]any
|
||||
if doc != nil {
|
||||
pdfPages = doc.Pages
|
||||
if mode == "raw" {
|
||||
writeErr(w, 400, "raw mode does not apply to PDFs", "")
|
||||
return
|
||||
}
|
||||
rp, err := s.renderPDF(doc)
|
||||
if err != nil {
|
||||
s.writeFetchErr(w, c, req.URL, err)
|
||||
return
|
||||
}
|
||||
full := pdfdoc.Full(doc)
|
||||
docLen = utf8.RuneCountInString(full)
|
||||
if mode == "digest" {
|
||||
content = full
|
||||
} else {
|
||||
content = rp.Content
|
||||
}
|
||||
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,
|
||||
}
|
||||
} else {
|
||||
content = cl.Markdown
|
||||
if cl.Kind == "json" {
|
||||
content = cl.HTML
|
||||
}
|
||||
if mode == "raw" {
|
||||
content = string(res.Body)
|
||||
if len(content) > s.OutCap {
|
||||
content = clean.TruncateUTF8(content, s.OutCap)
|
||||
cl.Trunc = true
|
||||
cl.Warning = "truncated to output cap"
|
||||
}
|
||||
}
|
||||
}
|
||||
warn := []string{}
|
||||
@@ -518,7 +580,7 @@ func (s *Server) v1fetch(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
s.Log.Info("fetch", "caller", c.Name, "host", s.logHost(c, res.FinalURL), "mode", mode,
|
||||
"q_len", qLen, "raw_tokens", len(res.Body)/4, "returned_tokens", len(content)/4,
|
||||
"digest_ms", digestMS, "fallback", fb)
|
||||
"pdf_pages", pdfPages, "digest_ms", digestMS, "fallback", fb)
|
||||
|
||||
src := map[string]any{
|
||||
"url": res.URL,
|
||||
@@ -549,12 +611,15 @@ func (s *Server) v1fetch(w http.ResponseWriter, r *http.Request) {
|
||||
"source": src,
|
||||
"warnings": warn,
|
||||
"sizes": map[string]int{
|
||||
"raw_tokens": len(res.Body) / 4,
|
||||
"raw_tokens": docLen / 4,
|
||||
"clean_tokens": digest.EstTokens(cl.Markdown),
|
||||
"returned_tokens": digest.EstTokens(content),
|
||||
},
|
||||
"timings": map[string]int64{"fetch_ms": fetchMS, "digest_ms": digestMS, "context_ms": ctxMS},
|
||||
}
|
||||
if pdfMap != nil {
|
||||
out["pdf"] = pdfMap
|
||||
}
|
||||
if digMeta != nil {
|
||||
out["digest"] = digMeta
|
||||
}
|
||||
@@ -585,13 +650,11 @@ func headings(md string) []string {
|
||||
return h
|
||||
}
|
||||
|
||||
func (s *Server) cleanFetched(res *fetch.Resource, jq string, onlyMain bool) (*fetch.Resource, *clean.Result, error) {
|
||||
kind := clean.MediaKind(res.ContentType)
|
||||
if kind == "sniff" {
|
||||
kind = clean.Sniff(res.Body)
|
||||
}
|
||||
if kind == "other" {
|
||||
return nil, nil, &fetch.Error{Status: 415, Hint: "browser", Msg: "unsupported content type"}
|
||||
func (s *Server) cleanFetched(res *fetch.Resource, jq string, onlyMain bool) (*fetch.Resource, *clean.Result, *pdftext.Doc, error) {
|
||||
kind := clean.Kind(res.ContentType, res.Body)
|
||||
if kind == "other" || kind == "pdf" {
|
||||
// The recover path (archive snapshots) never serves a PDF.
|
||||
return nil, nil, nil, &fetch.Error{Status: 415, Hint: "browser", Msg: "unsupported content type"}
|
||||
}
|
||||
var cl *clean.Result
|
||||
var cerr error
|
||||
@@ -604,31 +667,28 @@ func (s *Server) cleanFetched(res *fetch.Resource, jq string, onlyMain bool) (*f
|
||||
cl, cerr = clean.Text(res, s.OutCap)
|
||||
}
|
||||
if cerr != nil {
|
||||
return nil, nil, &fetch.Error{Status: 422, Msg: cerr.Error()}
|
||||
return nil, nil, nil, &fetch.Error{Status: 422, Msg: cerr.Error()}
|
||||
}
|
||||
return res, cl, nil
|
||||
return res, cl, nil, nil
|
||||
}
|
||||
|
||||
func (s *Server) doFetch(r *http.Request, c *auth.Caller, rawURL, jq string, onlyMain bool) (*fetch.Resource, *clean.Result, error) {
|
||||
func (s *Server) doFetch(r *http.Request, c *auth.Caller, rawURL, jq string, onlyMain bool) (*fetch.Resource, *clean.Result, *pdftext.Doc, error) {
|
||||
if err := s.acquire(c.Name, rawURL); err != nil {
|
||||
return nil, nil, err
|
||||
return nil, nil, nil, err
|
||||
}
|
||||
defer s.release(c.Name, rawURL)
|
||||
|
||||
res, err := s.Fetch.Fetch(r.Context(), rawURL)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
kind := clean.MediaKind(res.ContentType)
|
||||
if kind == "sniff" {
|
||||
kind = clean.Sniff(res.Body)
|
||||
return nil, nil, nil, err
|
||||
}
|
||||
kind := clean.Kind(res.ContentType, res.Body)
|
||||
if kind == "other" {
|
||||
ct := res.ContentType
|
||||
if ct == "" {
|
||||
ct = "unknown"
|
||||
}
|
||||
return nil, nil, &fetch.Error{Status: 415, Hint: "browser", Msg: "unsupported content type " + ct + "; hint: browser"}
|
||||
return nil, nil, nil, &fetch.Error{Status: 415, Hint: "browser", Msg: "unsupported content type " + ct + "; hint: browser"}
|
||||
}
|
||||
var cl *clean.Result
|
||||
var cerr error
|
||||
@@ -637,17 +697,25 @@ func (s *Server) doFetch(r *http.Request, c *auth.Caller, rawURL, jq string, onl
|
||||
cl, cerr = clean.HTML(res, onlyMain, s.OutCap)
|
||||
case "json":
|
||||
cl, cerr = clean.JSON(res, jq, s.OutCap)
|
||||
default:
|
||||
case "text":
|
||||
cl, cerr = clean.Text(res, s.OutCap)
|
||||
}
|
||||
if cerr != nil {
|
||||
var tl *clean.TooLargeError
|
||||
if errors.As(cerr, &tl) {
|
||||
return nil, nil, &fetch.Error{Status: 400, Msg: tl.Error()}
|
||||
return nil, nil, nil, &fetch.Error{Status: 400, Msg: tl.Error()}
|
||||
}
|
||||
return nil, nil, &fetch.Error{Status: 422, Msg: cerr.Error()}
|
||||
return nil, nil, nil, &fetch.Error{Status: 422, Msg: cerr.Error()}
|
||||
}
|
||||
return res, cl, nil
|
||||
var doc *pdftext.Doc
|
||||
if kind == "pdf" {
|
||||
doc, err = s.extractPDF(r.Context(), c, res)
|
||||
if err != nil {
|
||||
return nil, nil, nil, err
|
||||
}
|
||||
cl = &clean.Result{Kind: "pdf", Title: doc.Title}
|
||||
}
|
||||
return res, cl, doc, nil
|
||||
}
|
||||
|
||||
func (s *Server) writeFetchErr(w http.ResponseWriter, c *auth.Caller, rawURL string, err error) {
|
||||
@@ -670,6 +738,9 @@ func (s *Server) writeFetchErr(w http.ResponseWriter, c *auth.Caller, rawURL str
|
||||
} else {
|
||||
s.Log.Warn("fetch failed", "caller", name, "host", host, "reason", s.logReason(c, fe), "status", fe.Status)
|
||||
}
|
||||
if fe.RetryAfter > 0 {
|
||||
w.Header().Set("Retry-After", strconv.Itoa(fe.RetryAfter))
|
||||
}
|
||||
writeErr(w, fe.Status, fe.Msg, fe.Hint)
|
||||
return
|
||||
}
|
||||
|
||||
@@ -0,0 +1,86 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"git.wntrmute.dev/kyle/sift/internal/auth"
|
||||
"git.wntrmute.dev/kyle/sift/internal/fetch"
|
||||
"git.wntrmute.dev/kyle/sift/internal/pdfdoc"
|
||||
"git.wntrmute.dev/kyle/sift/internal/pdfpool"
|
||||
"git.wntrmute.dev/kyle/sift/internal/pdftext"
|
||||
)
|
||||
|
||||
const (
|
||||
// DefaultPDFTTL bounds a cached PDF's life. Task 07; declared now so the test compiles.
|
||||
DefaultPDFTTL = 24 * time.Hour
|
||||
// DefaultPDFMaxQuestionChunks caps a digest question's length. Task 07; declared now.
|
||||
DefaultPDFMaxQuestionChunks = 200
|
||||
)
|
||||
|
||||
// runner returns the configured extraction child, or a zero Runner that runs
|
||||
// /proc/self/exe pdftext.
|
||||
func (s *Server) runner() *pdftext.Runner {
|
||||
if s.PDFRunner != nil {
|
||||
return s.PDFRunner
|
||||
}
|
||||
return &pdftext.Runner{}
|
||||
}
|
||||
|
||||
// pool returns the extraction pool, built once and only when none is configured.
|
||||
func (s *Server) pool() *pdfpool.Pool {
|
||||
if s.PDFPool == nil {
|
||||
s.pdfPoolOnce.Do(func() {
|
||||
s.PDFPool = pdfpool.New(2, 4, 20*time.Second)
|
||||
})
|
||||
}
|
||||
return s.PDFPool
|
||||
}
|
||||
|
||||
// extractPDF runs one extraction under the pool, coalesced by document hash and
|
||||
// bounded to one per caller.
|
||||
func (s *Server) extractPDF(ctx context.Context, c *auth.Caller, res *fetch.Resource) (*pdftext.Doc, error) {
|
||||
fn := func(jobCtx context.Context) (any, error) {
|
||||
return s.runner().Run(jobCtx, res.Body)
|
||||
}
|
||||
v, err := s.pool().Do(ctx, s.callerClass(c)+"|"+res.SHA256(), c.Name, fn)
|
||||
if err != nil {
|
||||
return nil, pdfError(err)
|
||||
}
|
||||
doc, ok := v.(*pdftext.Doc)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("pdf: bad result %T", v)
|
||||
}
|
||||
return doc, nil
|
||||
}
|
||||
|
||||
// pdfError maps every PDF-path error to the status sift returns to its caller;
|
||||
// a PDF never produces a 500.
|
||||
func pdfError(err error) error {
|
||||
switch {
|
||||
case errors.Is(err, pdftext.ErrNoText):
|
||||
return &fetch.Error{Status: 422, Msg: "no text layer (scanned PDF?)", Hint: "ocr"}
|
||||
case errors.Is(err, pdftext.ErrEncrypted):
|
||||
return &fetch.Error{Status: 422, Msg: "encrypted PDF (password required)"}
|
||||
case errors.Is(err, pdftext.ErrTimeout):
|
||||
return &fetch.Error{Status: 422, Msg: "pdf extraction timed out"}
|
||||
case errors.Is(err, pdftext.ErrTooMuchText):
|
||||
return &fetch.Error{Status: 422, Msg: "pdf text over the output limit"}
|
||||
case errors.Is(err, pdfpool.ErrBusy):
|
||||
return &fetch.Error{Status: 503, Msg: "pdf workers busy", RetryAfter: 10}
|
||||
case errors.Is(err, pdfpool.ErrCallerBusy):
|
||||
return &fetch.Error{Status: 429, Msg: "busy"}
|
||||
case errors.Is(err, pdftext.ErrUnreadable):
|
||||
return &fetch.Error{Status: 422, Msg: "unreadable PDF"}
|
||||
default:
|
||||
return &fetch.Error{Status: 422, Msg: "unreadable PDF"}
|
||||
}
|
||||
}
|
||||
|
||||
// renderPDF draws the document for an agent: the whole text when it fits the
|
||||
// output cap, otherwise the map.
|
||||
func (s *Server) renderPDF(doc *pdftext.Doc) (pdfdoc.Rendered, error) {
|
||||
return pdfdoc.Render(doc, pdfdoc.Options{OutCap: s.OutCap})
|
||||
}
|
||||
@@ -0,0 +1,402 @@
|
||||
package api
|
||||
|
||||
// P5a task 06: PDFs through /v1/fetch and /v2/scrape — extraction in the child, the pool, the
|
||||
// document cache, the map, pages, grep, and the errors.
|
||||
//
|
||||
// The child is the test binary itself (TestMain below), as in internal/pdftext's tests.
|
||||
// SIFT_PDFTEXT_RUNLOG makes every child start append a line to a file, so a test can count
|
||||
// extractions; SIFT_PDFTEXT_SLEEP slows a child down. Both hooks live only in this file.
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/netip"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.wntrmute.dev/kyle/sift/internal/auth"
|
||||
"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/guard"
|
||||
"git.wntrmute.dev/kyle/sift/internal/pdfpool"
|
||||
"git.wntrmute.dev/kyle/sift/internal/pdftest"
|
||||
"git.wntrmute.dev/kyle/sift/internal/pdftext"
|
||||
)
|
||||
|
||||
func TestMain(m *testing.M) {
|
||||
if os.Getenv("SIFT_PDFTEXT_CHILD") == "1" {
|
||||
if p := os.Getenv("SIFT_PDFTEXT_RUNLOG"); p != "" {
|
||||
f, err := os.OpenFile(p, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o600)
|
||||
if err == nil {
|
||||
_, _ = f.WriteString("run\n")
|
||||
_ = f.Close()
|
||||
}
|
||||
}
|
||||
if d, err := time.ParseDuration(os.Getenv("SIFT_PDFTEXT_SLEEP")); err == nil {
|
||||
time.Sleep(d)
|
||||
}
|
||||
os.Exit(pdftext.ChildMain(os.Args[1:], os.Stdin, os.Stdout))
|
||||
}
|
||||
os.Exit(m.Run())
|
||||
}
|
||||
|
||||
func threePDF() []byte {
|
||||
return pdftest.Build(pdftest.Doc{
|
||||
Title: "Three Pages",
|
||||
Pages: []string{"Chapter one begins here.", "The quick brown fox is on page two.", "Last page."},
|
||||
Outline: []pdftest.Outline{{Title: "1 Intro", Page: 1, Level: 1}, {Title: "2 Fox", Page: 2, Level: 1}},
|
||||
})
|
||||
}
|
||||
|
||||
func longPDF(n int) []byte {
|
||||
var pages []string
|
||||
for p := 1; p <= n; p++ {
|
||||
pages = append(pages, pdftest.Lorem(p, 40))
|
||||
}
|
||||
return pdftest.Build(pdftest.Doc{Title: "Long Thesis", Pages: pages, Outline: []pdftest.Outline{{Title: "1 Introduction", Page: 1, Level: 1}, {Title: "2 Method", Page: n / 2, Level: 1}}})
|
||||
}
|
||||
|
||||
// origin serves one PDF and counts GETs. It honours If-None-Match against its current ETag.
|
||||
type origin struct {
|
||||
mu sync.Mutex
|
||||
ct string
|
||||
body []byte
|
||||
etag string
|
||||
gets atomic.Int32
|
||||
srv *httptest.Server
|
||||
}
|
||||
|
||||
func newOrigin(t *testing.T, ct string, body []byte) *origin {
|
||||
o := &origin{ct: ct, body: body, etag: `"v1"`}
|
||||
o.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
o.gets.Add(1)
|
||||
o.mu.Lock()
|
||||
defer o.mu.Unlock()
|
||||
if r.Header.Get("If-None-Match") == o.etag {
|
||||
w.WriteHeader(http.StatusNotModified)
|
||||
return
|
||||
}
|
||||
w.Header().Set("ETag", o.etag)
|
||||
if o.ct != "" {
|
||||
w.Header().Set("Content-Type", o.ct)
|
||||
}
|
||||
_, _ = w.Write(o.body)
|
||||
}))
|
||||
t.Cleanup(o.srv.Close)
|
||||
return o
|
||||
}
|
||||
|
||||
func (o *origin) replace(body []byte, etag string) {
|
||||
o.mu.Lock()
|
||||
o.body, o.etag = body, etag
|
||||
o.mu.Unlock()
|
||||
}
|
||||
|
||||
type pdfRig struct {
|
||||
s *Server
|
||||
fleet string // bearer key of caller "test" (hermes-fleet)
|
||||
other string // bearer key of caller "other" (hermes-fleet)
|
||||
gs string // bearer key of restricted caller "straylight-gs"
|
||||
runlog string
|
||||
logs *recHandler
|
||||
}
|
||||
|
||||
func newPDFRig(t *testing.T, childEnv ...string) *pdfRig {
|
||||
t.Helper()
|
||||
dir := t.TempDir()
|
||||
keys := filepath.Join(dir, "keys.toml")
|
||||
key := func(name, ws string) string {
|
||||
out := filepath.Join(dir, name)
|
||||
if err := auth.Add(keys, name, ws, out); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b, err := os.ReadFile(out)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return strings.TrimSpace(string(b))
|
||||
}
|
||||
r := &pdfRig{fleet: key("test", "hermes-fleet"), other: key("other", "hermes-fleet"), gs: key("straylight-gs", "guardian-star"), runlog: filepath.Join(dir, "runs"), logs: &recHandler{}}
|
||||
store, err := auth.Load(keys)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
g := &guard.Guard{AllowPrefixes: []netip.Prefix{netip.MustParsePrefix("127.0.0.0/8")}}
|
||||
s := New("test", g, &fetch.Fetcher{Guard: g, UA: "sift-test"}, store, slog.New(r.logs))
|
||||
s.PermitRestricted = []string{"straylight-gs"}
|
||||
db, err := dcache.Open(filepath.Join(dir, "cache.db"), 0, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = db.Close() })
|
||||
s.Digest = &digest.Engine{Cache: db} // the cache lives on the digest engine; no model is configured
|
||||
s.PDFRunner = &pdftext.Runner{Exe: os.Args[0], Env: append([]string{"SIFT_PDFTEXT_CHILD=1", "SIFT_PDFTEXT_RUNLOG=" + r.runlog}, childEnv...), Deadline: 20 * time.Second}
|
||||
r.s = s
|
||||
return r
|
||||
}
|
||||
|
||||
func (r *pdfRig) runs() int {
|
||||
b, _ := os.ReadFile(r.runlog)
|
||||
return strings.Count(string(b), "run\n")
|
||||
}
|
||||
|
||||
func (r *pdfRig) fetch(t *testing.T, key string, body map[string]any) (int, map[string]any, http.Header) {
|
||||
t.Helper()
|
||||
rec := postJSON(t, r.s, "/v1/fetch", key, body)
|
||||
var out map[string]any
|
||||
_ = json.Unmarshal(rec.Body.Bytes(), &out)
|
||||
return rec.Code, out, rec.Header()
|
||||
}
|
||||
|
||||
func pdfInfo(t *testing.T, out map[string]any) map[string]any {
|
||||
t.Helper()
|
||||
p, ok := out["pdf"].(map[string]any)
|
||||
if !ok {
|
||||
t.Fatalf("envelope has no pdf object: %v", out)
|
||||
}
|
||||
return p
|
||||
}
|
||||
|
||||
func TestPDFCleanFetch(t *testing.T) {
|
||||
for _, ct := range []string{"application/pdf", "application/octet-stream", ""} {
|
||||
r := newPDFRig(t)
|
||||
o := newOrigin(t, ct, threePDF())
|
||||
code, out, _ := r.fetch(t, r.fleet, map[string]any{"url": o.srv.URL})
|
||||
if code != 200 {
|
||||
t.Fatalf("%q: %d %v", ct, code, out)
|
||||
}
|
||||
content, _ := out["content"].(string)
|
||||
if !strings.Contains(content, "--- page 2 ---") || !strings.Contains(content, "quick brown fox") || out["mode"] != "clean" {
|
||||
t.Errorf("%q: content %q mode %v", ct, content, out["mode"])
|
||||
}
|
||||
p := pdfInfo(t, out)
|
||||
if p["pages"] != 3.0 || p["title"] != "Three Pages" || p["map"] != false || p["first"] != 1.0 || p["last"] != 3.0 {
|
||||
t.Errorf("%q: pdf = %v", ct, p)
|
||||
}
|
||||
if ol, _ := p["outline"].([]any); len(ol) != 2 {
|
||||
t.Errorf("%q: outline = %v", ct, p["outline"])
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestPDFMap(t *testing.T) {
|
||||
r := newPDFRig(t)
|
||||
r.s.OutCap = 20000
|
||||
o := newOrigin(t, "application/pdf", longPDF(40)) // ~40 × 2,700 characters
|
||||
code, out, _ := r.fetch(t, r.fleet, map[string]any{"url": o.srv.URL})
|
||||
if code != 200 {
|
||||
t.Fatalf("%d %v", code, out)
|
||||
}
|
||||
content, _ := out["content"].(string)
|
||||
p := pdfInfo(t, out)
|
||||
if p["map"] != true || !strings.Contains(content, "## Outline") || !strings.Contains(content, "- 2 Method (p. 20)") || len([]rune(content)) > 20000 {
|
||||
t.Errorf("map: pdf %v, %d chars:\n%.400s", p, len([]rune(content)), content)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPDFPagesAndGrep(t *testing.T) {
|
||||
r := newPDFRig(t)
|
||||
o := newOrigin(t, "application/pdf", threePDF())
|
||||
code, out, _ := r.fetch(t, r.fleet, map[string]any{"url": o.srv.URL, "pages": "2-3"})
|
||||
content, _ := out["content"].(string)
|
||||
if code != 200 || strings.Contains(content, "--- page 1 ---") || !strings.Contains(content, "--- page 3 ---") {
|
||||
t.Errorf("pages 2-3: %d %q", code, content)
|
||||
}
|
||||
if p := pdfInfo(t, out); p["first"] != 2.0 || p["last"] != 3.0 {
|
||||
t.Errorf("pages 2-3: pdf %v", p)
|
||||
}
|
||||
for _, bad := range []map[string]any{{"pages": "0-1"}, {"pages": "9"}, {"grep": strings.Repeat("a", 201)}, {"grep": "(unclosed"}} {
|
||||
bad["url"] = o.srv.URL
|
||||
if code, out, _ := r.fetch(t, r.fleet, bad); code != 400 || out["success"] != false {
|
||||
t.Errorf("%v: %d %v, want 400", bad, code, out)
|
||||
}
|
||||
}
|
||||
code, out, _ = r.fetch(t, r.fleet, map[string]any{"url": o.srv.URL, "grep": "quick brown"})
|
||||
content, _ = out["content"].(string)
|
||||
if code != 200 || !strings.Contains(content, "p. 2: The quick brown fox is on page two.") || pdfInfo(t, out)["grep_hits"] != 1.0 {
|
||||
t.Errorf("grep: %d %q %v", code, content, out["pdf"])
|
||||
}
|
||||
// pages and grep on a non-PDF are refused rather than ignored.
|
||||
html := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.Header().Set("Content-Type", "text/html")
|
||||
_, _ = io.WriteString(w, "<html><body><p>hi</p></body></html>")
|
||||
}))
|
||||
defer html.Close()
|
||||
if code, out, _ := r.fetch(t, r.fleet, map[string]any{"url": html.URL, "pages": "1"}); code != 400 {
|
||||
t.Errorf("pages on html: %d %v, want 400", code, out)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPDFCache(t *testing.T) {
|
||||
r := newPDFRig(t)
|
||||
o := newOrigin(t, "application/pdf", threePDF())
|
||||
get := func(extra map[string]any) string {
|
||||
body := map[string]any{"url": o.srv.URL}
|
||||
for k, v := range extra {
|
||||
body[k] = v
|
||||
}
|
||||
code, out, _ := r.fetch(t, r.fleet, body)
|
||||
if code != 200 {
|
||||
t.Fatalf("%v: %d %v", extra, code, out)
|
||||
}
|
||||
c, _ := out["content"].(string)
|
||||
return c
|
||||
}
|
||||
get(nil)
|
||||
if o.gets.Load() != 1 || r.runs() != 1 {
|
||||
t.Fatalf("first fetch: %d GETs, %d extractions; want 1 and 1", o.gets.Load(), r.runs())
|
||||
}
|
||||
if c := get(map[string]any{"pages": "2"}); !strings.Contains(c, "quick brown fox") {
|
||||
t.Errorf("cached pages=2: %q", c)
|
||||
}
|
||||
if o.gets.Load() != 1 || r.runs() != 1 {
|
||||
t.Errorf("fresh cache hit: %d GETs, %d extractions; want still 1 and 1", o.gets.Load(), r.runs())
|
||||
}
|
||||
// Stale: a conditional GET, answered 304, re-uses the text.
|
||||
r.s.PDFTTL = time.Nanosecond
|
||||
time.Sleep(2 * time.Millisecond)
|
||||
get(map[string]any{"grep": "Chapter"})
|
||||
if o.gets.Load() != 2 || r.runs() != 1 {
|
||||
t.Errorf("after 304: %d GETs, %d extractions; want 2 and 1", o.gets.Load(), r.runs())
|
||||
}
|
||||
// The document changed: a full GET and a new extraction.
|
||||
o.replace(pdftest.Build(pdftest.Doc{Pages: []string{"Brand new text."}}), `"v2"`)
|
||||
if c := get(nil); !strings.Contains(c, "Brand new text.") {
|
||||
t.Errorf("changed document: %q", c)
|
||||
}
|
||||
if o.gets.Load() != 3 || r.runs() != 2 {
|
||||
t.Errorf("after change: %d GETs, %d extractions; want 3 and 2", o.gets.Load(), r.runs())
|
||||
}
|
||||
if DefaultPDFTTL != 24*time.Hour {
|
||||
t.Errorf("DefaultPDFTTL = %v, want 24h", DefaultPDFTTL)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPDFErrors(t *testing.T) {
|
||||
r := newPDFRig(t)
|
||||
enc, err := os.ReadFile(filepath.Join("..", "..", "tests", "testdata", "pdf", "encrypted-user.pdf"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
body []byte
|
||||
want string
|
||||
hint string
|
||||
}{
|
||||
{"scanned", pdftest.Build(pdftest.Doc{Pages: []string{"", ""}, Scanned: []int{1, 2}}), "no text layer", "ocr"},
|
||||
{"encrypted", enc, "encrypted", ""},
|
||||
{"garbage", []byte("%PDF-1.4\nthis is not really a pdf"), "unreadable", ""},
|
||||
} {
|
||||
o := newOrigin(t, "application/pdf", tc.body)
|
||||
code, out, _ := r.fetch(t, r.fleet, map[string]any{"url": o.srv.URL})
|
||||
msg, _ := out["error"].(string)
|
||||
if code != 422 || !strings.Contains(msg, tc.want) || (tc.hint != "" && out["hint"] != tc.hint) {
|
||||
t.Errorf("%s: %d %v; want 422 %q hint %q", tc.name, code, out, tc.want, tc.hint)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestPDFTimeout(t *testing.T) {
|
||||
r := newPDFRig(t, "SIFT_PDFTEXT_SLEEP=10s")
|
||||
r.s.PDFRunner.Deadline = 300 * time.Millisecond
|
||||
o := newOrigin(t, "application/pdf", threePDF())
|
||||
code, out, _ := r.fetch(t, r.fleet, map[string]any{"url": o.srv.URL})
|
||||
if msg, _ := out["error"].(string); code != 422 || !strings.Contains(msg, "timed out") {
|
||||
t.Errorf("slow child: %d %v; want 422 … timed out", code, out)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPDFBusy(t *testing.T) {
|
||||
r := newPDFRig(t, "SIFT_PDFTEXT_SLEEP=1500ms")
|
||||
r.s.PDFPool = pdfpool.New(1, 0, 100*time.Millisecond)
|
||||
a := newOrigin(t, "application/pdf", threePDF())
|
||||
b := newOrigin(t, "application/pdf", longPDF(2))
|
||||
done := make(chan int, 1)
|
||||
go func() {
|
||||
code, _, _ := r.fetch(t, r.fleet, map[string]any{"url": a.srv.URL})
|
||||
done <- code
|
||||
}()
|
||||
time.Sleep(400 * time.Millisecond) // the first extraction holds the only slot
|
||||
code, out, hdr := r.fetch(t, r.other, map[string]any{"url": b.srv.URL})
|
||||
if code != 503 || hdr.Get("Retry-After") != "10" || !strings.Contains(fmt.Sprint(out["error"]), "busy") {
|
||||
t.Errorf("second caller with the pool full: %d %v Retry-After %q; want 503, busy, 10", code, out, hdr.Get("Retry-After"))
|
||||
}
|
||||
code, out, _ = r.fetch(t, r.fleet, map[string]any{"url": b.srv.URL})
|
||||
if code != 429 {
|
||||
t.Errorf("same caller, second document while its first is running: %d %v; want 429", code, out)
|
||||
}
|
||||
if c := <-done; c != 200 {
|
||||
t.Errorf("first fetch: %d", c)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPDFRestrictedCaller(t *testing.T) {
|
||||
r := newPDFRig(t)
|
||||
o := newOrigin(t, "application/pdf", threePDF())
|
||||
if code, out, _ := r.fetch(t, r.gs, map[string]any{"url": o.srv.URL}); code != 200 {
|
||||
t.Fatalf("restricted fetch: %d %v", code, out)
|
||||
}
|
||||
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 })
|
||||
line := b.String()
|
||||
if strings.Contains(line, "127.0.0.1") || strings.Contains(line, "Three Pages") || strings.Contains(line, "quick brown") {
|
||||
t.Errorf("a restricted caller's PDF fetch logged %q", line)
|
||||
}
|
||||
}
|
||||
r.logs.mu.Unlock()
|
||||
// The fleet does not see the restricted caller's cached text: it extracts its own copy.
|
||||
if code, _, _ := r.fetch(t, r.fleet, map[string]any{"url": o.srv.URL}); code != 200 || r.runs() != 2 {
|
||||
t.Errorf("fleet after restricted: %d, %d extractions; want 200 and 2 (separate cache partitions)", code, r.runs())
|
||||
}
|
||||
}
|
||||
|
||||
func TestPDFScrape(t *testing.T) {
|
||||
r := newPDFRig(t)
|
||||
o := newOrigin(t, "application/pdf", threePDF())
|
||||
rec := postJSON(t, r.s, "/v2/scrape", r.fleet, map[string]any{"url": o.srv.URL, "formats": []string{"markdown"}})
|
||||
var out struct {
|
||||
Success bool `json:"success"`
|
||||
Data struct {
|
||||
Markdown string `json:"markdown"`
|
||||
Metadata map[string]any `json:"metadata"`
|
||||
} `json:"data"`
|
||||
}
|
||||
if err := json.Unmarshal(rec.Body.Bytes(), &out); err != nil || rec.Code != 200 || !out.Success {
|
||||
t.Fatalf("%d %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
if !strings.Contains(out.Data.Markdown, "--- page 2 ---") || out.Data.Metadata["numPages"] != 3.0 || out.Data.Metadata["title"] != "Three Pages" {
|
||||
t.Errorf("scrape: markdown %q metadata %v", out.Data.Markdown, out.Data.Metadata)
|
||||
}
|
||||
if ct, _ := out.Data.Metadata["contentType"].(string); !strings.HasPrefix(ct, "application/pdf") {
|
||||
t.Errorf("contentType = %q", ct)
|
||||
}
|
||||
}
|
||||
|
||||
// A question on a document longer than PDFMaxQuestionChunks is refused with a pointer to grep
|
||||
// and pages, before any model call.
|
||||
func TestPDFQuestionTooLong(t *testing.T) {
|
||||
r := newPDFRig(t)
|
||||
r.s.PDFMaxQuestionChunks = 2
|
||||
o := newOrigin(t, "application/pdf", longPDF(12))
|
||||
code, out, _ := r.fetch(t, r.fleet, map[string]any{"url": o.srv.URL, "question": "what is the method?"})
|
||||
if msg, _ := out["error"].(string); code != 422 || !strings.Contains(msg, "too long") || !strings.Contains(msg, "grep") {
|
||||
t.Errorf("long question: %d %v; want 422 … too long … grep", code, out)
|
||||
}
|
||||
if DefaultPDFMaxQuestionChunks != 200 {
|
||||
t.Errorf("DefaultPDFMaxQuestionChunks = %d, want 200", DefaultPDFMaxQuestionChunks)
|
||||
}
|
||||
}
|
||||
@@ -43,9 +43,10 @@ func (r *Resource) SHA256() string {
|
||||
|
||||
// Error is a fetch failure with the HTTP status sift should return to its caller.
|
||||
type Error struct {
|
||||
Status int
|
||||
Msg string
|
||||
Hint string
|
||||
Status int
|
||||
Msg string
|
||||
Hint string
|
||||
RetryAfter int // seconds to wait before retrying; 0 = none
|
||||
}
|
||||
|
||||
func (e *Error) Error() string { return e.Msg }
|
||||
|
||||
Reference in New Issue
Block a user