diff --git a/CHANGELOG.d/post-keymen-operator-bounds.md b/CHANGELOG.d/post-keymen-operator-bounds.md
new file mode 100644
index 000000000..c85bc7b1e
--- /dev/null
+++ b/CHANGELOG.d/post-keymen-operator-bounds.md
@@ -0,0 +1,7 @@
+# Post-Keyman operator bounds
+
+The bounded Post Keyman backfill now admits only exact batch limits from 1 through 100, so one `--all` invocation cannot become an effectively unbounded serial crawl. Direct programmatic calls receive the same admission checks as the CLI. A non-default `--limit` is rejected unless `--all` is selected, so a requested batch size cannot be silently accepted and then ignored by the default one-post path; `--post-id` and `--all` remain mutually exclusive.
+
+An explicit `--post-id` must now be the canonical lowercase, hyphenated UUID for `source_post.post_id`. Malformed values and alternate UUID spellings fail before gateway or database work instead of creating multiple textual representations of the same internal post identity; opaque source-system record keys remain separate evidence under ADR 0046.
+
+Per-post timeout admission now also fails closed for malformed direct-call numeric values whose magnitude cannot be represented by the runtime finite-number check, instead of leaking an `OverflowError` from validation. The admitted operator timeout is forwarded to both Keyman extraction and synchronous Vision requests through the contextual-orchestrator client boundary. This removes hidden shorter 180-second transport caps without introducing provider/model configuration in LineageWeave.
diff --git a/docs/adr/0082-bounded-keyman-backfill.md b/docs/adr/0082-bounded-keyman-backfill.md
index 397f18df7..b5f01ad58 100644
--- a/docs/adr/0082-bounded-keyman-backfill.md
+++ b/docs/adr/0082-bounded-keyman-backfill.md
@@ -11,6 +11,11 @@ come from the existing Keyman extraction and persistence projection. Imported
 real data has many posts without a `post_person_mention` row, so relying only
 on the per-post operator button leaves the author-group view mostly empty.
 
+ADR 0046 defines `source_post.post_id` as the internal UUID identity and keeps
+that UUID distinct from an opaque source-system record key. An operator selector
+therefore must not accept arbitrary opaque text or alternate UUID spellings as
+though they were the internal post identity.
+
 ## Decision
 
 Provide `scripts/backfill_post_keymen.py` as an operator-only, bounded runner.
@@ -18,6 +23,11 @@ It will:
 
 - select eligible, non-deleted, non-draft posts that have no existing
   `post_person_mention`, or one explicit `--post-id`;
+- admit an explicit `--post-id` only when it is the canonical lowercase,
+  hyphenated UUID text for the ADR 0046 internal post identity. Blank, padded,
+  malformed, uppercase, braced, and hyphenless aliases fail before gateway or
+  database work rather than being normalized into a different textual
+  identity;
 - normalize HTML, OOXML-derived text, embedded images, and image regions with
   the existing VISION normalization path before extraction;
 - pass source author, account, PU, sales-pool, customer, company, and project
@@ -27,9 +37,32 @@ It will:
   reconciliation;
 - carry `build_post_llm_metadata` and `use_llm_metadata` across all LLM/VISION
   calls for one post, yielding the same deterministic post session id;
-- default to one post and require explicit `--all --limit N` for a batch.
-- enforce a per-post timeout, returning a typed failure count instead of
-  allowing a provider workflow to hold an operator process indefinitely.
+- default to one post and require explicit `--all --limit N` for a batch;
+- admit a batch limit only as an exact integer in the inclusive `1..100`
+  range, applying the same check to direct programmatic runner calls before
+  gateway or database work. The upper bound keeps one invocation genuinely
+  bounded even when a caller bypasses the CLI; larger work is split into
+  repeated observable invocations instead of turning one process into an
+  effectively unbounded serial crawl;
+- reject a non-default `--limit` unless batch mode is explicitly selected with
+  `--all`. The same cross-field admission applies to direct programmatic calls,
+  so a requested limit cannot be silently ignored by falling back to the
+  default one-post mode. `--post-id` and `--all` remain mutually exclusive;
+- require the programmatic batch-mode selector to be an exact boolean before
+  using its truth value, so strings or integer-like transport values cannot
+  silently switch a direct call into or out of batch mode;
+- admit the per-post administrative timeout only when validation itself is
+  total: malformed direct-call values, including an integer too large for the
+  runtime finite-number check, fail closed instead of escaping admission with
+  an arithmetic exception before the operator can return its normal validation
+  error;
+- enforce one admitted per-post timeout across the operator and its Keyman and
+  Vision contextual-orchestrator transports. Neither synchronous Vision work
+  nor Keyman extraction may impose an unrelated shorter fixed timeout that can
+  terminate a valid long-running workflow before the operator's explicit
+  administrative budget;
+- return a typed timeout failure count instead of allowing a provider workflow
+  to hold an operator process indefinitely.
 
 Gateway credentials are read from runtime-injected environment variables. The
 script never reads or copies `~/.env`, and it is not exposed as a buyer HTTP
@@ -38,11 +71,24 @@ route. No analysis-run registry tables are modified.
 ## Consequences
 
 - Keyman coverage can be increased incrementally with a bounded cost and
-  auditable operator output.
+  auditable operator output. One invocation processes at most 100 posts; larger
+  backfills require repeated invocations whose result summaries remain
+  independently attributable.
+- An explicit non-default batch size is either honored under `--all` or rejected
+  before external work; it is never accepted and then silently collapsed to a
+  one-post execution.
+- Explicit reruns use one stable textual form for the internal UUID in operator
+  logs and LLM metadata; source-system record keys remain separate ADR 0046
+  evidence and are never accepted as `--post-id`.
 - Empty extraction remains a real empty result; the script does not create a
   placeholder person or retry indefinitely through an implicit attempt table.
 - Re-running a selected post is idempotent through `ingest_post_keymen`'s
   replacement semantics, while the default selector may revisit an empty
   extraction because no evidence row exists.
