api: v1fetch's per-target pipeline becomes fetchTarget (no behaviour change)

Move v1fetch's inline per-target pipeline into a new internal/api/target.go
method, `fetchTarget(ctx, c, t batch.Target) (batch.Result, error)`, so the
batch worker can drive the same work. The body moves unchanged in order:
question truncation, mode default/check, wantCtx, doFetch, the recover path,
the PDF branch (render, pages/grep errors, question-chunk limit), the
clean/raw branch, digest, context search, the fetch log line and the
envelope. `content` is returned as Result.Content; every other envelope key
goes in Result.Meta, never content. v1fetch keeps reqN/caller/decode/400,
builds a batch.Target from the request, calls fetchTarget, writeFetchErr on
error and writeJSON(w, 200, env) on success where env is a copy of Meta plus
"content". doFetch takes context.Context instead of *http.Request; scrape
passes r.Context(). No new behaviour, fields or JSON keys.

Co-Authored-By: ornith-1.5-35b-a3b <ornith-1.5-35b-a3b@llama.cpp.invalid>
This commit is contained in:
2026-09-26 22:34:56 -07:00
co-authored by ornith-1.5-35b-a3b
parent e0168f7d65
commit 0d923bc366
4 changed files with 352 additions and 220 deletions
+1
View File
@@ -21,6 +21,7 @@ One row per task session that ends in a commit. Newest last.
| p6/02-runner-core | 2026-09-26 | done | 1 | gate: ok | caller-cleanup hand-off to task 03 | Runner core in new internal/batch with no workers (task 03 adds them; jobs stay queued, so a queued job counts as live for the caps). Public surface in batch.go: Options/Target/Result/TargetError, FetchFunc, Job/TargetStatus, Hooks and Config with defaults (TTL 24h, 32 targets/job, 1 job/caller, 8 total, worker clamped to 2, 64 MiB, 2 s/10 m retry). Submit runs the given order — empty→ErrNoTargets, idempotency via hex SHA-256 of `name+NUL+json.Marshal(targets)` (a live hit returns the id reused, touching no caps), then TargetsPerJob/JobsPerCaller/JobsTotal caps, then stores the job queued (random 16-byte hex id, CreatedAt=Lease=now) and its targets (pending, Options=json of t.Options) before returning so a queued job is live; the submitting caller is held in memory by job id, never persisted. Status maps rows and decodes Result without writing; Touch moves the lease; Cancel sets a queued job cancelled at once and leaves its targets pending; Close is idempotent. Store errors are Warn-logged with the job id, never a URL/host. | | p6/02-runner-core | 2026-09-26 | done | 1 | gate: ok | caller-cleanup hand-off to task 03 | Runner core in new internal/batch with no workers (task 03 adds them; jobs stay queued, so a queued job counts as live for the caps). Public surface in batch.go: Options/Target/Result/TargetError, FetchFunc, Job/TargetStatus, Hooks and Config with defaults (TTL 24h, 32 targets/job, 1 job/caller, 8 total, worker clamped to 2, 64 MiB, 2 s/10 m retry). Submit runs the given order — empty→ErrNoTargets, idempotency via hex SHA-256 of `name+NUL+json.Marshal(targets)` (a live hit returns the id reused, touching no caps), then TargetsPerJob/JobsPerCaller/JobsTotal caps, then stores the job queued (random 16-byte hex id, CreatedAt=Lease=now) and its targets (pending, Options=json of t.Options) before returning so a queued job is live; the submitting caller is held in memory by job id, never persisted. Status maps rows and decodes Result without writing; Touch moves the lease; Cancel sets a queued job cancelled at once and leaves its targets pending; Close is idempotent. Store errors are Warn-logged with the job id, never a URL/host. |
| p6/03-runner-workers | 2026-09-26 | done | 1 | gate: ok | none | Workers in new internal/batch/worker.go: one rotation of live jobs in submit order, round-robin via an absolute cursor (advances in pick, wraps at access, so it survives a second job being added while the first is in flight — the mod-n cursor from the design stuck at 0 for one job and re-picked it), serial per job (target i+1 never starts before i is stored), one target per fetch under the Close-cancelled context (panic→ErrInternal, never echoed; ErrRetryLater retried after RetryWait, bounded by RetryFor then stored failed with the wrapped text). Terminal status once per job: all done→done, all failed→failed, mix→partial. Cancel of a running job sets cancelled, stores the in-flight target without overwriting cancelled, leaves later targets pending. Close sets closed, cancels the context, broadcasts, waits the workers, idempotent. batch.go: New starts cfg.Workers, Submit appends the liveJob and wakes, Cancel handles the in-flight case, Close stops. Absolute cursor is the one deviation from the design's "index in live" comment. All race tests pass -count=3. | | | p6/03-runner-workers | 2026-09-26 | done | 1 | gate: ok | none | Workers in new internal/batch/worker.go: one rotation of live jobs in submit order, round-robin via an absolute cursor (advances in pick, wraps at access, so it survives a second job being added while the first is in flight — the mod-n cursor from the design stuck at 0 for one job and re-picked it), serial per job (target i+1 never starts before i is stored), one target per fetch under the Close-cancelled context (panic→ErrInternal, never echoed; ErrRetryLater retried after RetryWait, bounded by RetryFor then stored failed with the wrapped text). Terminal status once per job: all done→done, all failed→failed, mix→partial. Cancel of a running job sets cancelled, stores the in-flight target without overwriting cancelled, leaves later targets pending. Close sets closed, cancels the context, broadcasts, waits the workers, idempotent. batch.go: New starts cfg.Workers, Submit appends the liveJob and wakes, Cancel handles the in-flight case, Close stops. Absolute cursor is the one deviation from the design's "index in live" comment. All race tests pass -count=3. | |
| p6/04-metrics | 2026-09-26 | done | 1 | gate: ok | none | Batch series in new internal/metrics/batch.go, registered in New: `BatchJobsTotal`/`BatchTargetsTotal` CounterVecs (label "outcome", any value outside the known set folded to "other" via `batchJobOutcomes`/`batchTargetOutcomes` maps so a raw string never becomes a label), `BatchQueueDepth`/`BatchTargetsInflight` gauges, and `BatchJobSeconds`/`BatchTargetSeconds` histograms (durationBuckets) plus `BatchResultBytes` (1K,4K,16K,64K,256K,1M). Nil-receiver-safe `Observe…`/`Set…` methods mirroring ObservePDF/ObserveDigest. Series served through the existing /metrics handler; no new endpoint, no Snapshot change. Given test byte-identical; metrics tests pass under -race -count=3; make gate ok. | | p6/04-metrics | 2026-09-26 | done | 1 | gate: ok | none | Batch series in new internal/metrics/batch.go, registered in New: `BatchJobsTotal`/`BatchTargetsTotal` CounterVecs (label "outcome", any value outside the known set folded to "other" via `batchJobOutcomes`/`batchTargetOutcomes` maps so a raw string never becomes a label), `BatchQueueDepth`/`BatchTargetsInflight` gauges, and `BatchJobSeconds`/`BatchTargetSeconds` histograms (durationBuckets) plus `BatchResultBytes` (1K,4K,16K,64K,256K,1M). Nil-receiver-safe `Observe…`/`Set…` methods mirroring ObservePDF/ObserveDigest. Series served through the existing /metrics handler; no new endpoint, no Snapshot change. Given test byte-identical; metrics tests pass under -race -count=3; make gate ok. |
| p6/05-fetchtarget | 2026-09-26 | done | 1 | gate: ok | none | Extracted v1fetch's per-target pipeline into new `internal/api/target.go`: `fetchTarget(ctx, c, t batch.Target) (batch.Result, error)`. Body moved unchanged in order — question truncation to 500 runes, mode default/check, wantCtx, doFetch, recover path, PDF branch (render, pages/grep errors, question-chunk limit), clean/raw branch, digest, context search, the fetch log line, and the envelope. `content` is returned as `Result.Content`; every other envelope key (`mode`,`untrusted`,`source`,`warnings`,`sizes`,`timings`, and `pdf`/`digest`/`context` when present) goes in `Result.Meta`, never content. `v1fetch` keeps reqN/caller/decode/400, builds a `batch.Target` (`URL` + `Mode`/`Question`/`Pages`/`Grep`/`JQ`/`Context`/`Recover` into `Options`), calls `fetchTarget`, `writeFetchErr` on error and `writeJSON(w,200,env)` on success where `env` is a fresh copy of `Meta` plus `"content"` (a copy, so `len(env)==len(Meta)+1` holds and Meta is not mutated). The two `writeErr` failures become `*fetch.Error` with the same status/msg/hint: 400 "invalid url" (bad mode), 400 "raw mode does not apply to PDFs". `doFetch` takes `context.Context` instead of `*http.Request` (its three `r.Context()` → `ctx`); `scrape` passes `r.Context()`. Removed now-unused `fmt`/`unicode/utf8`/`recover` imports, added `batch`. All api tests pass under -race -count=3; make gate ok; given test byte-identical. |
## P5a review (owner, 2026-09-26) ## P5a review (owner, 2026-09-26)
+25 -220
View File
@@ -6,7 +6,6 @@ import (
"encoding/hex" "encoding/hex"
"encoding/json" "encoding/json"
"errors" "errors"
"fmt"
"io" "io"
"log/slog" "log/slog"
"net/http" "net/http"
@@ -16,9 +15,9 @@ import (
"sync" "sync"
"sync/atomic" "sync/atomic"
"time" "time"
"unicode/utf8"
"git.wntrmute.dev/kyle/sift/internal/auth" "git.wntrmute.dev/kyle/sift/internal/auth"
"git.wntrmute.dev/kyle/sift/internal/batch"
"git.wntrmute.dev/kyle/sift/internal/clean" "git.wntrmute.dev/kyle/sift/internal/clean"
"git.wntrmute.dev/kyle/sift/internal/dcache" "git.wntrmute.dev/kyle/sift/internal/dcache"
"git.wntrmute.dev/kyle/sift/internal/digest" "git.wntrmute.dev/kyle/sift/internal/digest"
@@ -29,7 +28,6 @@ import (
"git.wntrmute.dev/kyle/sift/internal/pdfdoc" "git.wntrmute.dev/kyle/sift/internal/pdfdoc"
"git.wntrmute.dev/kyle/sift/internal/pdfpool" "git.wntrmute.dev/kyle/sift/internal/pdfpool"
"git.wntrmute.dev/kyle/sift/internal/pdftext" "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/search"
"git.wntrmute.dev/kyle/sift/internal/selfip" "git.wntrmute.dev/kyle/sift/internal/selfip"
) )
@@ -389,7 +387,7 @@ func (s *Server) scrape(w http.ResponseWriter, r *http.Request) {
} }
} }
s.scrapeN.Add(1) s.scrapeN.Add(1)
res, cl, doc, ferr := s.doFetch(r, c, req.URL, "", onlyMain) res, cl, doc, ferr := s.doFetch(r.Context(), c, req.URL, "", onlyMain)
if ferr != nil { if ferr != nil {
s.writeFetchErr(w, c, req.URL, ferr) s.writeFetchErr(w, c, req.URL, ferr)
return return
@@ -449,222 +447,29 @@ func (s *Server) v1fetch(w http.ResponseWriter, r *http.Request) {
writeErr(w, 400, "invalid url", "") writeErr(w, 400, "invalid url", "")
return return
} }
q := req.Question t := batch.Target{
if len([]rune(q)) > 500 { URL: req.URL,
rns := []rune(q) Options: batch.Options{
q = string(rns[:500]) Mode: req.Mode,
} Question: req.Question,
mode := req.Mode Pages: req.Pages,
if mode == "" { Grep: req.Grep,
if q != "" { JQ: req.JQ,
mode = "digest" Context: req.Context,
} else { Recover: req.Recover,
mode = "clean"
}
}
if mode != "clean" && mode != "raw" && mode != "digest" {
writeErr(w, 400, "invalid url", "")
return
}
wantCtx := q != ""
if req.Context != nil {
wantCtx = *req.Context
}
t0 := time.Now()
onlyMain := true
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, doc, ferr = s.cleanFetched(sr.Res, req.JQ, onlyMain)
snap = sr
}
}
if ferr != nil {
s.writeFetchErr(w, c, req.URL, ferr)
return
}
fetchMS := time.Since(t0).Milliseconds()
if doc == nil && (req.Pages != "" || req.Grep != "") {
s.writeFetchErr(w, c, req.URL, &fetch.Error{Status: 400, Msg: "pages and grep apply to PDFs only"})
return
}
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, pdfdoc.Options{Pages: req.Pages, Grep: req.Grep, OutCap: s.OutCap})
if err != nil {
if errors.Is(err, pdfdoc.ErrBadPages) || errors.Is(err, pdfdoc.ErrBadGrep) {
s.writeFetchErr(w, c, req.URL, &fetch.Error{Status: 400, Msg: err.Error()})
return
}
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{}
if cl.Warning != "" {
warn = append(warn, cl.Warning)
}
var digMeta map[string]any
var digestMS int64
if mode == "digest" {
if doc != nil {
n := len(digest.Split(pdfdoc.Full(doc), digest.ChunkTokens))
limit := s.PDFMaxQuestionChunks
if limit == 0 {
limit = DefaultPDFMaxQuestionChunks
}
if n > limit {
s.writeFetchErr(w, c, req.URL, &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, q, 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,
}
}
}
}
var ctxItems []fleetctx.Item
var ctxMS int64
if wantCtx && s.Ctx != nil {
ct0 := time.Now()
query := fleetctx.Query(q, cl.Title, headings(cl.Markdown))
items, cw := s.Ctx.Search(r.Context(), c.Workspace, c.Name, query)
ctxMS = time.Since(ct0).Milliseconds()
if cw != "" {
warn = append(warn, cw)
}
ctxItems = items
}
qLen := len(q)
if qLen > 80 {
qLen = 80
}
fb, _ := "", false
if digMeta != nil {
if s, ok := digMeta["fallback"].(string); ok {
fb = s
}
}
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,
"pdf_pages", pdfPages, "digest_ms", digestMS, "fallback", fb)
src := map[string]any{
"url": res.URL,
"final_url": res.FinalURL,
"status": res.Status,
"content_type": res.ContentType,
"fetched_at": res.FetchedAt.Format(time.RFC3339),
"hash": res.SHA256(),
"bytes": len(res.Body),
"cached": len(res.Body) == 0,
"kind": "live",
}
if snap != nil {
src["kind"] = "snapshot"
src["via"] = string(snap.Via)
src["original_url"] = snap.OriginalURL
if snap.SnapshotAt != "" {
src["snapshot_at"] = snap.SnapshotAt
} else {
src["snapshot_at"] = nil
warn = append(warn, "snapshot date unknown")
}
}
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(cl.Markdown),
"returned_tokens": digest.EstTokens(content),
}, },
"timings": map[string]int64{"fetch_ms": fetchMS, "digest_ms": digestMS, "context_ms": ctxMS},
} }
if pdfMap != nil { res, err := s.fetchTarget(r.Context(), c, t)
out["pdf"] = pdfMap if err != nil {
s.writeFetchErr(w, c, req.URL, err)
return
} }
if digMeta != nil { env := make(map[string]any, len(res.Meta)+1)
out["digest"] = digMeta for k, v := range res.Meta {
env[k] = v
} }
if wantCtx { env["content"] = res.Content
out["context"] = ctxItems writeJSON(w, 200, env)
}
writeJSON(w, 200, out)
} }
func recoverWorthy(err error) bool { func recoverWorthy(err error) bool {
@@ -710,8 +515,8 @@ func (s *Server) cleanFetched(res *fetch.Resource, jq string, onlyMain bool) (*f
return res, cl, nil, 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, *pdftext.Doc, error) { func (s *Server) doFetch(ctx context.Context, c *auth.Caller, rawURL, jq string, onlyMain bool) (*fetch.Resource, *clean.Result, *pdftext.Doc, error) {
res, hitDoc, ok, err := s.pdfFromCache(r.Context(), c, rawURL) res, hitDoc, ok, err := s.pdfFromCache(ctx, c, rawURL)
if err != nil { if err != nil {
return nil, nil, nil, err return nil, nil, nil, err
} }
@@ -724,7 +529,7 @@ func (s *Server) doFetch(r *http.Request, c *auth.Caller, rawURL, jq string, onl
} }
defer s.release(c.Name, rawURL) defer s.release(c.Name, rawURL)
res, err = s.Fetch.Fetch(r.Context(), rawURL) res, err = s.Fetch.Fetch(ctx, rawURL)
if err != nil { if err != nil {
return nil, nil, nil, err return nil, nil, nil, err
} }
@@ -756,7 +561,7 @@ func (s *Server) doFetch(r *http.Request, c *auth.Caller, rawURL, jq string, onl
} }
var doc *pdftext.Doc var doc *pdftext.Doc
if kind == "pdf" { if kind == "pdf" {
doc, err = s.extractPDF(r.Context(), c, res) doc, err = s.extractPDF(ctx, c, res)
if err != nil { if err != nil {
return nil, nil, nil, err return nil, nil, nil, err
} }
+94
View File
@@ -0,0 +1,94 @@
package api
// P6 task 05: v1fetch's per-target pipeline is extracted into s.fetchTarget, which the batch
// runner also calls. These tests pin that the two agree; every existing v1fetch test must also
// pass unchanged.
import (
"context"
"encoding/json"
"errors"
"net/http"
"net/http/httptest"
"net/netip"
"strings"
"testing"
"git.wntrmute.dev/kyle/sift/internal/auth"
"git.wntrmute.dev/kyle/sift/internal/batch"
"git.wntrmute.dev/kyle/sift/internal/fetch"
"git.wntrmute.dev/kyle/sift/internal/guard"
)
// newTargetServer is setupAPI with loopback allowed; its key belongs to caller "test".
func newTargetServer(t *testing.T) (*Server, string) {
t.Helper()
return setupAPI(t, &guard.Guard{AllowPrefixes: []netip.Prefix{netip.MustParsePrefix("127.0.0.0/8")}})
}
func TestFetchTargetMatchesV1Fetch(t *testing.T) {
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "text/html; charset=utf-8")
_, _ = w.Write([]byte(`<html><head><title>Same</title></head><body><article><h1>Same</h1><p>one pipeline, two callers</p></article></body></html>`))
}))
defer ts.Close()
s, key := newTargetServer(t)
c := &auth.Caller{Name: "test", Workspace: "hermes-fleet"}
res, err := s.fetchTarget(context.Background(), c, batch.Target{URL: ts.URL})
if err != nil {
t.Fatal(err)
}
if !strings.Contains(res.Content, "one pipeline, two callers") {
t.Fatalf("content %q", res.Content)
}
if _, has := res.Meta["content"]; has {
t.Fatal("Meta must not repeat the content")
}
for _, k := range []string{"mode", "untrusted", "source", "warnings", "sizes", "timings"} {
if _, has := res.Meta[k]; !has {
t.Errorf("Meta lacks %q (it carries every envelope field except content)", k)
}
}
if res.Meta["mode"] != "clean" || res.Meta["untrusted"] != true {
t.Fatalf("meta %v", res.Meta)
}
// The same URL through /v1/fetch: identical content and mode, and the envelope has exactly
// the fields of Meta plus content.
rec := postJSON(t, s, "/v1/fetch", key, map[string]any{"url": ts.URL})
if rec.Code != 200 {
t.Fatalf("v1fetch %d %s", rec.Code, rec.Body.String())
}
var env map[string]any
if err := json.Unmarshal(rec.Body.Bytes(), &env); err != nil {
t.Fatal(err)
}
if env["content"] != res.Content || env["mode"] != res.Meta["mode"] {
t.Fatalf("v1fetch and fetchTarget disagree:\n%v\n%+v", env, res)
}
if len(env) != len(res.Meta)+1 {
t.Fatalf("envelope keys %d, Meta keys %d: want Meta + content", len(env), len(res.Meta))
}
}
func TestFetchTargetErrors(t *testing.T) {
s, _ := newTargetServer(t)
c := &auth.Caller{Name: "test", Workspace: "hermes-fleet"}
cases := []struct {
name string
target batch.Target
status int
msg string
}{
{"blocked", batch.Target{URL: "http://10.0.0.5/"}, 403, "blocked"},
{"bad mode", batch.Target{URL: "http://127.0.0.1/", Options: batch.Options{Mode: "shout"}}, 400, ""},
}
for _, tc := range cases {
_, err := s.fetchTarget(context.Background(), c, tc.target)
var fe *fetch.Error
if !errors.As(err, &fe) || fe.Status != tc.status || !strings.Contains(fe.Msg, tc.msg) {
t.Errorf("%s: err %v, want a *fetch.Error %d containing %q", tc.name, err, tc.status, tc.msg)
}
}
}
+232
View File
@@ -0,0 +1,232 @@
package api
import (
"context"
"errors"
"fmt"
"time"
"unicode/utf8"
"git.wntrmute.dev/kyle/sift/internal/auth"
"git.wntrmute.dev/kyle/sift/internal/batch"
"git.wntrmute.dev/kyle/sift/internal/clean"
"git.wntrmute.dev/kyle/sift/internal/digest"
"git.wntrmute.dev/kyle/sift/internal/fetch"
"git.wntrmute.dev/kyle/sift/internal/fleetctx"
"git.wntrmute.dev/kyle/sift/internal/pdfdoc"
"git.wntrmute.dev/kyle/sift/internal/recover"
)
// fetchTarget runs one target's per-target pipeline: fetch → recover → PDF render or clean →
// question limit → digest → context → log line → envelope. It is the single body both /v1/fetch
// and the batch worker drive, so it must not change what v1fetch returned.
func (s *Server) fetchTarget(ctx context.Context, c *auth.Caller, t batch.Target) (batch.Result, error) {
q := t.Options.Question
if len([]rune(q)) > 500 {
rns := []rune(q)
q = string(rns[:500])
}
mode := t.Options.Mode
if mode == "" {
if q != "" {
mode = "digest"
} else {
mode = "clean"
}
}
if mode != "clean" && mode != "raw" && mode != "digest" {
return batch.Result{}, &fetch.Error{Status: 400, Msg: "invalid url"}
}
wantCtx := q != ""
if t.Options.Context != nil {
wantCtx = *t.Options.Context
}
t0 := time.Now()
onlyMain := true
res, cl, doc, ferr := s.doFetch(ctx, c, t.URL, t.Options.JQ, onlyMain)
var snap *recover.Result
if ferr != nil && t.Options.Recover && recoverWorthy(ferr) {
sr, rerr := recover.Try(ctx, s.Fetch, t.URL)
if rerr == nil && sr != nil && sr.Res != nil {
res, cl, doc, ferr = s.cleanFetched(sr.Res, t.Options.JQ, onlyMain)
snap = sr
}
}
if ferr != nil {
return batch.Result{}, ferr
}
fetchMS := time.Since(t0).Milliseconds()
if doc == nil && (t.Options.Pages != "" || t.Options.Grep != "") {
return batch.Result{}, &fetch.Error{Status: 400, Msg: "pages and grep apply to PDFs only"}
}
var docLen, pdfPages int
var content string
var pdfMap map[string]any
if doc != nil {
pdfPages = doc.Pages
if mode == "raw" {
return batch.Result{}, &fetch.Error{Status: 400, Msg: "raw mode does not apply to PDFs"}
}
rp, err := s.renderPDF(doc, pdfdoc.Options{Pages: t.Options.Pages, Grep: t.Options.Grep, OutCap: s.OutCap})
if err != nil {
if errors.Is(err, pdfdoc.ErrBadPages) || errors.Is(err, pdfdoc.ErrBadGrep) {
return batch.Result{}, &fetch.Error{Status: 400, Msg: err.Error()}
}
return batch.Result{}, err
}
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{}
if cl.Warning != "" {
warn = append(warn, cl.Warning)
}
var digMeta map[string]any
var digestMS int64
if mode == "digest" {
if doc != nil {
n := len(digest.Split(pdfdoc.Full(doc), digest.ChunkTokens))
limit := s.PDFMaxQuestionChunks
if limit == 0 {
limit = DefaultPDFMaxQuestionChunks
}
if n > limit {
return batch.Result{}, &fetch.Error{Status: 422, Msg: fmt.Sprintf("document too long for question (%d chunks); use grep or pages", n)}
}
}
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(ctx, content, q, 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,
}
}
}
}
var ctxItems []fleetctx.Item
var ctxMS int64
if wantCtx && s.Ctx != nil {
ct0 := time.Now()
query := fleetctx.Query(q, cl.Title, headings(cl.Markdown))
items, cw := s.Ctx.Search(ctx, c.Workspace, c.Name, query)
ctxMS = time.Since(ct0).Milliseconds()
if cw != "" {
warn = append(warn, cw)
}
ctxItems = items
}
qLen := len(q)
if qLen > 80 {
qLen = 80
}
fb, _ := "", false
if digMeta != nil {
if s, ok := digMeta["fallback"].(string); ok {
fb = s
}
}
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,
"pdf_pages", pdfPages, "digest_ms", digestMS, "fallback", fb)
src := map[string]any{
"url": res.URL,
"final_url": res.FinalURL,
"status": res.Status,
"content_type": res.ContentType,
"fetched_at": res.FetchedAt.Format(time.RFC3339),
"hash": res.SHA256(),
"bytes": len(res.Body),
"cached": len(res.Body) == 0,
"kind": "live",
}
if snap != nil {
src["kind"] = "snapshot"
src["via"] = string(snap.Via)
src["original_url"] = snap.OriginalURL
if snap.SnapshotAt != "" {
src["snapshot_at"] = snap.SnapshotAt
} else {
src["snapshot_at"] = nil
warn = append(warn, "snapshot date unknown")
}
}
m := map[string]any{
"mode": mode,
"untrusted": true,
"source": src,
"warnings": warn,
"sizes": map[string]int{
"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 {
m["pdf"] = pdfMap
}
if digMeta != nil {
m["digest"] = digMeta
}
if wantCtx {
m["context"] = ctxItems
}
return batch.Result{Content: content, Meta: m}, nil
}