refactor(operator): name bounded entry points and validate Keyman admissions

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.

docs/adr/0082-bounded-keyman-backfill.md

@@ -11,13 +11,23 @@ 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.
 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.

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:

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__":

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__":