-- A provider workflow that exceeds the timeout is recorded as unavailable for
-  that attempt; it is not converted into an empty Keyman result.
+- A provider workflow that exceeds the operator-selected timeout is recorded as
+  unavailable for that attempt; it is not converted into an empty Keyman
+  result, and unrelated client-local fixed timeouts do not pre-empt that
+  budget.
+- Programmatic timeout admission remains fail-closed even for numeric values
+  whose magnitude cannot be represented by the runtime finite-number helper;
+  validation does not leak an `OverflowError` as an alternate control path.
diff --git a/lineageweave/image_content.py b/lineageweave/image_content.py
index 9d0282e23..c28fcb8a7 100644
--- a/lineageweave/image_content.py
+++ b/lineageweave/image_content.py
@@ -381,15 +381,23 @@ def locate_regions(self, image_bytes: bytes, mime_type: str) -> tuple[ImageRegio
         return tuple(accepted)
 
 
-def orchestrator_vision_client(base_url: str, api_key: str, model: str | None = None) -> ImageContentClient:
+def orchestrator_vision_client(
+    base_url: str,
+    api_key: str,
+    model: str | None = None,
+    *,
+    timeout: float = 180.0,
+) -> ImageContentClient:
     """Build a vision client against the same orchestrator root other channels use.
 
     Other clients POST ``{base_url}/v1/chat/completions``;
     :class:`OpenAiCompatibleVisionClient` POSTs ``{base_url}/chat/completions``,
     so this appends ``/v1`` unless already present. An ``http://`` orchestrator
     (local docker) is allowed because the other channels already talk to the
-    same URL. A construct-time error degrades to the unavailable null rather
-    than crashing the request that asked for a description.
+    same URL. The optional timeout lets a bounded caller share its admitted
+    transport budget with synchronous Vision requests. A construct-time error
+    degrades to the unavailable null rather than crashing the request that
+    asked for a description.
     """
     if not (base_url and api_key):
         return NullImageContentClient()
@@ -402,6 +410,7 @@ def orchestrator_vision_client(base_url: str, api_key: str, model: str | None =
             base_url=vision_base,
             api_key=api_key,
             model=model,
+            timeout=timeout,
             allow_insecure_http=parsed.scheme == "http",
         )
     except ValueError:
diff --git a/scripts/backfill_post_keymen.py b/scripts/backfill_post_keymen.py
index feec3e797..ba4541f83 100644
--- a/scripts/backfill_post_keymen.py
+++ b/scripts/backfill_post_keymen.py
@@ -10,6 +10,7 @@
 import argparse
 import asyncio
 import json
+import math
 import os
 import sys
 from collections import Counter
@@ -48,6 +49,55 @@ def _orchestrator_config() -> tuple[str, str]:
     return base_url, api_key
 
 
+def _post_timeout_is_valid(post_timeout: object) -> bool:
+    """Return whether an operator timeout is a finite, strictly positive number."""
+    if type(post_timeout) not in (int, float) or post_timeout <= 0:
+        return False
+    try:
+        return math.isfinite(post_timeout)
+    except OverflowError:
+        return False
+
+
+def _post_limit_is_valid(post_limit: object) -> bool:
+    """Return whether a batch limit is an exact integer in the bounded 1..100 range."""
+    return type(post_limit) is int and 1 <= post_limit <= 100
+
+
+def _post_all_is_valid(post_all: object) -> bool:
+    """Return whether the batch-mode selector is an exact boolean."""
+    return type(post_all) is bool
+
+
+def _post_id_is_valid(post_id: object) -> bool:
+    """Return whether an optional post identity is one canonical UUID string."""
+    if post_id is None:
+        return True
+    if type(post_id) is not str or not post_id or post_id != post_id.strip():
+        return False
+    from uuid import UUID
+
+    try:
+        return str(UUID(post_id)) == post_id
+    except ValueError:
+        return False
+
+
+def _post_selection_is_valid(post_all: object, post_id: object, post_limit: object) -> bool:
+    """Return whether selector and limit fields form one unambiguous bounded request."""
+    if (
+        not _post_all_is_valid(post_all)
+        or not _post_id_is_valid(post_id)
+        or not _post_limit_is_valid(post_limit)
+    ):
+        return False
+    if post_all and post_id is not None:
+        return False
+    if not post_all and post_limit != 1:
+        return False
+    return True
+
+
 async def _select_posts(
     conn: asyncpg.Connection, *, limit: int, post_id: str | None
 ) -> list[asyncpg.Record]:
@@ -159,31 +209,58 @@ async def _select_posts(
     )
 
 
-async def _run(args: argparse.Namespace) -> dict[str, object]:
-    if args.post_id and args.all:
-        raise ValueError("--post-id and --all cannot be combined")
+async def _run_post_keymen_backfill(
+    backfill_arguments: argparse.Namespace,
+) -> dict[str, object]:
+    """Execute one bounded post-Keyman backfill operation."""
+    if not _post_timeout_is_valid(backfill_arguments.post_timeout):
+        raise ValueError("--post-timeout must be finite and positive")
+    if not _post_limit_is_valid(backfill_arguments.limit):
+        raise ValueError("--limit must be an integer between 1 and 100")
+    if not _post_all_is_valid(backfill_arguments.all):
+        raise ValueError("--all must be a boolean selector")
+    if not _post_id_is_valid(backfill_arguments.post_id):
+        raise ValueError("--post-id must be a canonical UUID")
+    if not _post_selection_is_valid(
+        backfill_arguments.all,
+        backfill_arguments.post_id,
+        backfill_arguments.limit,
+    ):
+        raise ValueError("non-default --limit requires --all; --post-id and --all cannot be combined")
     base_url, api_key = _orchestrator_config()
     settings = load_settings()
     keyman_client = ContextualOrchestratorKeymanExtractionClient(
-        base_url=base_url, api_key=api_key, timeout=180.0
+        base_url=base_url,
+        api_key=api_key,
+        timeout=backfill_arguments.post_timeout,
+    )
+    vision_client = orchestrator_vision_client(
+        base_url,
+        api_key,
+        timeout=backfill_arguments.post_timeout,
     )
-    vision_client = orchestrator_vision_client(base_url, api_key)
     resolution_client = _organization_name_resolution_client()
     verification_client = _relation_verification_client()
     hierarchy_client = _corporate_hierarchy_inference_client()
-    limit = 1 if args.post_id or not args.all else args.limit
+    limit = (
+        1
+        if backfill_arguments.post_id or not backfill_arguments.all
+        else backfill_arguments.limit
+    )
 
     pool = await asyncpg.create_pool(settings.database_url, min_size=1, max_size=1)
     try:
         async with pool.acquire() as conn:
-            rows = await _select_posts(conn, limit=limit, post_id=args.post_id)
+            rows = await _select_posts(
+                conn, limit=limit, post_id=backfill_arguments.post_id
+            )
             failures: Counter[str] = Counter()
             processed = 0
             mention_count = 0
             for row in rows:
                 post_id = str(row["post_id"])
                 try:
-                    async with asyncio.timeout(args.post_timeout):
+                    async with asyncio.timeout(backfill_arguments.post_timeout):
                         with use_llm_metadata(build_post_llm_metadata(post_id, dict(row))):
                             normalized = normalize_post_body(row["post_body"] or "", vision_client)
                             context_hints = await _load_post_semantic_hints(conn, post_id)
@@ -218,21 +295,42 @@ async def _run(args: argparse.Namespace) -> dict[str, object]:
 def main() -> None:
     parser = argparse.ArgumentParser(description=__doc__)
     selector = parser.add_mutually_exclusive_group()
-    selector.add_argument("--post-id", help="Re-extract one eligible post")
+    selector.add_argument("--post-id", help="Re-extract one eligible post UUID")
     selector.add_argument("--all", action="store_true", help="Process the explicit --limit batch")
-    parser.add_argument("--limit", type=int, default=1, help="Maximum posts for --all (default: 1)")
+    parser.add_argument(
+        "--limit",
+        type=int,
+        default=1,
+        help="Maximum posts for --all, 1..100 (default: 1)",
+    )
     parser.add_argument(
         "--post-timeout",
         type=float,
         default=240.0,
         help="Maximum seconds per post including provider calls (default: 240)",
     )
-    args = parser.parse_args()
-    if args.limit < 1:
-        parser.error("--limit must be positive")
-    if args.post_timeout <= 0:
-        parser.error("--post-timeout must be positive")
-    print(json.dumps(asyncio.run(_run(args)), ensure_ascii=False, sort_keys=True))
+    backfill_arguments = parser.parse_args()
+    if not _post_limit_is_valid(backfill_arguments.limit):
+        parser.error("--limit must be an integer between 1 and 100")
+    if not _post_all_is_valid(backfill_arguments.all):
+        parser.error("--all must be a boolean selector")
+    if not _post_timeout_is_valid(backfill_arguments.post_timeout):
+        parser.error("--post-timeout must be finite and positive")
+    if not _post_id_is_valid(backfill_arguments.post_id):
+        parser.error("--post-id must be a canonical UUID")
+    if not _post_selection_is_valid(
+        backfill_arguments.all,
+        backfill_arguments.post_id,
+        backfill_arguments.limit,
+    ):
+        parser.error("non-default --limit requires --all; --post-id and --all cannot be combined")
+    print(
+        json.dumps(
+            asyncio.run(_run_post_keymen_backfill(backfill_arguments)),
+            ensure_ascii=False,
+            sort_keys=True,
+        )
+    )
 
 
 if __name__ == "__main__":
diff --git a/scripts/backfill_thread_group_keys.py b/scripts/backfill_thread_group_keys.py
index 8352c1ada..d50f4853e 100644
--- a/scripts/backfill_thread_group_keys.py
+++ b/scripts/backfill_thread_group_keys.py
@@ -132,14 +132,18 @@ def __init__(self, project_evidence: int, cleared: int) -> None:
         self.cleared = cleared
 
 
-async def _run(args: argparse.Namespace) -> dict[str, object]:
-    """Execute one pooled backfill and convert dry-run rollback into counts."""
+async def _run_thread_group_key_backfill(
+    backfill_arguments: argparse.Namespace,
+) -> dict[str, object]:
+    """Execute one pooled thread-group-key backfill and report aggregate counts."""
     settings = load_settings()
     pool = await asyncpg.create_pool(settings.database_url, min_size=1, max_size=1)
     try:
         async with pool.acquire() as conn:
             try:
-                counts = await backfill_thread_group_keys(conn, dry_run=args.dry_run)
+                counts = await backfill_thread_group_keys(
+                    conn, dry_run=backfill_arguments.dry_run
+                )
                 return {**counts, "dry_run": False}
             except _RollbackDryRun as rolled_back:
                 return {
@@ -159,8 +163,14 @@ def main() -> None:
         action="store_true",
         help="Report counts without writing (rolls back the transaction)",
     )
-    args = parser.parse_args()
-    print(json.dumps(asyncio.run(_run(args)), ensure_ascii=False, sort_keys=True))
+    backfill_arguments = parser.parse_args()
+    print(
+        json.dumps(
+            asyncio.run(_run_thread_group_key_backfill(backfill_arguments)),
+            ensure_ascii=False,
+            sort_keys=True,
+        )
+    )
 
 
 if __name__ == "__main__":
diff --git a/scripts/estimate_channel_weights.py b/scripts/estimate_channel_weights.py
index b8ab9534f..20997ea1f 100644
--- a/scripts/estimate_channel_weights.py
+++ b/scripts/estimate_channel_weights.py
@@ -173,7 +173,10 @@ async def persist_estimate(
     return estimation_run_id
 
 
-async def _run(args: argparse.Namespace) -> dict[str, object]:
+async def _run_channel_weight_estimation(
+    estimation_arguments: argparse.Namespace,
+) -> dict[str, object]:
+    """Execute one bounded lineage channel-weight estimation run."""
     settings = load_settings()
     # Short-lived fetch connection; nothing stays open while fitting.
     conn = await asyncpg.connect(settings.database_url)
@@ -184,7 +187,7 @@ async def _run(args: argparse.Namespace) -> dict[str, object]:
             "secondary_grouping_key "
             f"from source_post where {SOURCE_POST_ELIGIBILITY_SQL.format(alias='source_post')} "
             "order by created_at, post_id limit $1::bigint",
-            args.post_limit,
+            estimation_arguments.post_limit,
         )
     finally:
         await conn.close()
@@ -206,7 +209,7 @@ async def _run(args: argparse.Namespace) -> dict[str, object]:
             "named condition"
         )
     estimation_run_id = None
