Merge pull request 'Inject a pdbmux_source provenance fact' (#13) from benvin/source-fact into main

Reviewed-on: #13

README.md

@@ -24,8 +24,8 @@ not PQL) is forwarded verbatim.
 
 | Path | Behaviour |
 |---|---|
-| `GET /pdb/query/v4/nodes` | Fan out to all backends, dedupe by `certname`, keep the record with the newer `report_timestamp`. An `extract`/`count` query is **summed** instead. |
-| `GET /pdb/query/v4/facts` | Fan out to all, and per `certname` keep **all** facts from the backend that owns that node (see merge semantics). |
+| `GET /pdb/query/v4/nodes` | Fan out to all backends, dedupe by `certname`, keep the record with the newer `report_timestamp`, stamped with the winning backend's name (see provenance). An `extract`/`count` query is **summed** instead. |
+| `GET /pdb/query/v4/facts` | Fan out to all, and per `certname` keep **all** facts from the backend that owns that node (see merge semantics), plus a synthetic `pdbmux_source` fact naming it. |
 | `GET /pdb/query/v4/resources` | An `extract`/`count` query is fanned out and **summed**; any other query is an unmerged pass-through. |
 | `GET /pdb/query/v4/reports` | Fan out to all and serve the **union**, deduped by report `hash`, re-ordered and re-paged across backends. |
 | `GET /pdb/query/v4/events` | Fan out to all and serve the **union**, deduped by record identity, re-ordered and re-paged. |
@@ -91,6 +91,74 @@ unknown fields survive untouched.
   - `include_total=true` on a summed endpoint reports the **merged** row count,
     not the sum of the backends' `X-Records`, since shared keys collapse.
 
+### Provenance: the `pdbmux_source` fact
+
+Once several PuppetDBs sit behind one endpoint, a consumer can no longer tell
+which backend a node's data came from. `pdbmux` makes that visible in the
+response itself, so nothing has to query each backend to find out:
+
+- **`/facts`** gains one extra fact record per `certname`, alongside the node's
+  real facts, in the shape of a real fact record — `certname`, `name`, `value`,
+  `environment` — with `value` set to the **backend name** from `backends` /
+  `PDBMUX_BACKENDS`. `environment` is copied from that node's own facts (all
+  four keys are always present, since clients index them directly).
+- **`/nodes`** gains a `pdbmux_source` **key** on each merged node record. A
+  node record carries no facts, so this is a synthetic field, not a fact — the
+  one key outside PuppetDB's documented node schema. Clients read node fields by
+  name, so an extra key is ignored by anything that doesn't want it.
+
+The value always names the backend **whose data won that endpoint's merge**, not
+a backend that merely holds the node. The two endpoints resolve their winner
+separately, so under `merge: static` they can legitimately disagree: `/facts`
+attributes a shared node to the first backend in configured order, while
+`/nodes` always attributes it to the backend holding the newer
+`report_timestamp`. Each answer describes the record it is attached to.
+
+While the fact is enabled, any `/facts` record whose own `name` field equals the
+configured name is dropped, on every query shape — including the shapes below,
+where nothing is injected in its place. The rule reads the record, not the query,
+so a projection that filters on `name` without returning it — say
+`["extract",["certname","value"],["=","name","pdbmux_source"]]` — produces rows
+that no longer identify themselves, and an upstream value of that name comes
+through. Ask for the `name` column and the guarantee holds. Each request that
+drops a record logs it once. Only `source_fact_enabled: false` restores upstream
+records of that name; rename the synthetic fact via `source_fact` if the real one
+matters more.
+
+**Injection is skipped**, and no synthetic record is added, when:
+
+- the query contains an `extract` outside a subquery — it projects a column
+  subset, and with a `["function", ...]` column it aggregates. Injecting there
+  would break the row shape or silently inflate a `count()`, so **aggregate
+  results are never changed**. The whole query is walked, so an `extract` nested
+  under `and`/`or`/`not`/`from` skips injection too; an `extract` under `in`,
+  `subquery`, or `select_<entity>` projects that subquery rather than the
+  response, so it does not;
+- the query is not an AST array — every **PQL-syntax** query (`facts { certname
+  = "web1" }`) lands here. `pdbmux` cannot tell what such a query projects, so it
+  never injects into a PQL response. Use the AST form to get the fact;
+- (`/facts` only) the query constrains `name` — `["=","name","osfamily"]` and
+  friends ask for specific facts, and the synthetic record is not one of them.
+  Only the outer query is inspected: a `name` filter inside an `in`/`select_facts`
+  subquery narrows which *nodes* match, not which facts come back, so injection
+  still happens;
+- injection is turned off (see `source_fact_enabled`).
+
+**Not supported in v1: server-side filtering on the fact.** A query that selects
+it — `["=","name","pdbmux_source"]`, or an `extract` naming it — is forwarded to
+the backends like any other, and they return nothing, because the fact does not
+exist upstream. `pdbmux` does not evaluate the AST itself, so it cannot answer
+such a query correctly for every operator (`not`, `or`, subqueries) and does not
+pretend to for some. Read the fact from an unfiltered (or `certname`-filtered)
+`/facts` response and filter client-side. The same applies to the
+`/pdb/query/v4/facts/<name>` route, which is served unmerged pass-through.
+
+**Not covered:** `/factsets` and `/inventory`. Both carry facts, but `pdbmux`
+does not merge either today — they take the unmerged pass-through path, where
+the answer comes from whichever backend replied first rather than from a merge
+winner, so there is no owner to attribute. Injecting there would state a
+provenance that isn't true.
+
 ### Metadata and metrics
 
 - **`/pdb/meta/v1/version`** — when the backends agree, that version is served.
@@ -164,6 +232,8 @@ backends:                          # order is a tie-break only, not a ranking
 merge: freshness      # freshness | static
 timeout: 10s          # per-upstream request timeout
 freshness_ttl: 30s    # freshness-map cache TTL (freshness merge only)
+source_fact: pdbmux_source   # name of the synthetic provenance fact
+source_fact_enabled: true    # false serves backends' records untouched
 ```
 
 `backends[*].url` is a **base** URL (`scheme://host[:port]`); `pdbmux` appends
@@ -177,6 +247,8 @@ the `/pdb/query/v4/...` path per request.
 | `PDBMUX_TIMEOUT` | `timeout` (Go duration, e.g. `10s`) |
 | `PDBMUX_FRESHNESS_TTL` | `freshness_ttl` |
 | `PDBMUX_BACKENDS` | whole backend list, as `name=url,name=url` |
+| `PDBMUX_SOURCE_FACT` | `source_fact` (default `pdbmux_source`) |
+| `PDBMUX_SOURCE_FACT_ENABLED` | `source_fact_enabled` (default `true`); `false` disables injection |
 
 Flags: `--config`, `--listen`, `--merge`.
 

config.go

