feat: dynamic include-values source for the keep-filter

AGENT.md

@@ -232,6 +232,19 @@ Cast direction is values → column (the small array to the big one's type), not
 
 `MILLPOND_FILTER_DROP_FIELD_NAME` is reserved at the config layer (mutex with keep, both empty or exactly one set) and explicitly rejected at startup. The denylist implementation lives in a future commit; the namespace is locked today so that change doesn't require operator env-var churn.
 
+**Include-values source** (`include_values.py`) supplies the allowlist `_apply_filter` reads — the values are now an explicit third argument, fetched per batch via `include_source.current()`, not read off cfg. `build(cfg)` returns exactly ONE object (`StaticIncludeValues` | `ShadowIncludeValues` | `HttpIncludeValues`, distinguished by `.mode`); main starts/stops it around the consume loop and `metrics.include_values_mode{mode}` reports which one actually runs (fleet queries must not pass vacuously on replicas that never got the URL).
+
+Design spine is the consequence asymmetry: erroneous addition = surplus rows downstream ignores; erroneous removal = silent unrecoverable drop. The invariants, all tested in `tests/unit/test_include_values.py`:
+
+- Additions apply on first sight; removals need `removal_confirm_polls` consecutive successful ACCEPTED polls absent. Failed polls and refused polls advance nothing — refusal must never pre-charge the countdown (N refused-empty polls + one junk value would otherwise mass-remove instantly).
+- Refusal guards (whole poll rejected, `include_values_refused_total{reason}`): `empty` (an empty array is never removal evidence and never an acceptable first state), `bulk_removal` (> half of a multi-value set confirmed at once — world-replacement goes through static config, not an unattended poll), `type_flip` (int↔str: one junk element flips `_normalize` to strings and the filter's values→column cast then fails, dropping ENTIRE batches as `filter_field_missing` with offsets still committed — the worst silent-loss path in the module).
+- Authoritative `start()` BLOCKS until the first successful poll and raises on timeout — no proceed-on-bootstrap. The damping state is in-memory; a restart that silently ran on a stale static bootstrap would perform de-facto removals with zero polls of evidence. Halting is recoverable from Kafka; a silent drop is not.
+- Shadow mode: `current()` serves the static set; the prober is a private attribute precisely so reading the polled set as authority requires reaching into `_private`, not a one-line mistake.
+- Thread-safety: `current()` reads an immutable tuple swapped by attribute assignment (single writer = the poll thread); everything else is poll-thread-only. No locks.
+- `_fetch` refuses redirects (urllib re-sends the auth header cross-host otherwise), caps the response at 4 MB, and jitters the poll ±10% so replica fleets don't herd.
+
+Config validation (`_load_include_values_config` in `config.py`) refuses at startup: URL without keep-filter, MODE/auth without URL (an intended dynamic source silently degrading to static is a removal-by-omission), lone auth halves, non-numeric knobs. The URL⇒keep-field⇒non-empty-static-values chain is what makes the no-bootstrap path unreachable through config — `test_keep_filter_chain_guarantees_bootstrap` pins it.
+
 **Sort** (`_apply_sort` in `main.py`) runs inside `_flush()` after `pa.concat_tables` but before `sink.write()`. The sink sees pre-sorted data; sink-side partition columns (year/month/day/hour, computed by the ducklake extension) are not in scope by design — operators specify sort keys against the source schema.
 
 Missing-field handling: if any `cfg.sort_by` field is absent from the batch, the whole sort is skipped (rather than partially sorting on available keys, which would silently differ from intent). Records still flow through unsorted. The metric is `sort_skipped_total{reason="field_missing"}`, deliberately distinct from `records_skipped_total` because no data is being dropped — only the layout improvement is.

README.md

@@ -79,6 +79,30 @@ Two skip reasons are tracked on `millpond_records_skipped_total`:
 
 `MILLPOND_FILTER_DROP_FIELD_NAME` is reserved at the config layer (mutex with keep) and currently rejected at startup. It will become a denylist filter in a future release without env-var churn.
 
+### Dynamic allowlist source
+
+The allowlist can be sourced at runtime from an HTTP endpoint instead of being fixed at startup: `MILLPOND_INCLUDE_VALUES_URL` names a URL returning a JSON array of scalars (ints or strings, matching the static list's type), polled on a background thread (default every 60s, ±10% jitter). Millpond knows nothing about the endpoint's meaning — the URL and an optional auth header (`MILLPOND_INCLUDE_VALUES_AUTH_HEADER_NAME` + `_AUTH_TOKEN`) are plain config.
+
+How the static list and the polled set interact:
+
+| `MILLPOND_FILTER_VALUES` (static) | `MILLPOND_INCLUDE_VALUES_URL` | `MILLPOND_INCLUDE_VALUES_MODE` | Effective allowlist | Endpoint's role |
+|---|---|---|---|---|
+| set | unset | unset | the static list | none — today's behavior, unchanged |
+| set | set | `shadow` (default) | the static list | observability only: polled each interval, exports diff-vs-static gauges and staleness; its values are never applied |
+| set | set | `authoritative` | the polled set | live: the static list seeds the initial held set; startup **blocks** until the first successful poll (no proceed-on-stale-bootstrap) |
+| unset | unset | unset | no filter — all records kept | — |
+| unset | set | any | **startup error** | the URL requires an active keep-filter, which requires static values |
+| any | unset | set (or auth vars set) | **startup error** | MODE/auth without a URL means a dynamic source was intended; refusing beats silently running static-only |
+
+In `authoritative` mode the polled set changes under safety rules shaped by a consequence asymmetry — an erroneous addition writes surplus rows, an erroneous removal silently drops records with no recovery:
+
+- **Additions** apply on the first successful poll that shows them.
+- **Removals** require `MILLPOND_INCLUDE_VALUES_REMOVAL_POLLS` (default 5) *consecutive successful* polls with the value absent. Failed polls freeze the countdown; a reappearing value resets it.
+- **Poll failures** keep the last-known-good set indefinitely; staleness is observable via `millpond_include_values_last_success_timestamp_seconds`.
+- **Refused polls** (counted on `millpond_include_values_refused_total{reason}`) keep the set and advance nothing: empty arrays (`empty` — never removal evidence, never an acceptable first state), removals of more than half of a multi-value set at once (`bulk_removal`), and int↔str type changes (`type_flip` — a type-flipped set would fail the filter's cast against the column and drop whole batches).
+
+Rollout is designed to be shadow-first: run `shadow`, alert on `millpond_include_values_shadow_only_static` / `_shadow_only_remote` staying nonzero, and flip to `authoritative` once the symmetric difference holds at zero. `millpond_include_values_mode` reports which mode each replica actually runs, so a fleet-level flip gate can't pass vacuously on a replica that never got the URL.
+
 ### Pre-write sort
 
 Sorts the consolidated batch by one or more columns ascending, right before `sink.write()`. The sink sees pre-sorted data, which improves Parquet compression (especially for low-cardinality keys like `team_id`) and downstream reader predicate pushdown.
@@ -222,6 +246,14 @@ See [Record Handling](#record-handling) for context. All four variables below ar
 | `MILLPOND_FILTER_KEEP_FIELD_NAME` | no | | Column name to check against the allowlist. Must be set with `MILLPOND_FILTER_VALUES`. Validated as a safe identifier. |
 | `MILLPOND_FILTER_DROP_FIELD_NAME` | no | | Reserved for a future denylist filter; setting it today raises at startup. Mutually exclusive with `MILLPOND_FILTER_KEEP_FIELD_NAME`. |
 | `MILLPOND_FILTER_VALUES` | no | | Comma-separated allowed values. Auto-detected as int if every token parses as an integer, string otherwise. Required when either filter field name is set. |
+| `MILLPOND_INCLUDE_VALUES_URL` | no | | HTTP endpoint returning a JSON array of allowlist values (see [Dynamic allowlist source](#dynamic-allowlist-source)). Requires the keep-filter to be configured. |
+| `MILLPOND_INCLUDE_VALUES_MODE` | no | `shadow` | `shadow` (static authoritative, endpoint observed for diff metrics) or `authoritative` (polled set live). Only valid with the URL set. |
+| `MILLPOND_INCLUDE_VALUES_POLL_INTERVAL_S` | no | `60` | Poll cadence, jittered ±10%. |
+| `MILLPOND_INCLUDE_VALUES_REMOVAL_POLLS` | no | `5` | Consecutive successful polls a value must be absent before removal. `1` disables damping (warned at startup). |
+| `MILLPOND_INCLUDE_VALUES_REQUEST_TIMEOUT_S` | no | `10` | Per-request HTTP timeout. |
+| `MILLPOND_INCLUDE_VALUES_STARTUP_TIMEOUT_S` | no | `60` | Authoritative mode: how long startup blocks for the first successful poll before failing the pod. |
+| `MILLPOND_INCLUDE_VALUES_AUTH_HEADER_NAME` | no | | Header name sent with each poll (e.g. an internal-secret header). Must be set together with the token. Redirects are refused so the header can't leak cross-host. |
+| `MILLPOND_INCLUDE_VALUES_AUTH_TOKEN` | no | | Header value. Must be set together with the header name. |
 | `MILLPOND_SORT_BY` | no | | Comma-separated column names; the batch is sorted ascending by these in tuple order before each write. Missing fields cause the sort to be skipped (records still flow). |
 | `MILLPOND_TYPED_COLUMNS` | no | | Comma-separated `column:type` pairs pinning columns to a target type before write (types: `timestamptz`, `bigint`, `double`, `boolean`, `varchar`). Needed when writing into a table whose columns are already typed and JSON inference would diverge (date-times → `VARCHAR` vs `TIMESTAMPTZ`; all-null `project_id` → `VARCHAR` vs `BIGINT`). Column names validated as safe identifiers; types validated against the allowlist. |
 

millpond/config.py

@@ -77,6 +77,22 @@ class Config:
     filter_drop_field: str | None
     filter_values: tuple[int, ...] | tuple[str, ...] | None
 
+    # Optional dynamic source for the keep-filter's include set (see
+    # include_values.py). URL unset = today's static behavior. Mode
+    # "shadow" (default) polls and reports diff metrics while the static
+    # list stays authoritative; "authoritative" makes the polled set the
+    # live filter, with the static list as the bootstrap/fallback seed.
+    # The auth header is generic (name + token) — nothing here knows what
+    # the endpoint is.
+    include_values_url: str | None
+    include_values_mode: str
+    include_values_poll_interval_s: float
+    include_values_removal_polls: int
+    include_values_request_timeout_s: float
+    include_values_startup_timeout_s: float
+    include_values_auth_header_name: str | None
+    include_values_auth_token: str | None
+
     # Optional pre-write sort. Tuple of column names; sort is ascending
     # in tuple order. Applied to the consolidated batch right before
     # sink.write(). None disables the sort entirely.
@@ -213,6 +229,91 @@ def _load_filter_fields() -> tuple[str | None, str | None, tuple[int, ...] | tup
     return keep, None, _parse_filter_values(values_raw)
 
 
+def _load_include_values_config(
+    filter_keep_field: str | None,
+    filter_values: tuple[int, ...] | tuple[str, ...] | None,
+) -> dict:
+    """Read the MILLPOND_INCLUDE_VALUES_* group and validate it against the
+    static filter config. Startup-refusal beats runtime surprise:
+
+    - a URL without an active keep-filter has nothing to feed;
+    - MODE (or auth) without a URL means the operator INTENDED a dynamic
+      source and a typo'd/missing URL would silently degrade to static —
+      refuse rather than run on the wrong set;
+    - shadow mode without static values has nothing to diff against;
+    - a lone auth header name or token is always a misconfiguration.
+    """
+    url = os.environ.get("MILLPOND_INCLUDE_VALUES_URL", "").strip() or None
+    mode_raw = os.environ.get("MILLPOND_INCLUDE_VALUES_MODE", "").strip().lower()
+    mode = mode_raw or "shadow"
+    header_name = os.environ.get("MILLPOND_INCLUDE_VALUES_AUTH_HEADER_NAME", "").strip() or None
+    token = os.environ.get("MILLPOND_INCLUDE_VALUES_AUTH_TOKEN", "").strip() or None
+
+    if url is None:
+        if mode_raw:
+            raise RuntimeError(
+                "MILLPOND_INCLUDE_VALUES_MODE is set but MILLPOND_INCLUDE_VALUES_URL is not — "
+                "a dynamic source was intended; refusing to silently run static-only"
+            )
+        if header_name or token:
+            raise RuntimeError("MILLPOND_INCLUDE_VALUES_AUTH_* requires MILLPOND_INCLUDE_VALUES_URL")
+        return dict(
+            include_values_url=None,
+            include_values_mode="static",
+            include_values_poll_interval_s=60.0,
+            include_values_removal_polls=5,
+            include_values_request_timeout_s=10.0,
+            include_values_startup_timeout_s=60.0,
+            include_values_auth_header_name=None,
+            include_values_auth_token=None,
+        )
+
+    if mode not in ("shadow", "authoritative"):
+        raise RuntimeError(f"MILLPOND_INCLUDE_VALUES_MODE must be 'shadow' or 'authoritative', got {mode!r}")
+    if filter_keep_field is None:
+        raise RuntimeError("MILLPOND_INCLUDE_VALUES_URL requires MILLPOND_FILTER_KEEP_FIELD_NAME to be set")
+    if mode == "shadow" and filter_values is None:
+        raise RuntimeError("MILLPOND_INCLUDE_VALUES_MODE=shadow requires static MILLPOND_FILTER_VALUES to diff against")
+    if bool(header_name) != bool(token):
+        raise RuntimeError(
+            "MILLPOND_INCLUDE_VALUES_AUTH_HEADER_NAME and MILLPOND_INCLUDE_VALUES_AUTH_TOKEN must be set together"
+        )
+
+    def _parse_number(env_name: str, default: str, cast):
+        raw = os.environ.get(env_name, default)
+        try:
+            return cast(raw)
+        except ValueError:
+            raise RuntimeError(f"{env_name} must be a number, got {raw!r}") from None
+
+    poll_interval = _parse_number("MILLPOND_INCLUDE_VALUES_POLL_INTERVAL_S", "60", float)
+    removal_polls = _parse_number("MILLPOND_INCLUDE_VALUES_REMOVAL_POLLS", "5", int)
+    request_timeout = _parse_number("MILLPOND_INCLUDE_VALUES_REQUEST_TIMEOUT_S", "10", float)
+    startup_timeout = _parse_number("MILLPOND_INCLUDE_VALUES_STARTUP_TIMEOUT_S", "60", float)
+    if poll_interval <= 0:
+        raise RuntimeError("MILLPOND_INCLUDE_VALUES_POLL_INTERVAL_S must be positive")
+    if removal_polls < 1:
+        raise RuntimeError("MILLPOND_INCLUDE_VALUES_REMOVAL_POLLS must be >= 1")
+    if removal_polls == 1:
+        log.warning(
+            "MILLPOND_INCLUDE_VALUES_REMOVAL_POLLS=1 disables removal damping — a single poll "
+            "omitting a value removes it immediately"
+        )
+    if request_timeout <= 0 or startup_timeout <= 0:
+        raise RuntimeError("MILLPOND_INCLUDE_VALUES_*_TIMEOUT_S must be positive")
+
+    return dict(
+        include_values_url=url,
+        include_values_mode=mode,
+        include_values_poll_interval_s=poll_interval,
+        include_values_removal_polls=removal_polls,
+        include_values_request_timeout_s=request_timeout,
+        include_values_startup_timeout_s=startup_timeout,
+        include_values_auth_header_name=header_name,
+        include_values_auth_token=token,
+    )
+
+
 def _load_sort_by() -> tuple[str, ...] | None:
     """Parse MILLPOND_SORT_BY into a tuple of column names.
 
@@ -417,6 +518,7 @@ def load() -> Config:
         filter_keep_field=filter_keep_field,
         filter_drop_field=filter_drop_field,
         filter_values=filter_values,
+        **_load_include_values_config(filter_keep_field, filter_values),
         sort_by=sort_by,
         typed_columns=typed_columns,
         kafka_config_overrides=kafka_overrides,