-    if not args.dry_run:
+    if not estimation_arguments.dry_run:
         conn = await asyncpg.connect(settings.database_url)
         try:
             estimation_run_id = await persist_estimate(
@@ -227,7 +230,7 @@ async def _run(args: argparse.Namespace) -> dict[str, object]:
         "estimation_run_id": estimation_run_id,
         "source_snapshot_sha256": snapshot_sha256,
         "knowledge_cutoff": knowledge_cutoff.isoformat(),
-        "persisted": not args.dry_run,
+        "persisted": not estimation_arguments.dry_run,
         "activation": (
             "blocked_until_anchor_authorized (ADR 0200 point 3): the "
             "product loader refuses every anchor method today, so these "
@@ -250,10 +253,16 @@ def main() -> None:
         action="store_true",
         help="Estimate and report, but persist nothing",
     )
-    args = parser.parse_args()
-    if args.post_limit < 1:
+    estimation_arguments = parser.parse_args()
+    if estimation_arguments.post_limit < 1:
         parser.error("--post-limit must be positive")
-    print(json.dumps(asyncio.run(_run(args)), ensure_ascii=False, sort_keys=True))
+    print(
+        json.dumps(
+            asyncio.run(_run_channel_weight_estimation(estimation_arguments)),
+            ensure_ascii=False,
+            sort_keys=True,
+        )
+    )
 
 
 if __name__ == "__main__":
diff --git a/tests/test_channel_weight_estimation_script_naming_contract.py b/tests/test_channel_weight_estimation_script_naming_contract.py
new file mode 100644
index 000000000..09804cff0
--- /dev/null
+++ b/tests/test_channel_weight_estimation_script_naming_contract.py
@@ -0,0 +1,34 @@
+"""Naming contract for the channel-weight estimation operator script."""
+
+from __future__ import annotations
+
+import ast
+from pathlib import Path
+
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
+ESTIMATION_SCRIPT = REPOSITORY_ROOT / "scripts" / "estimate_channel_weights.py"
+
+
+def test_channel_weight_estimation_uses_bounded_operation_name() -> None:
+    """Require the async operator entry point to state the estimation it runs."""
+    syntax_tree = ast.parse(ESTIMATION_SCRIPT.read_text(encoding="utf-8"))
+    async_functions = {
+        node.name: node
+        for node in ast.walk(syntax_tree)
+        if isinstance(node, ast.AsyncFunctionDef)
+    }
+
+    assert "_run_channel_weight_estimation" in async_functions
+    assert "_run" not in async_functions
+    operation = async_functions["_run_channel_weight_estimation"]
+    assert operation.args.args[0].arg == "estimation_arguments"
+
+
+def test_channel_weight_estimation_main_uses_semantic_parsed_arguments() -> None:
+    """Keep parsed command-line arguments explicit at the script boundary."""
+    source_text = ESTIMATION_SCRIPT.read_text(encoding="utf-8")
+
+    assert "estimation_arguments = parser.parse_args()" in source_text
+    assert "asyncio.run(_run_channel_weight_estimation(estimation_arguments))" in source_text
+    assert "args = parser.parse_args()" not in source_text
diff --git a/tests/test_image_content_gateway.py b/tests/test_image_content_gateway.py
index 9a1728a17..e4177e444 100644
--- a/tests/test_image_content_gateway.py
+++ b/tests/test_image_content_gateway.py
@@ -52,6 +52,34 @@ def post_json(url, payload, *, headers, timeout):
     assert normalized.convert("RGB").getpixel((0, 0)) == (255, 255, 255)
 
 
+def test_vision_factory_forwards_bounded_transport_timeout(monkeypatch) -> None:
+    """Carry a caller-selected budget through the factory to synchronous HTTP work."""
+    captured: dict[str, object] = {}
+
+    def post_json(url, payload, *, headers, timeout):
+        captured["timeout"] = timeout
+        return {
+            "choices": [
+                {
+                    "message": {
+                        "content": "TEXT: NONE\nCAPTION: synthetic timeout probe\nTAGS: probe"
+                    }
+                }
+            ]
+        }
+
+    monkeypatch.setattr(image_content, "post_json", post_json)
+    client = image_content.orchestrator_vision_client(
+        "http://orchestrator",
+        "gateway-key",
+        timeout=420.0,
+    )
+
+    client.describe(_transparent_png(), "image/png")
+
+    assert captured["timeout"] == 420.0
+
+
 def test_vision_factory_fails_closed_for_unsupported_url_scheme() -> None:
     client = image_content.orchestrator_vision_client("ftp://orchestrator", "gateway-key", "vision-model")
     assert isinstance(client, image_content.NullImageContentClient)
diff --git a/tests/test_post_keymen_backfill_batch_selector_contract.py b/tests/test_post_keymen_backfill_batch_selector_contract.py
new file mode 100644
index 000000000..5550eb574
--- /dev/null
+++ b/tests/test_post_keymen_backfill_batch_selector_contract.py
@@ -0,0 +1,124 @@
+"""Safety contract for the post-Keyman operator batch selector."""
+
+from __future__ import annotations
+
+import ast
+from pathlib import Path
+
+import pytest
+
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
+BACKFILL_SCRIPT = REPOSITORY_ROOT / "scripts" / "backfill_post_keymen.py"
+
+
+def _load_batch_selector_validator():
+    """Load only the pure batch-selector validator without operator dependencies."""
+    syntax_tree = ast.parse(BACKFILL_SCRIPT.read_text(encoding="utf-8"))
+    validator = next(
+        (
+            node
+            for node in syntax_tree.body
+            if isinstance(node, ast.FunctionDef)
+            and node.name == "_post_all_is_valid"
+        ),
+        None,
+    )
+    assert validator is not None, "operator must expose a pure batch-selector admission check"
+    namespace: dict[str, object] = {}
+    module = ast.fix_missing_locations(ast.Module(body=[validator], type_ignores=[]))
+    exec(compile(module, str(BACKFILL_SCRIPT), "exec"), namespace)
+    return namespace["_post_all_is_valid"]
+
+
+def _load_selection_validator():
+    """Load the pure cross-field selector contract without importing operator dependencies."""
+    syntax_tree = ast.parse(BACKFILL_SCRIPT.read_text(encoding="utf-8"))
+    function_names = {
+        "_post_limit_is_valid",
+        "_post_all_is_valid",
+        "_post_id_is_valid",
+        "_post_selection_is_valid",
+    }
+    functions = [
+        node
+        for node in syntax_tree.body
+        if isinstance(node, ast.FunctionDef) and node.name in function_names
+    ]
+    assert any(
+        node.name == "_post_selection_is_valid" for node in functions
+    ), "operator must validate selector/limit combinations as one admission contract"
+    namespace: dict[str, object] = {}
+    module = ast.fix_missing_locations(ast.Module(body=functions, type_ignores=[]))
+    exec(compile(module, str(BACKFILL_SCRIPT), "exec"), namespace)
+    return namespace["_post_selection_is_valid"]
+
+
+def test_post_keymen_backfill_accepts_boolean_batch_selector() -> None:
+    validator = _load_batch_selector_validator()
+
+    assert validator(False) is True
+    assert validator(True) is True
+
+
+@pytest.mark.parametrize("invalid_selector", [0, 1, "false", "true", None, [], {}])
+def test_post_keymen_backfill_rejects_non_boolean_batch_selector(
+    invalid_selector: object,
+) -> None:
+    """Reject transport-shaped values whose truthiness could select batch mode."""
+    validator = _load_batch_selector_validator()
+
+    assert validator(invalid_selector) is False
+
+
+def test_post_keymen_backfill_rejects_ignored_non_default_limits() -> None:
+    """An explicit non-default limit must not be silently ignored outside batch mode."""
+    validator = _load_selection_validator()
+    post_id = "00000000-0000-0000-0000-000000000001"
+
+    assert validator(False, None, 1) is True
+    assert validator(False, post_id, 1) is True
+    assert validator(True, None, 1) is True
+    assert validator(True, None, 100) is True
+
+    assert validator(False, None, 2) is False
+    assert validator(False, post_id, 2) is False
+    assert validator(True, post_id, 1) is False
+
+
+def test_programmatic_runner_revalidates_batch_selector_before_external_work() -> None:
+    """Keep direct callers behind the same selector-mode admission boundary."""
+    syntax_tree = ast.parse(BACKFILL_SCRIPT.read_text(encoding="utf-8"))
+    runner = next(
+        node
+        for node in syntax_tree.body
+        if isinstance(node, ast.AsyncFunctionDef)
+        and node.name == "_run_post_keymen_backfill"
+    )
+
+    calls = [
+        node
+        for node in ast.walk(runner)
+        if isinstance(node, ast.Call)
+        and isinstance(node.func, ast.Name)
+        and node.func.id == "_post_all_is_valid"
+    ]
+    assert calls, "programmatic runner must revalidate batch selector before external work"
+
+
+def test_programmatic_runner_revalidates_selector_combination_before_external_work() -> None:
+    """Keep ignored-limit ambiguity out of the direct-call operator boundary."""
+    syntax_tree = ast.parse(BACKFILL_SCRIPT.read_text(encoding="utf-8"))
+    runner = next(
+        node
+        for node in syntax_tree.body
+        if isinstance(node, ast.AsyncFunctionDef)
+        and node.name == "_run_post_keymen_backfill"
+    )
+
+    assert any(
+        isinstance(node, ast.Call)
+        and isinstance(node.func, ast.Name)
+        and node.func.id == "_post_selection_is_valid"
+        for node in ast.walk(runner)
+    )
diff --git a/tests/test_post_keymen_backfill_limit_contract.py b/tests/test_post_keymen_backfill_limit_contract.py
new file mode 100644
index 000000000..40f45c489
--- /dev/null
+++ b/tests/test_post_keymen_backfill_limit_contract.py
@@ -0,0 +1,87 @@
+"""Safety contract for the bounded post-Keyman operator batch limit."""
+
+from __future__ import annotations
+
+import ast
+from pathlib import Path
+
+import pytest
+
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
+BACKFILL_SCRIPT = REPOSITORY_ROOT / "scripts" / "backfill_post_keymen.py"
+
+
+def _load_limit_validator():
+    """Load only the pure limit validator without importing operator dependencies."""
+    syntax_tree = ast.parse(BACKFILL_SCRIPT.read_text(encoding="utf-8"))
+    validator = next(
+        (
+            node
+            for node in syntax_tree.body
+            if isinstance(node, ast.FunctionDef)
+            and node.name == "_post_limit_is_valid"
+        ),
+        None,
+    )
+    assert validator is not None, "operator must expose a pure batch-limit admission check"
+    namespace: dict[str, object] = {}
+    module = ast.fix_missing_locations(ast.Module(body=[validator], type_ignores=[]))
+    exec(compile(module, str(BACKFILL_SCRIPT), "exec"), namespace)
+    return namespace["_post_limit_is_valid"]
+
+
+@pytest.mark.parametrize(
+    "invalid_limit",
+    [0, -1, True, False, 1.0, "1", None, 101, 1_000_000],
+)
+def test_post_keymen_backfill_rejects_malformed_or_unbounded_batch_limit(
+    invalid_limit: object,
+) -> None:
+    """Reject malformed or operationally unbounded direct-call batch limits."""
+    validator = _load_limit_validator()
+
+    assert validator(invalid_limit) is False
+
+
+def test_post_keymen_backfill_accepts_positive_integer_batch_limit() -> None:
+    validator = _load_limit_validator()
+
+    assert validator(1) is True
+    assert validator(100) is True
+
+
+def test_main_routes_limit_through_the_admission_check() -> None:
+    syntax_tree = ast.parse(BACKFILL_SCRIPT.read_text(encoding="utf-8"))
+    main_function = next(
+        node
+        for node in syntax_tree.body
+        if isinstance(node, ast.FunctionDef) and node.name == "main"
+    )
+
+    assert any(
+        isinstance(node, ast.Call)
+        and isinstance(node.func, ast.Name)
+        and node.func.id == "_post_limit_is_valid"
+        for node in ast.walk(main_function)
+    )
+
+
+def test_programmatic_runner_revalidates_limit_before_external_work() -> None:
+    """Keep direct callers behind the same bounded batch admission contract."""
+    syntax_tree = ast.parse(BACKFILL_SCRIPT.read_text(encoding="utf-8"))
+    runner = next(
+        node
+        for node in ast.walk(syntax_tree)
+        if isinstance(node, ast.AsyncFunctionDef)
+        and node.name == "_run_post_keymen_backfill"
+    )
+
+    calls = [
+        node
+        for node in ast.walk(runner)
+        if isinstance(node, ast.Call)
+        and isinstance(node.func, ast.Name)
+        and node.func.id == "_post_limit_is_valid"
+    ]
+    assert calls, "programmatic runner must revalidate limit before provider/database work"
diff --git a/tests/test_post_keymen_backfill_naming_contract.py b/tests/test_post_keymen_backfill_naming_contract.py
new file mode 100644
index 000000000..856ecfd11
--- /dev/null
+++ b/tests/test_post_keymen_backfill_naming_contract.py
@@ -0,0 +1,34 @@
+"""Naming contract for the bounded post-Keyman operator backfill."""
+
+from __future__ import annotations
+
+import ast
+from pathlib import Path
+
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
+BACKFILL_SCRIPT = REPOSITORY_ROOT / "scripts" / "backfill_post_keymen.py"
+
+
+def test_post_keymen_backfill_uses_bounded_operation_name() -> None:
+    """Require the operator coroutine to name the operation it owns."""
+    syntax_tree = ast.parse(BACKFILL_SCRIPT.read_text(encoding="utf-8"))
+    async_functions = {
+        node.name: node
+        for node in ast.walk(syntax_tree)
+        if isinstance(node, ast.AsyncFunctionDef)
+    }
+
+    assert "_run_post_keymen_backfill" in async_functions
+    assert "_run" not in async_functions
+    operation = async_functions["_run_post_keymen_backfill"]
+    assert operation.args.args[0].arg == "backfill_arguments"
+
+
+def test_post_keymen_backfill_main_uses_semantic_parsed_arguments() -> None:
+    """Keep parsed command-line arguments explicit at the script boundary."""
+    source_text = BACKFILL_SCRIPT.read_text(encoding="utf-8")
+
+    assert "backfill_arguments = parser.parse_args()" in source_text
+    assert "asyncio.run(_run_post_keymen_backfill(backfill_arguments))" in source_text
+    assert "args = parser.parse_args()" not in source_text
diff --git a/tests/test_post_keymen_backfill_post_id_contract.py b/tests/test_post_keymen_backfill_post_id_contract.py
new file mode 100644
index 000000000..8bcf041bc
--- /dev/null
+++ b/tests/test_post_keymen_backfill_post_id_contract.py
@@ -0,0 +1,97 @@
+"""Identity admission contract for the bounded post-Keyman operator selector."""
+
+from __future__ import annotations
+
+import ast
+from pathlib import Path
+
+import pytest
+
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
+BACKFILL_SCRIPT = REPOSITORY_ROOT / "scripts" / "backfill_post_keymen.py"
+VALID_POST_ID = "123e4567-e89b-42d3-a456-426614174000"
+
+
+def _load_post_id_validator():
+    """Load only the pure post-id validator without importing operator dependencies."""
+    syntax_tree = ast.parse(BACKFILL_SCRIPT.read_text(encoding="utf-8"))
+    validator = next(
+        (
+            node
+            for node in syntax_tree.body
+            if isinstance(node, ast.FunctionDef)
+            and node.name == "_post_id_is_valid"
+        ),
+        None,
+    )
+    assert validator is not None, "operator must expose a pure post-id admission check"
+    namespace: dict[str, object] = {"__builtins__": __builtins__}
+    module = ast.fix_missing_locations(ast.Module(body=[validator], type_ignores=[]))
+    exec(compile(module, str(BACKFILL_SCRIPT), "exec"), namespace)
+    return namespace["_post_id_is_valid"]
+
+
+@pytest.mark.parametrize(
+    "invalid_post_id",
+    ["", " ", "\t", f"{VALID_POST_ID} ", f" {VALID_POST_ID}", f"{VALID_POST_ID}\n"],
+)
+def test_post_keymen_backfill_rejects_blank_or_padded_explicit_post_id(
+    invalid_post_id: str,
+) -> None:
+    validator = _load_post_id_validator()
+
+    assert validator(invalid_post_id) is False
+
+
+@pytest.mark.parametrize(
+    "invalid_post_id",
+    [
+        "post-123",
+        "123E4567-E89B-42D3-A456-426614174000",
+        "{123e4567-e89b-42d3-a456-426614174000}",
+        "123e4567e89b42d3a456426614174000",
+    ],
+)
+def test_post_keymen_backfill_rejects_noncanonical_uuid_aliases(
+    invalid_post_id: str,
+) -> None:
+    """Keep the internal source-post UUID one exact identity across logs and metadata."""
+    validator = _load_post_id_validator()
+
+    assert validator(invalid_post_id) is False
+
+
+@pytest.mark.parametrize(
+    "invalid_post_id", [7, True, [VALID_POST_ID], {"id": VALID_POST_ID}]
+)
+def test_post_keymen_backfill_rejects_non_string_explicit_post_id(
+    invalid_post_id: object,
+) -> None:
+    """Do not let truthy transport values reach string operations or identity lookup."""
+    validator = _load_post_id_validator()
+
+    assert validator(invalid_post_id) is False
+
+
+def test_post_keymen_backfill_accepts_absent_or_exact_post_id() -> None:
+    validator = _load_post_id_validator()
+
+    assert validator(None) is True
+    assert validator(VALID_POST_ID) is True
+
+
+def test_main_routes_post_id_through_the_admission_check() -> None:
+    syntax_tree = ast.parse(BACKFILL_SCRIPT.read_text(encoding="utf-8"))
+    main_function = next(
+        node
+        for node in syntax_tree.body
+        if isinstance(node, ast.FunctionDef) and node.name == "main"
+    )
+
+    assert any(
+        isinstance(node, ast.Call)
+        and isinstance(node.func, ast.Name)
+        and node.func.id == "_post_id_is_valid"
+        for node in ast.walk(main_function)
+    )
diff --git a/tests/test_post_keymen_backfill_timeout_contract.py b/tests/test_post_keymen_backfill_timeout_contract.py
new file mode 100644
index 000000000..f8e492627
--- /dev/null
+++ b/tests/test_post_keymen_backfill_timeout_contract.py
@@ -0,0 +1,154 @@
+"""Safety contract for the bounded post-Keyman operator timeout."""
+
+from __future__ import annotations
+
+import ast
+import math
+from pathlib import Path
+
+import pytest
+
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
+BACKFILL_SCRIPT = REPOSITORY_ROOT / "scripts" / "backfill_post_keymen.py"
+
+
+def _load_timeout_validator():
+    """Load only the pure timeout validator without importing operator dependencies."""
+    syntax_tree = ast.parse(BACKFILL_SCRIPT.read_text(encoding="utf-8"))
+    validator = next(
+        (
+            node
+            for node in syntax_tree.body
+            if isinstance(node, ast.FunctionDef)
+            and node.name == "_post_timeout_is_valid"
+        ),
+        None,
+    )
+    assert validator is not None, "operator must expose a pure timeout admission check"
+    namespace = {"math": math}
+    module = ast.fix_missing_locations(ast.Module(body=[validator], type_ignores=[]))
+    exec(compile(module, str(BACKFILL_SCRIPT), "exec"), namespace)
+    return namespace["_post_timeout_is_valid"]
+
+
+@pytest.mark.parametrize("invalid_timeout", [0.0, -1.0, math.nan, math.inf, -math.inf])
+def test_post_keymen_backfill_rejects_non_positive_or_non_finite_timeout(
+    invalid_timeout: float,
+) -> None:
+    validator = _load_timeout_validator()
+
+    assert validator(invalid_timeout) is False
+
+
+@pytest.mark.parametrize("invalid_timeout", [True, False, "240", None])
+def test_post_keymen_backfill_rejects_non_numeric_or_boolean_timeout(
+    invalid_timeout: object,
+) -> None:
+    """Reject malformed programmatic values instead of relying on argparse coercion."""
+    validator = _load_timeout_validator()
+
+    assert validator(invalid_timeout) is False
+
+
+def test_post_keymen_backfill_rejects_unrepresentable_integer_timeout() -> None:
+    """Fail closed instead of raising while checking an enormous direct-call integer."""
+    validator = _load_timeout_validator()
+
+    assert validator(10**10000) is False
+
+
+def test_post_keymen_backfill_accepts_positive_finite_timeout() -> None:
+    validator = _load_timeout_validator()
+
+    assert validator(0.001) is True
+    assert validator(240.0) is True
+
+
+def test_main_routes_timeout_through_the_admission_check() -> None:
+    syntax_tree = ast.parse(BACKFILL_SCRIPT.read_text(encoding="utf-8"))
+    main_function = next(
+        node
+        for node in syntax_tree.body
+        if isinstance(node, ast.FunctionDef) and node.name == "main"
+    )
+
+    assert any(
+        isinstance(node, ast.Call)
+        and isinstance(node.func, ast.Name)
+        and node.func.id == "_post_timeout_is_valid"
+        for node in ast.walk(main_function)
+    )
+
+
+def test_programmatic_runner_revalidates_timeout_before_external_work() -> None:
+    """Keep direct callers behind the same bounded timeout admission contract."""
+    syntax_tree = ast.parse(BACKFILL_SCRIPT.read_text(encoding="utf-8"))
+    runner = next(
+        node
+        for node in syntax_tree.body
+        if isinstance(node, ast.AsyncFunctionDef)
+        and node.name == "_run_post_keymen_backfill"
+    )
+
+    calls = [
+        node
+        for node in ast.walk(runner)
+        if isinstance(node, ast.Call)
+        and isinstance(node.func, ast.Name)
+        and node.func.id == "_post_timeout_is_valid"
+    ]
+    assert calls, "programmatic runner must revalidate timeout before provider/database work"
+
+
+def _assert_timeout_keyword_uses_operator_budget(client_call: ast.Call) -> None:
+    """Require one transport call to consume the already-admitted operator budget."""
+    timeout_keyword = next(
+        (keyword for keyword in client_call.keywords if keyword.arg == "timeout"),
+        None,
+    )
+    assert timeout_keyword is not None, "transport must receive the admitted operator timeout"
+    assert isinstance(timeout_keyword.value, ast.Attribute)
+    assert isinstance(timeout_keyword.value.value, ast.Name)
+    assert timeout_keyword.value.value.id == "backfill_arguments"
+    assert timeout_keyword.value.attr == "post_timeout"
+
+
+def test_keyman_transport_uses_the_admitted_operator_timeout() -> None:
+    """Do not impose an unrelated shorter Keyman transport cap inside the batch budget."""
+    syntax_tree = ast.parse(BACKFILL_SCRIPT.read_text(encoding="utf-8"))
+    runner = next(
+        node
+        for node in syntax_tree.body
+        if isinstance(node, ast.AsyncFunctionDef)
+        and node.name == "_run_post_keymen_backfill"
+    )
+    client_call = next(
+        node
+        for node in ast.walk(runner)
+        if isinstance(node, ast.Call)
+        and isinstance(node.func, ast.Name)
+        and node.func.id == "ContextualOrchestratorKeymanExtractionClient"
+    )
+
+    _assert_timeout_keyword_uses_operator_budget(client_call)
+
+
+def test_vision_transport_uses_the_admitted_operator_timeout() -> None:
+    """Keep synchronous Vision work inside the same per-post administrative budget."""
+    syntax_tree = ast.parse(BACKFILL_SCRIPT.read_text(encoding="utf-8"))
+    runner = next(
+        node
+        for node in syntax_tree.body
+        if isinstance(node, ast.AsyncFunctionDef)
+        and node.name == "_run_post_keymen_backfill"
+    )
+    client_call = next(
+        node
+        for node in ast.walk(runner)
+        if isinstance(node, ast.Call)
+        and isinstance(node.func, ast.Name)
+        and node.func.id == "orchestrator_vision_client"
+    )
+
+    _assert_timeout_keyword_uses_operator_budget(client_call)
diff --git a/tests/test_thread_group_backfill_naming_contract.py b/tests/test_thread_group_backfill_naming_contract.py
new file mode 100644
index 000000000..8b88fcf75
--- /dev/null
+++ b/tests/test_thread_group_backfill_naming_contract.py
@@ -0,0 +1,34 @@
+"""Naming contract for the thread-group-key backfill operator script."""
+
+from __future__ import annotations
+
+import ast
+from pathlib import Path
+
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
+BACKFILL_SCRIPT = REPOSITORY_ROOT / "scripts" / "backfill_thread_group_keys.py"
+
+
+def test_thread_group_backfill_uses_bounded_operation_name() -> None:
+    """Require the pooled operator coroutine to name the backfill it executes."""
+    syntax_tree = ast.parse(BACKFILL_SCRIPT.read_text(encoding="utf-8"))
+    async_functions = {
+        node.name: node
+        for node in ast.walk(syntax_tree)
+        if isinstance(node, ast.AsyncFunctionDef)
+    }
+
+    assert "_run_thread_group_key_backfill" in async_functions
+    assert "_run" not in async_functions
+    operation = async_functions["_run_thread_group_key_backfill"]
+    assert operation.args.args[0].arg == "backfill_arguments"
+
+
+def test_thread_group_backfill_main_uses_semantic_parsed_arguments() -> None:
+    """Keep parsed command-line arguments explicit at the script boundary."""
+    source_text = BACKFILL_SCRIPT.read_text(encoding="utf-8")
+
+    assert "backfill_arguments = parser.parse_args()" in source_text
+    assert "asyncio.run(_run_thread_group_key_backfill(backfill_arguments))" in source_text
+    assert "args = parser.parse_args()" not in source_text