Same
one pipeline, two callers
`))
+ }))
+ 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)
+ }
+ }
+}
diff --git a/internal/api/target.go b/internal/api/target.go
new file mode 100644
index 0000000..415895e
--- /dev/null
+++ b/internal/api/target.go
@@ -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
+}