@@ -25,6 +25,8 @@ const (
 
 	defaultTimeout      = 10 * time.Second
 	defaultFreshnessTTL = 30 * time.Second
+
+	defaultSourceFact = "pdbmux_source"
 )
 
 var exampleBackends = []Backend{
@@ -44,6 +46,9 @@ type Config struct {
 	Timeout      time.Duration `yaml:"timeout"`
 	FreshnessTTL time.Duration `yaml:"freshness_ttl"`
 
+	SourceFact        string `yaml:"source_fact"`
+	SourceFactEnabled bool   `yaml:"source_fact_enabled"`
+
 	sourcePath string // file this config was read from, empty if none was found
 }
 
@@ -57,10 +62,12 @@ const (
 
 func DefaultConfig() Config {
 	return Config{
-		Listen:       defaultListen,
-		Merge:        mergeFreshness,
-		Timeout:      defaultTimeout,
-		FreshnessTTL: defaultFreshnessTTL,
+		Listen:            defaultListen,
+		Merge:             mergeFreshness,
+		Timeout:           defaultTimeout,
+		FreshnessTTL:      defaultFreshnessTTL,
+		SourceFact:        defaultSourceFact,
+		SourceFactEnabled: true,
 	}
 }
 
@@ -159,6 +166,14 @@ func applyEnv(cfg *Config, getenv func(string) string) {
 			cfg.FreshnessTTL = d
 		}
 	}
+	if v := getenv(envPrefix + "SOURCE_FACT"); v != "" {
+		cfg.SourceFact = v
+	}
+	if v := getenv(envPrefix + "SOURCE_FACT_ENABLED"); v != "" {
+		if b, err := strconv.ParseBool(v); err == nil {
+			cfg.SourceFactEnabled = b
+		}
+	}
 	if v := getenv(envPrefix + "BACKENDS"); v != "" {
 		if bs := parseBackends(v); len(bs) > 0 {
 			cfg.Backends = bs
@@ -216,6 +231,9 @@ func (c Config) Validate() error {
 	if c.Timeout <= 0 {
 		return fmt.Errorf("timeout must be positive")
 	}
+	if c.SourceFactEnabled && c.SourceFact == "" {
+		return fmt.Errorf("source_fact must be non-empty, or set source_fact_enabled to false")
+	}
 	return nil
 }
 
@@ -231,7 +249,8 @@ func writeDefaultConfig(path string) error {
 		"# A merging proxy presenting one PuppetDB v4 query surface over several\n" +
 		"# PuppetDB backends. The backend URLs below are placeholders — edit them.\n" +
 		"# Env overrides: PDBMUX_LISTEN, PDBMUX_MERGE, PDBMUX_TIMEOUT,\n" +
-		"#   PDBMUX_FRESHNESS_TTL, PDBMUX_BACKENDS (name=url,name=url).\n\n")
+		"#   PDBMUX_FRESHNESS_TTL, PDBMUX_BACKENDS (name=url,name=url),\n" +
+		"#   PDBMUX_SOURCE_FACT, PDBMUX_SOURCE_FACT_ENABLED.\n\n")
 	if err := os.WriteFile(path, append(header, data...), 0o644); err != nil {
 		return fmt.Errorf("writing config: %w", err)
 	}

config_test.go

@@ -118,6 +118,69 @@ func TestApplyEnv_Backends(t *testing.T) {
 	}
 }
 
+func TestDefaultConfig_SourceFact(t *testing.T) {
+	cfg := DefaultConfig()
+	if cfg.SourceFact != defaultSourceFact || !cfg.SourceFactEnabled {
+		t.Errorf("source fact defaults to %q enabled=%v, want %q enabled=true",
+			cfg.SourceFact, cfg.SourceFactEnabled, defaultSourceFact)
+	}
+}
+
+func TestApplyEnv_SourceFact(t *testing.T) {
+	cfg := testConfigValid()
+	env := map[string]string{envPrefix + "SOURCE_FACT": "origin_pdb"}
+	applyEnv(&cfg, func(k string) string { return env[k] })
+	if cfg.SourceFact != "origin_pdb" || !cfg.SourceFactEnabled {
+		t.Errorf("name override failed: %q enabled=%v", cfg.SourceFact, cfg.SourceFactEnabled)
+	}
+
+	cfg = testConfigValid()
+	env = map[string]string{envPrefix + "SOURCE_FACT_ENABLED": "false"}
+	applyEnv(&cfg, func(k string) string { return env[k] })
+	if cfg.SourceFactEnabled {
+		t.Error("PDBMUX_SOURCE_FACT_ENABLED=false must disable injection")
+	}
+
+	// A junk boolean leaves the default alone rather than disabling silently.
+	cfg = testConfigValid()
+	env = map[string]string{envPrefix + "SOURCE_FACT_ENABLED": "maybe"}
+	applyEnv(&cfg, func(k string) string { return env[k] })
+	if !cfg.SourceFactEnabled {
+		t.Error("unparseable bool must not change the setting")
+	}
+}
+
+// A config file omitting the key keeps the default; setting it false wins.
+func TestLoad_SourceFactFileOverride(t *testing.T) {
+	dir := t.TempDir()
+	t.Setenv("XDG_CONFIG_HOME", dir)
+	path := filepath.Join(dir, appName, configFileName)
+	if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	write := func(body string) Config {
+		t.Helper()
+		if err := os.WriteFile(path, []byte(body), 0o644); err != nil {
+			t.Fatal(err)
+		}
+		cfg, err := Load("")
+		if err != nil {
+			t.Fatal(err)
+		}
+		return cfg
+	}
+
+	cfg := write("backends:\n  - name: a\n    url: http://a:8080\n")
+	if cfg.SourceFact != defaultSourceFact || !cfg.SourceFactEnabled {
+		t.Errorf("omitted keys must keep defaults: %q enabled=%v", cfg.SourceFact, cfg.SourceFactEnabled)
+	}
+
+	cfg = write("backends:\n  - name: a\n    url: http://a:8080\nsource_fact: origin_pdb\nsource_fact_enabled: false\n")
+	if cfg.SourceFact != "origin_pdb" || cfg.SourceFactEnabled {
+		t.Errorf("file override failed: %q enabled=%v", cfg.SourceFact, cfg.SourceFactEnabled)
+	}
+}
+
 func TestValidate(t *testing.T) {
 	cases := []struct {
 		name    string
@@ -130,6 +193,8 @@ func TestValidate(t *testing.T) {
 		{"missing url", func(c *Config) { c.Backends[0].URL = "" }, true},
 		{"bad merge", func(c *Config) { c.Merge = "wrong" }, true},
 		{"zero timeout", func(c *Config) { c.Timeout = 0 }, true},
+		{"empty source fact while enabled", func(c *Config) { c.SourceFact = "" }, true},
+		{"empty source fact while disabled", func(c *Config) { c.SourceFact = ""; c.SourceFactEnabled = false }, false},
 	}
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {

main.go

@@ -161,6 +161,11 @@ func printConfig(cfg Config) {
 	fmt.Printf("merge        : %s\n", cfg.Merge)
 	fmt.Printf("timeout      : %s\n", durationString(cfg.Timeout))
 	fmt.Printf("freshness_ttl: %s\n", durationString(cfg.FreshnessTTL))
+	if cfg.SourceFactEnabled {
+		fmt.Printf("source_fact  : %s\n", cfg.SourceFact)
+	} else {
+		fmt.Printf("source_fact  : disabled\n")
+	}
 	fmt.Println("backends:")
 	for _, b := range cfg.Backends {
 		fmt.Printf("  - %-8s %s\n", b.Name, b.URL)

merge.go

@@ -11,12 +11,16 @@ type record struct {
 	Certname        string
 	ReportTimestamp string // only populated for /nodes records
 	Hash            string // only populated for /reports records
+	Name            string // only populated for /facts records
+	Environment     string
 }
 
 type recordMeta struct {
 	Certname        string `json:"certname"`
 	ReportTimestamp string `json:"report_timestamp"`
 	Hash            string `json:"hash"`
+	Name            string `json:"name"`
+	Environment     string `json:"environment"`
 }
 
 func decodeRecords(body []byte) ([]record, error) {
@@ -33,6 +37,8 @@ func decodeRecords(body []byte) ([]record, error) {
 			Certname:        m.Certname,
 			ReportTimestamp: m.ReportTimestamp,
 			Hash:            m.Hash,
+			Name:            m.Name,
+			Environment:     m.Environment,
 		})
 	}
 	return out, nil
@@ -50,10 +56,12 @@ func parseTimestamp(s string) time.Time {
 }
 
 // Ties keep the earlier backend's record — a deterministic tie-break, not a preference.
-func mergeNodes(results []backendResult) []json.RawMessage {
+// A non-nil inject stamps each surviving record with the backend that supplied it.
+func mergeNodes(results []backendResult, inject *sourceInjector) []json.RawMessage {
 	type pick struct {
-		raw json.RawMessage
-		ts  time.Time
+		raw     json.RawMessage
+		ts      time.Time
+		backend string
 	}
 	best := map[string]pick{}
 	var order []string
@@ -62,18 +70,19 @@ func mergeNodes(results []backendResult) []json.RawMessage {
 			ts := parseTimestamp(rec.ReportTimestamp)
 			cur, ok := best[rec.Certname]
 			if !ok {
-				best[rec.Certname] = pick{raw: rec.Raw, ts: ts}
+				best[rec.Certname] = pick{raw: rec.Raw, ts: ts, backend: res.name}
 				order = append(order, rec.Certname)
 				continue
 			}
 			if ts.After(cur.ts) {
-				best[rec.Certname] = pick{raw: rec.Raw, ts: ts}
+				best[rec.Certname] = pick{raw: rec.Raw, ts: ts, backend: res.name}
 			}
 		}
 	}
 	out := make([]json.RawMessage, 0, len(order))
 	for _, cn := range order {
-		out = append(out, best[cn].raw)
+		p := best[cn]
+		out = append(out, inject.stamp(p.raw, p.backend))
 	}
 	return out
 }
@@ -105,16 +114,17 @@ func buildFreshness(results []backendResult) freshness {
 }
 
 // owner names the winning backend per certname; a nil owner (static merge), or one holding no facts for that certname, falls back to configured order.
-func mergeFacts(results []backendResult, owner func(certname string) string) []json.RawMessage {
+// inject appends the synthetic source fact after each certname's block, naming the backend that won, and always drops upstream facts of that name.
+func mergeFacts(results []backendResult, owner func(certname string) string, inject *sourceInjector) []json.RawMessage {
 	present := map[string][]string{} // certname -> backend names, in configured order
-	byKey := map[string][]json.RawMessage{}
+	byKey := map[string][]record{}
 	for _, res := range results {
 		for _, rec := range res.records {
 			key := rec.Certname + "\x00" + res.name
 			if _, ok := byKey[key]; !ok {
 				present[rec.Certname] = append(present[rec.Certname], res.name)
 			}
-			byKey[key] = append(byKey[key], rec.Raw)
+			byKey[key] = append(byKey[key], rec)
 		}
 	}
 
@@ -140,11 +150,35 @@ func mergeFacts(results []backendResult, owner func(certname string) string) []j
 		if !contains(backends, chosen) {
 			chosen = backends[0]
 		}
-		out = append(out, byKey[cn+"\x00"+chosen]...)
+		recs := byKey[cn+"\x00"+chosen]
+		for _, rec := range recs {
+			// An upstream fact of the configured name is dropped on every query shape,
+			// injected or not: while the feature is on the name is pdbmux's alone.
+			if inject.claims(rec.Name) {
+				inject.suppressed++
+				continue
+			}
+			out = append(out, rec.Raw)
+		}
+		if !inject.injects() {
+			continue
+		}
+		if synth := inject.factRecord(cn, chosen, environmentOf(recs)); synth != nil {
+			out = append(out, synth)
+		}
 	}
 	return out
 }
 
+func environmentOf(recs []record) string {
+	for _, rec := range recs {
+		if rec.Environment != "" {
+			return rec.Environment
+		}
+	}
+	return ""
+}
+
 func contains(s []string, v string) bool {
 	for _, x := range s {
 		if x == v {

merge_test.go

@@ -92,7 +92,7 @@ func TestMergeNodes_NewerWins(t *testing.T) {
 	a := recs(t, "a", node("h1", "2026-07-01T00:00:00Z"), node("h2", "2026-07-10T00:00:00Z"))
 	b := recs(t, "b", node("h1", "2026-07-20T00:00:00Z"), node("h3", "2026-07-05T00:00:00Z"))
 
-	merged := mergeNodes([]backendResult{a, b})
+	merged := mergeNodes([]backendResult{a, b}, nil)
 	got := map[string]string{}
 	for _, r := range merged {
 		var m recordMeta
@@ -117,7 +117,7 @@ func TestMergeNodes_OneBackendOnly(t *testing.T) {
 	a := recs(t, "a", node("h1", "2026-07-01T00:00:00Z"))
 	// b returned nothing (e.g. empty result).
 	b := backendResult{name: "b"}
-	merged := mergeNodes([]backendResult{a, b})
+	merged := mergeNodes([]backendResult{a, b}, nil)
 	if len(merged) != 1 || certnames(t, merged)[0] != "h1" {
 		t.Fatalf("expected only h1, got %v", certnames(t, merged))
 	}
@@ -127,7 +127,7 @@ func TestMergeNodes_TieKeepsEarlierBackend(t *testing.T) {
 	// Equal timestamps: the backend listed first wins, as a tie-break.
 	first := recs(t, "b", node("h1", "2026-07-01T00:00:00Z"))
 	second := recs(t, "a", node("h1", "2026-07-01T00:00:00Z"))
-	merged := mergeNodes([]backendResult{first, second})
+	merged := mergeNodes([]backendResult{first, second}, nil)
 	if len(merged) != 1 {
 		t.Fatalf("expected 1 record, got %d", len(merged))
 	}
@@ -139,7 +139,7 @@ func TestMergeNodes_TieKeepsEarlierBackend(t *testing.T) {
 
 func TestMergeNodes_PreservesUnknownFields(t *testing.T) {
 	a := recs(t, "a", `{"certname":"h1","report_timestamp":"2026-07-01T00:00:00Z","extra":{"deep":42}}`)
-	merged := mergeNodes([]backendResult{a})
+	merged := mergeNodes([]backendResult{a}, nil)
 	if len(merged) != 1 {
 		t.Fatalf("expected 1 record")
 	}
@@ -156,7 +156,7 @@ func TestMergeFacts_NilOwnerUsesConfiguredOrder(t *testing.T) {
 	a := recs(t, "a", fact("h1", "role", "web-a", ""), fact("h2", "role", "db-a", ""))
 	b := recs(t, "b", fact("h1", "role", "web-b", ""))
 
-	merged := mergeFacts([]backendResult{b, a}, nil)
+	merged := mergeFacts([]backendResult{b, a}, nil, nil)
 	got := factValues(t, merged)
 	assertContains(t, got, "h1:role=web-b")
 	assertNotContains(t, got, "h1:role=web-a")
@@ -180,7 +180,7 @@ func TestMergeFacts_Freshness_NewerBackendWins(t *testing.T) {
 		}
 		return "b"
 	}
-	merged := mergeFacts([]backendResult{b, a}, owner)
+	merged := mergeFacts([]backendResult{b, a}, owner, nil)
 	got := factValues(t, merged)
 	// h1 -> all a facts, no b facts.
 	assertContains(t, got, "h1:role=web-a")
@@ -198,7 +198,7 @@ func TestMergeFacts_OwnerMissingFallsBackToConfiguredOrder(t *testing.T) {
 	// backend in the slice that has some.
 	first := recs(t, "b", fact("h1", "role", "web-b", ""))
 	second := recs(t, "a", fact("h1", "role", "web-a", ""))
-	merged := mergeFacts([]backendResult{first, second}, func(string) string { return "ghost" })
+	merged := mergeFacts([]backendResult{first, second}, func(string) string { return "ghost" }, nil)
 	got := factValues(t, merged)
 	assertContains(t, got, "h1:role=web-b") // b is first in slice
 	assertNotContains(t, got, "h1:role=web-a")

server.go

@@ -75,7 +75,7 @@ func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) {
 	case resourcesPath:
 		s.serveResources(w, r)
 	case factsPath:
-		s.serveMerged(w, r, factsPath, s.mergeFactsResponse)
+		s.serveMerged(w, r, factsPath, s.mergeFactsResponse(r))
 	case reportsPath:
 		s.serveReports(w, r)
 	case eventsPath:
@@ -146,7 +146,7 @@ func (s *Server) serveNodes(w http.ResponseWriter, r *http.Request) {
 		s.serveSummed(w, r, nodesPath, spec.columns)
 		return
 	}
-	s.serveMerged(w, r, nodesPath, s.mergeNodesResponse)
+	s.serveMerged(w, r, nodesPath, s.mergeNodesResponse(r))
 }
 
 // Only aggregates merge: a resource record has no cross-backend identity to dedupe on, so a plain query stays on the pass-through path.
@@ -248,16 +248,26 @@ func rawRecords(recs []record) []json.RawMessage {
 	return out
 }
 
-func (s *Server) mergeNodesResponse(results []backendResult) []json.RawMessage {
-	return mergeNodes(results)
+func (s *Server) mergeNodesResponse(r *http.Request) func([]backendResult) []json.RawMessage {
+	inject := s.newSourceInjector(r.URL.Query().Get("query"), false)
+	return func(results []backendResult) []json.RawMessage {
+		return mergeNodes(results, inject)
+	}
 }
 
-func (s *Server) mergeFactsResponse(results []backendResult) []json.RawMessage {
-	if s.cfg.Merge == mergeStatic {
-		return mergeFacts(results, nil)
+func (s *Server) mergeFactsResponse(r *http.Request) func([]backendResult) []json.RawMessage {
+	inject := s.newSourceInjector(r.URL.Query().Get("query"), true)
+	return func(results []backendResult) []json.RawMessage {
+		var merged []json.RawMessage
+		if s.cfg.Merge == mergeStatic {
+			merged = mergeFacts(results, nil, inject)
+		} else {
+			fresh := s.freshnessMap(context.Background(), results)
+			merged = mergeFacts(results, func(cn string) string { return fresh[cn] }, inject)
+		}
+		inject.logSuppressed(s.log)
+		return merged
 	}
-	fresh := s.freshnessMap(context.Background(), results)
-	return mergeFacts(results, func(cn string) string { return fresh[cn] })
 }
 
 // Queries /nodes unfiltered rather than reusing the request's results, because a /facts query's certname set can differ.

server_test.go

@@ -133,11 +133,13 @@ func truncate(t *testing.T, body, limit string) string {
 
 func testConfig(aURL, bURL, merge string) Config {
 	return Config{
-		Listen:       ":0",
-		Backends:     []Backend{{Name: "a", URL: aURL}, {Name: "b", URL: bURL}},
-		Merge:        merge,
-		Timeout:      2 * time.Second,
-		FreshnessTTL: 30 * time.Second,
+		Listen:            ":0",
+		Backends:          []Backend{{Name: "a", URL: aURL}, {Name: "b", URL: bURL}},
+		Merge:             merge,
+		Timeout:           2 * time.Second,
+		FreshnessTTL:      30 * time.Second,
+		SourceFact:        defaultSourceFact,
+		SourceFactEnabled: true,
 	}
 }
 

source.go

@@ -0,0 +1,198 @@
+package main
+
+import (
+	"encoding/json"
+	"log"
+	"strings"
+)
+
+// sourceInjector owns the configured fact name for one request. A nil
+// *sourceInjector is the feature-disabled case, so every method is nil-safe and
+// callers need no branch.
+type sourceInjector struct {
+	name string
+	// inject is false when the query shape rules synthesis out. Suppression of an
+	// upstream fact of the same name does not depend on it.
+	inject     bool
+	suppressed int
+}
+
+// newSourceInjector returns nil only when the feature is off; a gated query
+// yields an injector that suppresses but does not synthesise.
+func (s *Server) newSourceInjector(query string, factEntity bool) *sourceInjector {
+	if !s.cfg.SourceFactEnabled || s.cfg.SourceFact == "" {
+		return nil
+	}
+	return &sourceInjector{name: s.cfg.SourceFact, inject: injectable(query, factEntity)}
+}
+
+// claims reports whether an upstream record carries the name pdbmux owns, and is
+// keyed on the record's own name field: a projection that omits the name column
+// yields records that cannot be identified, so they pass through. Suppression
+// does not depend on the query gate.
+func (si *sourceInjector) claims(factName string) bool {
+	return si != nil && factName != "" && factName == si.name
+}
+
+// injects reports whether this response may carry the synthetic record.
+func (si *sourceInjector) injects() bool {
+	return si != nil && si.inject
+}
+
+// logSuppressed reports, once per request, that upstream records were dropped.
+func (si *sourceInjector) logSuppressed(l *log.Logger) {
+	if si == nil || si.suppressed == 0 || l == nil {
+		return
+	}
+	l.Printf("info: dropped %d upstream %q fact record(s); pdbmux owns that fact name", si.suppressed, si.name)
+}
+
+// factRecord builds the synthetic /facts record, or nil when disabled.
+// environment is copied from the node's real facts. All four keys of a fact
+// record are always emitted, empty environment included: pypuppetdb indexes them
+// directly (types.py Fact.create_from_dict), so an omitted key is a KeyError.
+func (si *sourceInjector) factRecord(certname, backend, environment string) json.RawMessage {
+	if !si.injects() {
+		return nil
+	}
+	raw, err := json.Marshal(struct {
+		Certname    string `json:"certname"`
+		Environment string `json:"environment"`
+		Name        string `json:"name"`
+		Value       string `json:"value"`
+	}{Certname: certname, Environment: environment, Name: si.name, Value: backend})
+	if err != nil {
+		return nil
+	}
+	return raw
+}
+
+// stamp adds the provenance key to a /nodes record, overwriting any existing
+// key of that name. A record that is not a JSON object passes through untouched.
+func (si *sourceInjector) stamp(raw json.RawMessage, backend string) json.RawMessage {
+	if !si.injects() {
+		return raw
+	}
+	var obj map[string]json.RawMessage
+	if json.Unmarshal(raw, &obj) != nil || obj == nil {
+		return raw
+	}
+	value, err := json.Marshal(backend)
+	if err != nil {
+		return raw
+	}
+	obj[si.name] = value
+	out, err := json.Marshal(obj)
+	if err != nil {
+		return raw
+	}
+	return out
+}
+
+// injectable reports whether a response to this query may carry the synthetic
+// record. Three shapes are excluded, each because the client asked for something
+// the synthetic record is not part of:
+//
+//   - a query that is not an AST array, which includes every PQL-syntax query:
+//     pdbmux cannot tell what it projects, so it changes nothing;
+//   - an `extract` anywhere in the query's own projection scope, which projects a
+//     column subset and, with a `["function", ...]` column, aggregates — injecting
+//     there would corrupt the row shape or silently inflate a count(). Operands
+//     scoped to a subquery are excluded; see hasExtract;
+//   - on the facts entity, any outer constraint on `name`, which selects
+//     specific facts. Subquery operands are not descended into: they choose which
+//     nodes match, not which facts come back.
+func injectable(query string, factEntity bool) bool {
+	if query == "" {
+		return true
+	}
+	var ast []json.RawMessage
+	if json.Unmarshal([]byte(query), &ast) != nil || len(ast) == 0 {
+		// Not an AST array pdbmux can reason about; leave the response alone.
+		return false
+	}
+	var op string
+	if json.Unmarshal(ast[0], &op) != nil {
+		return false
+	}
+	if hasExtract(ast) {
+		return false
+	}
+	if !factEntity {
+		return true
+	}
+	return !constrainsField(ast, "name")
+}
+
+// hasExtract reports whether an extract appears anywhere in the query's own
+// projection scope. openvoxdb accepts an extract as an operand of a boolean
+// operator — engine.clj's user-node->plan-node sends every and/or/not operand
+// back through itself (src/puppetlabs/puppetdb/query_eng/engine.clj:2697-2733)
+// and valid-operator? lists "extract" (:2779-2784) — so the row shape can be
+// rewritten below the top level, and the whole tree is walked to fail closed.
+//
+// Operators whose operand is scoped to a subquery are not descended into,
+// because an extract there projects the subquery rather than the response:
+//
+//   - `in`, whose operand becomes InExpression's :subquery (:2705-2712);
+//   - `subquery`, which the AST-rewrite stage expands into
+//     ["in" cols ["extract" cols ["select_<entity>" expr]]] (:2111-2123)
+//     before any plan node is built;
+//   - `select_<entity>`, the explicit subquery form (:1889-1911), reachable
+//     only under one of the two above in a query openvoxdb accepts.
+func hasExtract(parts []json.RawMessage) bool {
+	if len(parts) == 0 {
+		return false
+	}
+	var op string
+	if json.Unmarshal(parts[0], &op) != nil {
+		return false
+	}
+	switch {
+	case op == "extract":
+		return true
+	case op == "in", op == "subquery", strings.HasPrefix(op, "select_"):
+		return false
+	}
+	for _, p := range parts[1:] {
+		var sub []json.RawMessage
+		if json.Unmarshal(p, &sub) != nil {
+			continue
+		}
+		if hasExtract(sub) {
+			return true
+		}
+	}
+	return false
+}
+
+// constrainsField walks the boolean skeleton of an AST node looking for a
+// comparison whose field operand is field. Only and/or/not are descended into;
+// anything else, including the subquery operand of `in`, is left alone.
+func constrainsField(parts []json.RawMessage, field string) bool {
+	if len(parts) == 0 {
+		return false
+	}
+	var op string
+	if json.Unmarshal(parts[0], &op) != nil {
+		return false
+	}
+	switch op {
+	case "and", "or", "not":
+		for _, p := range parts[1:] {
+			var sub []json.RawMessage
+			if json.Unmarshal(p, &sub) != nil {
+				continue
+			}
+			if constrainsField(sub, field) {
+				return true
+			}
+		}
+		return false
+	}
+	if len(parts) < 2 {
+		return false
+	}
+	var name string
+	return json.Unmarshal(parts[1], &name) == nil && name == field
+}

source_test.go

@@ -0,0 +1,578 @@
+package main
+
+import (
+	"bytes"
+	"encoding/json"
+	"io"
+	"log"
+	"net/http"
+	"slices"
+	"strings"
+	"testing"
+)
+
+// sourceValues returns certname -> value of the synthetic fact record, and the
+// number of records carrying that fact name.
+func sourceValues(t *testing.T, body []byte, factName string) (map[string]string, int) {
+	t.Helper()
+	var raws []json.RawMessage
+	if err := json.Unmarshal(body, &raws); err != nil {
+		t.Fatalf("unmarshal %s: %v", body, err)
+	}
+	out := map[string]string{}
+	n := 0
+	for _, raw := range raws {
+		var m struct {
+			Certname string `json:"certname"`
+			Name     string `json:"name"`
+			Value    string `json:"value"`
+		}
+		if json.Unmarshal(raw, &m) != nil || m.Name != factName {
+			continue
+		}
+		out[m.Certname] = m.Value
+		n++
+	}
+	return out, n
+}
+
+// nodeSources returns certname -> the stamped provenance field on /nodes records.
+func nodeSources(t *testing.T, body []byte, field string) map[string]string {
+	t.Helper()
+	var raws []json.RawMessage
+	if err := json.Unmarshal(body, &raws); err != nil {
+		t.Fatalf("unmarshal %s: %v", body, err)
+	}
+	out := map[string]string{}
+	for _, raw := range raws {
+		var obj map[string]json.RawMessage
+		if err := json.Unmarshal(raw, &obj); err != nil {
+			t.Fatalf("unmarshal record %s: %v", raw, err)
+		}
+		var m recordMeta
+		_ = json.Unmarshal(raw, &m)
+		v, ok := obj[field]
+		if !ok {
+			continue
+		}
+		var s string
+		if err := json.Unmarshal(v, &s); err != nil {
+			t.Fatalf("provenance field of %s is not a string: %v", raw, err)
+		}
+		out[m.Certname] = s
+	}
+	return out
+}
+
+func factEnv(cn, name, val, env string) string {
+	return `{"certname":"` + cn + `","name":"` + name + `","value":"` + val + `","environment":"` + env + `"}`
+}
+
+// Both backends hold h1; a holds its newer report, so h1's facts and its
+// provenance fact must both come from a.
+func TestHandler_FactsSourceFollowsMergeOwner(t *testing.T) {
+	a := newFakeBackend(t,
+		`[`+node("h1", "2026-07-20T00:00:00Z")+`,`+node("h2", "2026-07-01T00:00:00Z")+`]`,
+		`[`+factEnv("h1", "role", "web-a", "production")+`,`+factEnv("h2", "role", "db-a", "production")+`]`)
+	b := newFakeBackend(t,
+		`[`+node("h1", "2026-07-01T00:00:00Z")+`,`+node("h2", "2026-07-20T00:00:00Z")+`]`,
+		`[`+factEnv("h1", "role", "web-b", "staging")+`,`+factEnv("h2", "role", "db-b", "staging")+`]`)
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeFreshness))
+
+	rec := doGet(t, srv.Handler(), factsPath, "")
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
+	}
+	got, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact)
+	if n != 2 {
+		t.Fatalf("expected one %s record per certname, got %d: %s", defaultSourceFact, n, rec.Body.String())
+	}
+	if got["h1"] != "a" || got["h2"] != "b" {
+		t.Errorf("provenance must name the backend that won the merge, got %v", got)
+	}
+}
+
+// The synthetic record carries the node's own environment so it groups with the
+// real facts rather than landing in an unrelated environment.
+func TestHandler_FactsSourceCopiesEnvironment(t *testing.T) {
+	a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`,
+		`[`+factEnv("h1", "role", "web", "staging")+`]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGet(t, srv.Handler(), factsPath, "")
+	var raws []json.RawMessage
+	if err := json.Unmarshal(rec.Body.Bytes(), &raws); err != nil {
+		t.Fatal(err)
+	}
+	var found bool
+	for _, raw := range raws {
+		var m recordMeta
+		if json.Unmarshal(raw, &m) != nil || m.Name != defaultSourceFact {
+			continue
+		}
+		found = true
+		if m.Environment != "staging" {
+			t.Errorf("environment = %q, want staging: %s", m.Environment, raw)
+		}
+	}
+	if !found {
+		t.Fatalf("no %s record: %s", defaultSourceFact, rec.Body.String())
+	}
+}
+
+// A node record's provenance names the backend whose node record won /nodes'
+// own report_timestamp merge.
+func TestHandler_NodesSourceStamped(t *testing.T) {
+	a := newFakeBackend(t,
+		`[`+node("h1", "2026-07-01T00:00:00Z")+`,`+node("h2", "2026-07-20T00:00:00Z")+`]`, `[]`)
+	b := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[]`)
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGet(t, srv.Handler(), nodesPath, "")
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
+	}
+	got := nodeSources(t, rec.Body.Bytes(), defaultSourceFact)
+	if got["h1"] != "b" || got["h2"] != "a" {
+		t.Errorf("node provenance = %v, want h1=b h2=a", got)
+	}
+}
+
+// Stamping must not drop unknown upstream fields.
+func TestHandler_NodesSourceKeepsUpstreamFields(t *testing.T) {
+	a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGet(t, srv.Handler(), nodesPath, "")
+	var raws []json.RawMessage
+	if err := json.Unmarshal(rec.Body.Bytes(), &raws); err != nil {
+		t.Fatal(err)
+	}
+	if len(raws) != 1 {
+		t.Fatalf("expected 1 node, got %d", len(raws))
+	}
+	var obj map[string]json.RawMessage
+	if err := json.Unmarshal(raws[0], &obj); err != nil {
+		t.Fatal(err)
+	}
+	for _, k := range []string{"certname", "report_timestamp", "latest_report_status", defaultSourceFact} {
+		if _, ok := obj[k]; !ok {
+			t.Errorf("field %q missing from stamped record: %s", k, raws[0])
+		}
+	}
+}
+
+func TestHandler_SourceDisabled(t *testing.T) {
+	a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`,
+		`[`+fact("h1", "role", "web", "")+`]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	cfg := testConfig(a.srv.URL, b.srv.URL, mergeStatic)
+	cfg.SourceFactEnabled = false
+	srv := newTestServer(cfg)
+
+	facts := doGet(t, srv.Handler(), factsPath, "")
+	if _, n := sourceValues(t, facts.Body.Bytes(), defaultSourceFact); n != 0 {
+		t.Errorf("disabled injection still produced %d records: %s", n, facts.Body.String())
+	}
+	nodes := doGet(t, srv.Handler(), nodesPath, "")
+	if got := nodeSources(t, nodes.Body.Bytes(), defaultSourceFact); len(got) != 0 {
+		t.Errorf("disabled injection still stamped nodes: %v", got)
+	}
+}
+
+func TestHandler_SourceFactNameOverride(t *testing.T) {
+	a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`,
+		`[`+fact("h1", "role", "web", "")+`]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	cfg := testConfig(a.srv.URL, b.srv.URL, mergeStatic)
+	cfg.SourceFact = "origin_pdb"
+	srv := newTestServer(cfg)
+
+	facts := doGet(t, srv.Handler(), factsPath, "")
+	got, n := sourceValues(t, facts.Body.Bytes(), "origin_pdb")
+	if n != 1 || got["h1"] != "a" {
+		t.Errorf("override name not honoured: %s", facts.Body.String())
+	}
+	if _, n := sourceValues(t, facts.Body.Bytes(), defaultSourceFact); n != 0 {
+		t.Errorf("default name still emitted alongside the override: %s", facts.Body.String())
+	}
+
+	nodes := doGet(t, srv.Handler(), nodesPath, "")
+	if got := nodeSources(t, nodes.Body.Bytes(), "origin_pdb"); got["h1"] != "a" {
+		t.Errorf("override name not honoured on /nodes: %v", got)
+	}
+}
+
+// An upstream fact of the configured name is replaced, not duplicated: pdbmux's
+// own attribution is authoritative.
+func TestHandler_UpstreamSourceFactOverridden(t *testing.T) {
+	a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`,
+		`[`+fact("h1", "role", "web", "")+`,`+fact("h1", defaultSourceFact, "stale-value", "")+`]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGet(t, srv.Handler(), factsPath, "")
+	got, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact)
+	if n != 1 {
+		t.Fatalf("expected exactly 1 %s record, got %d: %s", defaultSourceFact, n, rec.Body.String())
+	}
+	if got["h1"] != "a" {
+		t.Errorf("upstream value survived: %v", got)
+	}
+}
+
+// The configured name means one thing on every query shape: an upstream fact of
+// that name is dropped whether or not the query gate allows synthesis.
+func TestHandler_UpstreamSourceFactSuppressedOnEveryGateState(t *testing.T) {
+	const upstream = "REAL-UPSTREAM-VALUE"
+	tests := []struct {
+		name  string
+		query string
+		want  string // synthetic value, or "" when the gate blocks injection
+	}{
+		{"injection on", "", "a"},
+		{"gated by extract", `["extract",["certname","name","value"],["=","certname","h1"]]`, ""},
+		{"gated by name filter", `["=","name","` + defaultSourceFact + `"]`, ""},
+		{"gated by nested extract", `["and",["=","certname","h1"],["extract",["certname"]]]`, ""},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`,
+				`[`+fact("h1", "role", "web", "")+`,`+fact("h1", defaultSourceFact, upstream, "")+`]`)
+			b := newFakeBackend(t, `[]`, `[]`)
+			srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+			rec := doGet(t, srv.Handler(), factsPath, tc.query)
+			if rec.Code != http.StatusOK {
+				t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
+			}
+			if strings.Contains(rec.Body.String(), upstream) {
+				t.Fatalf("upstream value survived: %s", rec.Body.String())
+			}
+			got, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact)
+			if tc.want == "" {
+				if n != 0 {
+					t.Errorf("gated query produced %d %s records: %s", n, defaultSourceFact, rec.Body.String())
+				}
+			} else if n != 1 || got["h1"] != tc.want {
+				t.Errorf("provenance = %v (%d records), want h1=%s: %s", got, n, tc.want, rec.Body.String())
+			}
+			if !strings.Contains(rec.Body.String(), `"role"`) {
+				t.Errorf("real facts were dropped: %s", rec.Body.String())
+			}
+		})
+	}
+}
+
+// Dropping an upstream record is invisible in the response, so it is logged —
+// once per request, not once per record.
+func TestHandler_SuppressedUpstreamFactLoggedOncePerRequest(t *testing.T) {
+	a := newFakeBackend(t,
+		`[`+node("h1", "2026-07-20T00:00:00Z")+`,`+node("h2", "2026-07-20T00:00:00Z")+`]`,
+		`[`+fact("h1", defaultSourceFact, "old-h1", "")+`,`+fact("h2", defaultSourceFact, "old-h2", "")+`]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	var buf bytes.Buffer
+	srv := NewServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic), log.New(&buf, "", 0))
+
+	doGet(t, srv.Handler(), factsPath, "")
+	if n := strings.Count(buf.String(), defaultSourceFact); n != 1 {
+		t.Fatalf("expected 1 log line naming the fact, got %d: %s", n, buf.String())
+	}
+	if !strings.Contains(buf.String(), "dropped 2") {
+		t.Errorf("log does not report the number dropped: %s", buf.String())
+	}
+}
+
+// Disabled means untouched: an upstream fact of the configured name is served as
+// the backend reported it.
+func TestHandler_SourceDisabledKeepsUpstreamFact(t *testing.T) {
+	a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`,
+		`[`+fact("h1", defaultSourceFact, "upstream-value", "")+`]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	cfg := testConfig(a.srv.URL, b.srv.URL, mergeStatic)
+	cfg.SourceFactEnabled = false
+	srv := newTestServer(cfg)
+
+	rec := doGet(t, srv.Handler(), factsPath, "")
+	got, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact)
+	if n != 1 || got["h1"] != "upstream-value" {
+		t.Errorf("disabled injection altered the upstream fact: %s", rec.Body.String())
+	}
+}
+
+// Suppression matches a record's own name field, so rows from a projection that
+// filters on name without returning it are not self-identifying and pass
+// through. Pinned as a known limit of the guarantee, and documented as one.
+func TestHandler_ProjectionWithoutNameColumnCarriesUpstreamValue(t *testing.T) {
+	a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	a.bodies[factsPath] = `[{"certname":"h1","value":"upstream-value"}]`
+	var buf bytes.Buffer
+	srv := NewServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic), log.New(&buf, "", 0))
+
+	rec := doGet(t, srv.Handler(), factsPath,
+		`["extract",["certname","value"],["=","name","`+defaultSourceFact+`"]]`)
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
+	}
+	if got := rec.Body.String(); !strings.Contains(got, "upstream-value") {
+		t.Errorf("unidentifiable row was dropped: %s", got)
+	}
+	if strings.Contains(buf.String(), "dropped") {
+		t.Errorf("a row with no name field was counted as suppressed: %s", buf.String())
+	}
+	if _, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact); n != 0 {
+		t.Errorf("gated projection gained %d synthetic records: %s", n, rec.Body.String())
+	}
+}
+
+// Covers only that an aggregate /facts response gains no synthetic record and no
+// rewritten row. It does not cover whether the aggregate rows are correct:
+// /facts has no parseAggregate branch, so its rows take the certname merge
+// instead of serveSummed and are not summed across backends.
+func TestHandler_SourceNotInjectedOnFactsAggregate(t *testing.T) {
+	a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	a.bodies[factsPath] = `[{"count":3}]`
+	b.bodies[factsPath] = `[{"count":2}]`
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGet(t, srv.Handler(), factsPath, `["extract",[["function","count"]]]`)
+	if _, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact); n != 0 {
+		t.Errorf("aggregate response gained %d synthetic records: %s", n, rec.Body.String())
+	}
+	// Every row must be one an upstream actually returned: an aggregate row pdbmux
+	// invented or rewrote would change the count the client sees.
+	var raws []json.RawMessage
+	if err := json.Unmarshal(rec.Body.Bytes(), &raws); err != nil {
+		t.Fatal(err)
+	}
+	upstream := []string{`{"count":3}`, `{"count":2}`}
+	for _, raw := range raws {
+		if !slices.Contains(upstream, string(raw)) {
+			t.Errorf("aggregate row %s is not an upstream row", raw)
+		}
+	}
+}
+
+// /nodes aggregates are summed rather than merged, so nothing may stamp them.
+func TestHandler_NodesAggregateNotStamped(t *testing.T) {
+	a := newFakeBackend(t, `[]`, `[]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	a.bodies[nodesPath] = `[{"count":3}]`
+	b.bodies[nodesPath] = `[{"count":2}]`
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGet(t, srv.Handler(), nodesPath, `["extract",[["function","count"]]]`)
+	if strings.Contains(rec.Body.String(), defaultSourceFact) {
+		t.Errorf("aggregate rows were stamped: %s", rec.Body.String())
+	}
+	if got := rec.Body.String(); !strings.Contains(got, `"count":5`) {
+		t.Errorf("count = %s, want the summed 5", got)
+	}
+}
+
+// A plain extract projects columns and skips the aggregate path, so the stamp
+// must not add a key the client did not ask for.
+func TestHandler_NodesProjectionNotStamped(t *testing.T) {
+	a := newFakeBackend(t, `[]`, `[]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	a.bodies[nodesPath] = `[{"certname":"h1"}]`
+	b.bodies[nodesPath] = `[]`
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGet(t, srv.Handler(), nodesPath, `["extract",["certname"]]`)
+	if strings.Contains(rec.Body.String(), defaultSourceFact) {
+		t.Errorf("projection gained a stamp: %s", rec.Body.String())
+	}
+}
+
+// A query naming a specific fact asked for that fact only.
+func TestHandler_SourceNotInjectedWhenNameFiltered(t *testing.T) {
+	a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`,
+		`[`+fact("h1", "role", "web", "")+`]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	for _, q := range []string{
+		`["=","name","role"]`,
+		`["and",["=","certname","h1"],["=","name","role"]]`,
+		`["=","name","` + defaultSourceFact + `"]`,
+	} {
+		rec := doGet(t, srv.Handler(), factsPath, q)
+		if _, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact); n != 0 {
+			t.Errorf("query %s gained %d synthetic records: %s", q, n, rec.Body.String())
+		}
+	}
+}
+
+// A certname filter selects nodes, not facts, so the full fact set — synthetic
+// record included — is still the right answer.
+func TestHandler_SourceInjectedWhenOnlyCertnameFiltered(t *testing.T) {
+	a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`,
+		`[`+fact("h1", "role", "web", "")+`]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGet(t, srv.Handler(), factsPath, `["=","certname","h1"]`)
+	if _, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact); n != 1 {
+		t.Errorf("expected 1 synthetic record, got %d: %s", n, rec.Body.String())
+	}
+}
+
+func TestInjectable(t *testing.T) {
+	tests := []struct {
+		name       string
+		query      string
+		factEntity bool
+		want       bool
+	}{
+		{"empty query", "", true, true},
+		{"certname filter", `["=","certname","h1"]`, true, true},
+		{"regex certname filter", `["~","certname","^web"]`, true, true},
+		{"name filter", `["=","name","os"]`, true, false},
+		{"name regex filter", `["~","name","^net"]`, true, false},
+		{"name under and", `["and",["=","certname","h1"],["=","name","os"]]`, true, false},
+		{"name under or", `["or",["=","name","os"],["=","name","kernel"]]`, true, false},
+		{"name under not", `["not",["=","name","os"]]`, true, false},
+		{"name in list", `["in","name",["array",["os"]]]`, true, false},
+		// A select_facts subquery narrows which nodes match; the outer response is
+		// still whole fact sets, so the synthetic record belongs in it.
+		{"name only inside subquery", `["in","certname",["extract",["certname"],["select_facts",["=","name","os"]]]]`, true, true},
+		{"top-level extract", `["extract",["certname","value"],["=","certname","h1"]]`, true, false},
+		{"aggregate extract", `["extract",[["function","count"]]]`, true, false},
+		// openvoxdb hands each boolean operand back to user-node->plan-node, which
+		// builds an extract node from it, so a nested extract can reshape the rows.
+		{"extract under and", `["and",["=","certname","h1"],["extract",["certname"]]]`, true, false},
+		{"extract under or", `["or",["extract",["certname"]],["=","certname","h1"]]`, true, false},
+		{"extract under not", `["not",["extract",["certname"]]]`, true, false},
+		{"extract nested two deep", `["and",["or",["extract",[["function","count"]]]]]`, true, false},
+		{"extract under from", `["from","facts",["extract",["certname"]]]`, true, false},
+		{"nested extract on nodes", `["and",["extract",["certname"]]]`, false, false},
+		// An extract inside an `in` operand projects the subquery, not the response.
+		{"extract under in stays injectable", `["in","certname",["extract",["certname"],["select_facts",["=","name","os"]]]]`, true, true},
+		// `subquery` is rewritten to ["in" ... ["extract" ... ["select_x" ...]]]
+		// before any plan node is built, so its operand is subquery-scoped too.
+		{"subquery operand stays injectable", `["and",["=","certname","h1"],["subquery","facts",["extract",["certname"],["=","name","os"]]]]`, true, true},
+		{"bare subquery stays injectable", `["subquery","facts",["extract",["certname"]]]`, true, true},
+		{"extract under select_facts stays injectable", `["and",["select_facts",["extract",["certname"]]]]`, true, true},
+		{"nodes name filter is not a fact filter", `["=","name","os"]`, false, true},
+		{"nodes extract", `["extract",["certname"]]`, false, false},
+		{"unparseable query", `not json`, true, false},
+		{"non-array query", `{"a":1}`, true, false},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := injectable(tc.query, tc.factEntity); got != tc.want {
+				t.Errorf("injectable(%s, %v) = %v, want %v", tc.query, tc.factEntity, got, tc.want)
+			}
+		})
+	}
+}
+
+// pypuppetdb reads certname/name/value/environment by direct index, so a
+// missing key is a KeyError there — all four are always present.
+func TestSourceInjector_FactRecordHasEveryFactKey(t *testing.T) {
+	si := &sourceInjector{name: defaultSourceFact, inject: true}
+	for _, env := range []string{"production", ""} {
+		var obj map[string]json.RawMessage
+		if err := json.Unmarshal(si.factRecord("h1", "a", env), &obj); err != nil {
+			t.Fatal(err)
+		}
+		for _, k := range []string{"certname", "name", "value", "environment"} {
+			if _, ok := obj[k]; !ok {
+				t.Errorf("environment=%q: key %q missing from synthetic fact", env, k)
+			}
+		}
+		if len(obj) != 4 {
+			t.Errorf("synthetic fact has %d keys, want the 4 of a real fact record: %v", len(obj), obj)
+		}
+	}
+}
+
+// A nil injector is the disabled path and must leave every input untouched.
+func TestSourceInjector_NilIsInert(t *testing.T) {
+	var si *sourceInjector
+	if si.claims(defaultSourceFact) {
+		t.Error("nil injector claims a fact name")
+	}
+	if si.injects() {
+		t.Error("nil injector injects")
+	}
+	si.logSuppressed(log.New(io.Discard, "", 0))
+	if si.factRecord("h1", "a", "production") != nil {
+		t.Error("nil injector produced a record")
+	}
+	raw := json.RawMessage(`{"certname":"h1"}`)
+	if got := si.stamp(raw, "a"); string(got) != string(raw) {
+		t.Errorf("nil injector rewrote %s to %s", raw, got)
+	}
+}
+
+// A response element that is not a JSON object cannot be stamped, and must be
+// passed through rather than dropped or mangled.
+func TestSourceInjector_StampNonObject(t *testing.T) {
+	si := &sourceInjector{name: defaultSourceFact, inject: true}
+	for _, raw := range []string{`"scalar"`, `[1,2]`, `null`} {
+		if got := si.stamp(json.RawMessage(raw), "a"); string(got) != raw {
+			t.Errorf("stamp(%s) = %s, want unchanged", raw, got)
+		}
+	}
+}
+
+// A gated injector suppresses without synthesising anything, on either endpoint.
+func TestSourceInjector_GatedSuppressesButDoesNotInject(t *testing.T) {
+	si := &sourceInjector{name: defaultSourceFact}
+	if !si.claims(defaultSourceFact) {
+		t.Error("gated injector does not claim its own fact name")
+	}
+	if si.factRecord("h1", "a", "production") != nil {
+		t.Error("gated injector produced a record")
+	}
+	raw := json.RawMessage(`{"certname":"h1"}`)
+	if got := si.stamp(raw, "a"); string(got) != string(raw) {
+		t.Errorf("gated injector rewrote %s to %s", raw, got)
+	}
+}
+
+// The disabled path must return the backends' records verbatim, source fact
+// included.
+func TestMergeFacts_DisabledIsUntouched(t *testing.T) {
+	a := recs(t, "a", fact("h1", "role", "web-a", ""), fact("h1", defaultSourceFact, "upstream", ""))
+	merged := mergeFacts([]backendResult{a}, nil, nil)
+
+	got := factValues(t, merged)
+	want := []string{"h1:role=web-a", "h1:" + defaultSourceFact + "=upstream"}
+	if !slices.Equal(got, want) {
+		t.Errorf("merged = %v, want %v", got, want)
+	}
+}
+
+// A gated query still gets the upstream record removed, and nothing added.
+func TestMergeFacts_GatedSuppressesUpstream(t *testing.T) {
+	a := recs(t, "a", fact("h1", "role", "web-a", ""), fact("h1", defaultSourceFact, "upstream", ""))
+	si := &sourceInjector{name: defaultSourceFact}
+	merged := mergeFacts([]backendResult{a}, nil, si)
+
+	got := factValues(t, merged)
+	if !slices.Equal(got, []string{"h1:role=web-a"}) {
+		t.Errorf("merged = %v, want only the real fact", got)
+	}
+	if si.suppressed != 1 {
+		t.Errorf("suppressed = %d, want 1", si.suppressed)
+	}
+}
+
+func TestMergeFacts_SourceOrderedAfterOwnersFacts(t *testing.T) {
+	a := recs(t, "a", fact("h1", "role", "web-a", ""), fact("h1", "kernel", "Linux", ""))
+	b := recs(t, "b", fact("h1", "role", "web-b", ""))
+	merged := mergeFacts([]backendResult{a, b}, nil, &sourceInjector{name: defaultSourceFact, inject: true})
+
+	got := factValues(t, merged)
+	want := []string{"h1:role=web-a", "h1:kernel=Linux", "h1:" + defaultSourceFact + "=a"}
+	if !slices.Equal(got, want) {
+		t.Errorf("merged = %v, want %v", got, want)
+	}
+}