diff --git a/CHANGELOG.md b/CHANGELOG.md index 7a2724eb8..63e8e1ecf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -262,6 +262,40 @@ All notable changes to this project are documented here. Format follows ### Changed +- The synchronous post-content backfill now uses semantic command, database, + source-post, normalized-content, image-result, aggregate-result, and SQL + aliases while preserving CLI flags, JSON result keys, source selection, + persistence, transaction, and connection-close behavior. + +- The post-content queue backfill now uses semantic command, database, Valkey, + source-record, job-request, and aggregate-result identifiers while preserving + CLI flags, JSON result keys, selection SQL, transaction boundaries, and event + publication behavior. + +- The ADR 0250 occupational catalog synchronization command now uses semantic + catalog, database, payload, configuration, and result identifiers while + preserving the fixed O*NET URL, `--target-dsn`, release/result output, digest + validation, transactional UPSERT, and connection-close behavior. + +- The deterministic channel-weight estimator now uses semantic source-post, + candidate-window, database, estimate, and command identifiers while + preserving pair sampling, fitting, CLI, JSON, SQL, and persisted provenance + contracts. The fast-mlsirm `v0.9.1` consumer cutover remains isolated in + #967. + +- The queued LLM channel-weight estimator now uses semantic batch-submission, + estimation-run, pair-judgment, orchestrator, and command identifiers while + preserving provider payload fields, CLI and JSON contracts, SQL, transaction + boundaries, and incomplete-judgment behavior. + +- The bounded thread-group-key backfill now uses semantic command, database, + record, and count identifiers while preserving `--dry-run`, aggregate JSON, + SQL, transaction rollback, and persistence behavior. + +- The bounded post-Keyman operator now uses semantic package-owned command, + database, record, and result identifiers while preserving every CLI flag, + JSON result field, SQL statement, and persistence boundary. + - ADRs 0011 and 0065 now include APA 7th References for the dated W3C PROV-O and PROV-DM Recommendations (30 April 2013). Decisions are unchanged. diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index b5d31877b..c0beb0e7d 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -1,5 +1,78 @@ # Product & Technical Gap Baseline +> Synchronous post-content backfill naming overlay: 2026-09-07 KST. Protected +> `main` is `83eba56149eb802cd63642c507c324c9976ec78e`. The operator used +> generic package-owned command, database, record, normalized-content, image, +> result, and limit identifiers (`_parser`, `args`, `conn`, `row`, `item`, +> `result`, `limit`) plus generic SQL aliases. Action: translate the complete +> private caller surface to post-content backfill language while preserving CLI +> flags, JSON result keys, source-selection semantics, persistence and +> transaction boundaries, and connection close. Status: RED naming/contract +> regression followed by production GREEN locally; GitHub exact-head checks and +> independent review remain pending. + +> Post-content queue backfill naming overlay: 2026-09-07 KST. Protected +> `main` is `83eba56149eb802cd63642c507c324c9976ec78e`. The private +> operator used generic package-owned command, database, queue, record, request, +> and result identifiers (`_parser`, `args`, `connection`, `client`, `rows`, +> `row`, `complete`, `request`, `result`, `settings`, `limit`) plus generic SQL +> aliases. Action: translate the complete repository-local surface to +> post-content queue language while preserving CLI flags, JSON result keys, +> source-selection SQL semantics, transaction boundaries, Valkey publication, +> and resource close behavior. Status: RED naming/contract regression followed +> by production GREEN locally; GitHub exact-head checks and independent review +> remain pending. + +> Occupational catalog synchronizer naming overlay: 2026-09-07 KST. Protected +> `main` is `83eba56149eb802cd63642c507c324c9976ec78e`. The ADR 0250 +> operator used generic package-owned command, database, payload, +> configuration, and result identifiers (`_parser`, `synchronize_catalog`, +> `args`, `conn`, `payload`, `settings`, `count`). Action: translate that +> complete private caller surface to occupational-catalog language while +> preserving the fixed O*NET release URL, CLI flag, output keys, digest gate, +> transactional UPSERT, and connection close. Status: RED naming/contract +> regression followed by production GREEN locally; GitHub exact-head checks +> and independent review remain pending. + +> Deterministic channel-weight estimator naming overlay: 2026-09-07 KST. +> Protected `main` is `83eba56149eb802cd63642c507c324c9976ec78e`. +> The ADR 0200 operator used generic package-owned sampling, database, +> estimate, and command identifiers (`rows`, `record`, `window`, `conn`, +> `estimate`, `_run`, `args`). Action: translate those private identifiers to +> source-post, candidate-window, channel-weight, and command language while +> preserving pair geometry, weight fitting, CLI flags, JSON fields, SQL, and +> persisted provenance. The immutable fast-mlsirm `v0.9.1` consumer cutover is +> owned separately by #967; this naming slice adds no source fallback or +> dependency change. Status: implementation, behavioral tests, and AST +> regression GREEN locally; GitHub exact-head verification pending. + +> Queued LLM channel-weight estimator overlay: the ADR 0200 batch adapter on +> the same protected head used generic owned submit/collect, database, run, +> result, score, and command identifiers (`_submit`, `_collect`, `conn`, `run`, +> `results`, `score`, `args`). Action: translate the private operator surface to +> batch-estimation, estimation-run, pair-judgment, and orchestrator language +> while preserving provider request/response fields, CLI flags, JSON output, +> SQL, transaction boundaries, and fail-closed incomplete judgment behavior. +> Status: implementation, behavioral tests, and AST regression GREEN locally; +> GitHub exact-head verification pending. + +> Exact-head naming overlay: 2026-09-07 KST. Protected `main` is +> `83eba56149eb802cd63642c507c324c9976ec78e`. The bounded post-Keyman +> operator still used generic package-owned command, database, record, and +> result identifiers (`_run`, `args`, `conn`, `row`, `rows`, `settings`). +> Action: rename the complete private caller surface to the Keyman-backfill +> ubiquitous language, preserve CLI flags, JSON result fields, SQL, and +> persistence contracts, and keep the change Proposed until fresh exact-head +> checks and independent review complete. Status: implementation and AST +> regression GREEN locally; GitHub verification pending. + +> Thread-group-key naming overlay: the separate bounded backfill command on the +> same exact protected head also used `_run`, `args`, `conn`, `pool`, `row`, and +> `rows`. Action: carry the same semantic naming rule through that complete +> private caller surface while preserving `--dry-run`, aggregate JSON fields, +> SQL, transaction rollback, and persistence behavior. Status: implementation, +> behavior tests, and AST regression GREEN locally; GitHub verification pending. + > Exact-head loop overlay: 2026-08-29 13:20 KST. Protected `main` is > `fc13acaa20adca11968238e398d4aafcf62b6cee` (v2.23.0 leftover-map > explained leftover share, #775). Open ready PRs still lack independent diff --git a/scripts/backfill_post_content.py b/scripts/backfill_post_content.py index 8fd9cda8c..69b8b21ba 100644 --- a/scripts/backfill_post_content.py +++ b/scripts/backfill_post_content.py @@ -23,45 +23,56 @@ sys.path.insert(0, str(REPOSITORY_ROOT)) from backend.app.post_content_queue import record_post_content_backfill_success -from lineageweave.embedding_client import NullEmbeddingClient, orchestrator_embedding_client -from lineageweave.image_content import NullImageContentClient, orchestrator_vision_client +from lineageweave.embedding_client import ( + NullEmbeddingClient, + orchestrator_embedding_client, +) +from lineageweave.image_content import ( + NullImageContentClient, + orchestrator_vision_client, +) from lineageweave.llm_context import build_post_llm_metadata, use_llm_metadata from lineageweave.post_content_normalization import normalize_post_body from lineageweave.post_content_persistence import persist_post_content -from lineageweave.post_structure import ContextualOrchestratorPostStructureClient, NullPostStructureClient +from lineageweave.post_structure import ( + ContextualOrchestratorPostStructureClient, + NullPostStructureClient, +) -def _parser() -> argparse.ArgumentParser: - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument( +def _post_content_backfill_parser() -> argparse.ArgumentParser: + argument_parser = argparse.ArgumentParser(description=__doc__) + argument_parser.add_argument( "--target-dsn", default=os.environ.get( "DATABASE_URL", "postgresql://lineageweave:lineageweave_dev_only@localhost:15432/lineageweave", ), ) - parser.add_argument("--post-id", action="append", dest="post_ids") - parser.add_argument("--limit", type=int, default=5) - parser.add_argument( + argument_parser.add_argument("--post-id", action="append", dest="post_ids") + argument_parser.add_argument("--limit", type=int, default=5) + argument_parser.add_argument( "--all", action="store_true", help="process every eligible post without persisted content units", ) - parser.add_argument( + argument_parser.add_argument( "--normalize-only", action="store_true", help="persist deterministic DOM/text units without VISION, structure, or embedding calls", ) - return parser + return argument_parser async def backfill_post_content( target_dsn: str, raw_post_ids: list[str] | None, - limit: int | None, + post_limit: int | None, normalize_only: bool = False, ) -> dict[str, int]: - post_ids = [str(uuid.UUID(post_id)) for post_id in dict.fromkeys(raw_post_ids or [])] + post_ids = [ + str(uuid.UUID(post_id)) for post_id in dict.fromkeys(raw_post_ids or []) + ] if normalize_only: vision_client = NullImageContentClient() embedding_client = NullEmbeddingClient() @@ -74,7 +85,9 @@ async def backfill_post_content( orchestrator_api_key, ) if not vision_client.available: - raise RuntimeError("VISION is unavailable; configure contextual-orchestrator before backfill") + raise RuntimeError( + "VISION is unavailable; configure contextual-orchestrator before backfill" + ) embedding_client = orchestrator_embedding_client( orchestrator_base_url, @@ -85,56 +98,58 @@ async def backfill_post_content( "embedding is unavailable; configure contextual-orchestrator before backfill" ) structure_client = ( - ContextualOrchestratorPostStructureClient(orchestrator_base_url, orchestrator_api_key) + ContextualOrchestratorPostStructureClient( + orchestrator_base_url, orchestrator_api_key + ) if orchestrator_base_url and orchestrator_api_key else NullPostStructureClient() ) - conn = await asyncpg.connect(target_dsn) + database_connection = await asyncpg.connect(target_dsn) try: - selected_rows = await conn.fetch( + selected_post_records = await database_connection.fetch( """ - select post.post_id - from source_post post - where nullif(btrim(post.source_draft_code), '') is null - and nullif(btrim(post.source_deleted_flag), '') is null + select source_record.post_id + from source_post source_record + where nullif(btrim(source_record.source_draft_code), '') is null + and nullif(btrim(source_record.source_deleted_flag), '') is null and not ( ( - nullif(btrim(post.source_author_code), '') is null - and nullif(btrim(post.source_author_name), '') is null - and nullif(btrim(post.source_company_code), '') is null - and nullif(btrim(post.source_company_name), '') is null - and nullif(btrim(post.source_process_unit_code), '') is null - and nullif(btrim(post.source_process_unit_name), '') is null - and nullif(btrim(post.source_sales_pool_code), '') is null - and nullif(btrim(post.source_sales_pool_name), '') is null - and nullif(btrim(post.source_customer_code), '') is null - and nullif(btrim(post.source_customer_name), '') is null - and nullif(btrim(post.source_project_code), '') is null - and nullif(btrim(post.source_project_name), '') is null + nullif(btrim(source_record.source_author_code), '') is null + and nullif(btrim(source_record.source_author_name), '') is null + and nullif(btrim(source_record.source_company_code), '') is null + and nullif(btrim(source_record.source_company_name), '') is null + and nullif(btrim(source_record.source_process_unit_code), '') is null + and nullif(btrim(source_record.source_process_unit_name), '') is null + and nullif(btrim(source_record.source_sales_pool_code), '') is null + and nullif(btrim(source_record.source_sales_pool_name), '') is null + and nullif(btrim(source_record.source_customer_code), '') is null + and nullif(btrim(source_record.source_customer_name), '') is null + and nullif(btrim(source_record.source_project_code), '') is null + and nullif(btrim(source_record.source_project_name), '') is null ) and exists ( select 1 - from source_post real_post + from source_post attributed_post where ( - nullif(btrim(real_post.source_author_code), '') is not null - or nullif(btrim(real_post.source_author_name), '') is not null - or nullif(btrim(real_post.source_company_code), '') is not null - or nullif(btrim(real_post.source_company_name), '') is not null - or nullif(btrim(real_post.source_process_unit_code), '') is not null - or nullif(btrim(real_post.source_process_unit_name), '') is not null - or nullif(btrim(real_post.source_sales_pool_code), '') is not null - or nullif(btrim(real_post.source_sales_pool_name), '') is not null - or nullif(btrim(real_post.source_customer_code), '') is not null - or nullif(btrim(real_post.source_customer_name), '') is not null - or nullif(btrim(real_post.source_project_code), '') is not null - or nullif(btrim(real_post.source_project_name), '') is not null + nullif(btrim(attributed_post.source_author_code), '') is not null + or nullif(btrim(attributed_post.source_author_name), '') is not null + or nullif(btrim(attributed_post.source_company_code), '') is not null + or nullif(btrim(attributed_post.source_company_name), '') is not null + or nullif(btrim(attributed_post.source_process_unit_code), '') is not null + or nullif(btrim(attributed_post.source_process_unit_name), '') is not null + or nullif(btrim(attributed_post.source_sales_pool_code), '') is not null + or nullif(btrim(attributed_post.source_sales_pool_name), '') is not null + or nullif(btrim(attributed_post.source_customer_code), '') is not null + or nullif(btrim(attributed_post.source_customer_name), '') is not null + or nullif(btrim(attributed_post.source_project_code), '') is not null + or nullif(btrim(attributed_post.source_project_name), '') is not null ) ) ) and ( ( $1::uuid[] is not null - and post.post_id = any($1::uuid[]) + and source_record.post_id = any($1::uuid[]) ) or ( $1::uuid[] is null @@ -143,8 +158,8 @@ async def backfill_post_content( $2::boolean and not exists ( select 1 - from post_content_unit unit - where unit.post_id = post.post_id + from post_content_unit content_unit + where content_unit.post_id = source_record.post_id ) ) or ( @@ -152,35 +167,35 @@ async def backfill_post_content( and ( not exists ( select 1 - from post_content_unit unit - where unit.post_id = post.post_id + from post_content_unit content_unit + where content_unit.post_id = source_record.post_id ) or exists ( select 1 - from post_content_unit unit - left join post_content_embedding embedding - on embedding.post_content_unit_id = unit.post_content_unit_id - where unit.post_id = post.post_id - and embedding.post_content_unit_id is null + from post_content_unit content_unit + left join post_content_embedding content_embedding + on content_embedding.post_content_unit_id = content_unit.post_content_unit_id + where content_unit.post_id = source_record.post_id + and content_embedding.post_content_unit_id is null ) ) ) ) ) ) - order by post.created_at, post.post_id + order by source_record.created_at, source_record.post_id limit $3::bigint """, post_ids or None, normalize_only, - limit, + post_limit, ) - if post_ids and len(selected_rows) != len(post_ids): + if post_ids and len(selected_post_records) != len(post_ids): raise ValueError("one or more requested post IDs were not found") - result = { + backfill_summary = { "requested_posts": len(post_ids), - "selected_posts": len(selected_rows), + "selected_posts": len(selected_post_records), "processed_posts": 0, "described_posts": 0, "described_images": 0, @@ -188,85 +203,100 @@ async def backfill_post_content( "embedding_rows": 0, "skipped_posts": 0, } - for selected_row in selected_rows: - row = await conn.fetchrow( + for selected_post_record in selected_post_records: + source_post_record = await database_connection.fetchrow( """ - select post.post_id, post.post_title, post.post_body, post.author_account_id, - post.source_process_unit_code, post.source_author_code, - post.source_company_code, post.source_customer_code, - post.source_project_code, post.source_sales_pool_code, - entity.corporate_entity_code - from source_post post - left join corporate_entity entity - on entity.corporate_entity_id = post.corporate_entity_id - where post.post_id = $1 + select source_record.post_id, source_record.post_title, + source_record.post_body, source_record.author_account_id, + source_record.source_process_unit_code, source_record.source_author_code, + source_record.source_company_code, source_record.source_customer_code, + source_record.source_project_code, source_record.source_sales_pool_code, + owning_entity.corporate_entity_code + from source_post source_record + left join corporate_entity owning_entity + on owning_entity.corporate_entity_id = source_record.corporate_entity_id + where source_record.post_id = $1 """, - selected_row["post_id"], + selected_post_record["post_id"], ) - if row is None: + if source_post_record is None: continue - with use_llm_metadata(build_post_llm_metadata(str(row["post_id"]), row)): - normalized = normalize_post_body(row["post_body"], vision_client=vision_client) - described_images = sum( - item.status_code == "described" for item in normalized.image_results + with use_llm_metadata( + build_post_llm_metadata( + str(source_post_record["post_id"]), source_post_record ) - if described_images == 0 and not normalized.text.strip(): - result["skipped_posts"] += 1 + ): + normalized_post_content = normalize_post_body( + source_post_record["post_body"], vision_client=vision_client + ) + described_image_count = sum( + image_result.status_code == "described" + for image_result in normalized_post_content.image_results + ) + if ( + described_image_count == 0 + and not normalized_post_content.text.strip() + ): + backfill_summary["skipped_posts"] += 1 continue await persist_post_content( - conn, - str(row["post_id"]), - row["post_body"], + database_connection, + str(source_post_record["post_id"]), + source_post_record["post_body"], vision_client=vision_client, embedding_client=embedding_client, - normalized_result=normalized, + normalized_result=normalized_post_content, structure_client=structure_client, - post_title=row["post_title"], + post_title=source_post_record["post_title"], ) - async with conn.transaction(): + async with database_connection.transaction(): await record_post_content_backfill_success( - conn, - str(row["post_id"]), - str(row["post_body"] or ""), + database_connection, + str(source_post_record["post_id"]), + str(source_post_record["post_body"] or ""), ) - result["processed_posts"] += 1 - if described_images: - result["described_posts"] += 1 - result["described_images"] += described_images - result["described_regions"] += sum( - len(item.regions) - for item in normalized.image_results - if item.status_code == "described" + backfill_summary["processed_posts"] += 1 + if described_image_count: + backfill_summary["described_posts"] += 1 + backfill_summary["described_images"] += described_image_count + backfill_summary["described_regions"] += sum( + len(image_result.regions) + for image_result in normalized_post_content.image_results + if image_result.status_code == "described" ) - result["embedding_rows"] += await conn.fetchval( + backfill_summary["embedding_rows"] += await database_connection.fetchval( """ select count(*) - from post_content_embedding embedding - join post_content_unit unit using (post_content_unit_id) - where unit.post_id = $1 + from post_content_embedding content_embedding + join post_content_unit content_unit using (post_content_unit_id) + where content_unit.post_id = $1 """, - row["post_id"], + source_post_record["post_id"], ) - return result + return backfill_summary finally: - await conn.close() + await database_connection.close() def main() -> None: - args = _parser().parse_args() - if args.limit < 1: + command_arguments = _post_content_backfill_parser().parse_args() + if command_arguments.limit < 1: raise SystemExit("--limit must be positive") - if args.all and args.post_ids: + if command_arguments.all and command_arguments.post_ids: raise SystemExit("--all cannot be combined with --post-id") - limit = None if args.all or args.post_ids else args.limit + post_limit = ( + None + if command_arguments.all or command_arguments.post_ids + else command_arguments.limit + ) print( json.dumps( asyncio.run( backfill_post_content( - args.target_dsn, - args.post_ids, - limit, - args.normalize_only, + command_arguments.target_dsn, + command_arguments.post_ids, + post_limit, + command_arguments.normalize_only, ) ), sort_keys=True, diff --git a/scripts/backfill_post_keymen.py b/scripts/backfill_post_keymen.py index 59e2efd6c..9c15fe340 100644 --- a/scripts/backfill_post_keymen.py +++ b/scripts/backfill_post_keymen.py @@ -36,25 +36,39 @@ from lineageweave.post_content_normalization import normalize_post_body -def _first_env(*names: str) -> str: - return next((os.environ.get(name, "").strip() for name in names if os.environ.get(name, "").strip()), "") +def _first_env(*variable_names: str) -> str: + return next( + ( + os.environ.get(variable_name, "").strip() + for variable_name in variable_names + if os.environ.get(variable_name, "").strip() + ), + "", + ) def _orchestrator_config() -> tuple[str, str]: - base_url = _first_env("ORCHESTRATOR_BASE_URL", "LLM_GATEWAY_API_URL", "LLM_GATEWAY_URL") + base_url = _first_env( + "ORCHESTRATOR_BASE_URL", "LLM_GATEWAY_API_URL", "LLM_GATEWAY_URL" + ) api_key = _first_env("ORCHESTRATOR_API_KEY", "LLM_GATEWAY_API_KEY") if not base_url or not api_key: - raise RuntimeError("contextual-orchestrator gateway configuration is unavailable") + raise RuntimeError( + "contextual-orchestrator gateway configuration is unavailable" + ) return base_url, api_key async def _select_posts( - conn: asyncpg.Connection, *, limit: int, post_id: str | None + database_connection: asyncpg.Connection, + *, + post_limit: int, + post_id: str | None, ) -> list[asyncpg.Record]: """Select one explicit post or one bounded unprojected batch.""" if post_id: return list( - await conn.fetch( + await database_connection.fetch( """ select post_id, post_title, post_body, author_account_id, source_author_code, source_company_code, @@ -103,7 +117,7 @@ async def _select_posts( ) ) return list( - await conn.fetch( + await database_connection.fetch( """ select post_id, post_title, post_body, author_account_id, source_author_code, source_company_code, @@ -154,16 +168,18 @@ async def _select_posts( order by post.created_at, post.post_id limit $1::bigint """, - limit, + post_limit, ) ) -async def _run(args: argparse.Namespace) -> dict[str, object]: - if args.post_id and args.all: +async def _run_post_keyman_backfill( + command_arguments: argparse.Namespace, +) -> dict[str, object]: + if command_arguments.post_id and command_arguments.all: raise ValueError("--post-id and --all cannot be combined") base_url, api_key = _orchestrator_config() - settings = load_settings() + runtime_settings = load_settings() keyman_client = ContextualOrchestratorKeymanExtractionClient( base_url=base_url, api_key=api_key, timeout=180.0 ) @@ -171,68 +187,100 @@ async def _run(args: argparse.Namespace) -> dict[str, object]: 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 + post_limit = ( + 1 + if command_arguments.post_id or not command_arguments.all + else command_arguments.limit + ) - pool = await asyncpg.create_pool(settings.database_url, min_size=1, max_size=1) + database_pool = await asyncpg.create_pool( + runtime_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) - failures: Counter[str] = Counter() - processed = 0 - mention_count = 0 - for row in rows: - post_id = str(row["post_id"]) + async with database_pool.acquire() as database_connection: + post_records = await _select_posts( + database_connection, + post_limit=post_limit, + post_id=command_arguments.post_id, + ) + failure_counts: Counter[str] = Counter() + processed_post_count = 0 + persisted_mention_count = 0 + for post_record in post_records: + post_id = str(post_record["post_id"]) try: - async with asyncio.timeout(args.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) - mentions = await ingest_post_keymen( - conn, + async with asyncio.timeout(command_arguments.post_timeout): + with use_llm_metadata( + build_post_llm_metadata(post_id, dict(post_record)) + ): + normalized_post_content = normalize_post_body( + post_record["post_body"] or "", vision_client + ) + context_hints = await _load_post_semantic_hints( + database_connection, post_id + ) + persisted_mentions = await ingest_post_keymen( + database_connection, keyman_client, post_id, - row["post_title"] or "", - normalized.text, + post_record["post_title"] or "", + normalized_post_content.text, resolution_client=resolution_client, verification_client=verification_client, hierarchy_inference_client=hierarchy_client, context_hints=context_hints, ) - processed += 1 - mention_count += len(mentions) + processed_post_count += 1 + persisted_mention_count += len(persisted_mentions) except TimeoutError: - failures["TimeoutError"] += 1 - except (HttpClientError, OSError, RuntimeError, ValueError, asyncpg.PostgresError) as exc: - failures[type(exc).__name__] += 1 + failure_counts["TimeoutError"] += 1 + except ( + HttpClientError, + OSError, + RuntimeError, + ValueError, + asyncpg.PostgresError, + ) as backfill_error: + failure_counts[type(backfill_error).__name__] += 1 return { - "failed_posts": sum(failures.values()), - "failure_types": dict(sorted(failures.items())), - "mentions_persisted": mention_count, - "processed_posts": processed, - "requested_posts": len(rows), + "failed_posts": sum(failure_counts.values()), + "failure_types": dict(sorted(failure_counts.items())), + "mentions_persisted": persisted_mention_count, + "processed_posts": processed_post_count, + "requested_posts": len(post_records), } finally: - await pool.close() + await database_pool.close() 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("--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( + argument_parser = argparse.ArgumentParser(description=__doc__) + post_selector = argument_parser.add_mutually_exclusive_group() + post_selector.add_argument("--post-id", help="Re-extract one eligible post") + post_selector.add_argument( + "--all", action="store_true", help="Process the explicit --limit batch" + ) + argument_parser.add_argument( + "--limit", type=int, default=1, help="Maximum posts for --all (default: 1)" + ) + argument_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)) + command_arguments = argument_parser.parse_args() + if command_arguments.limit < 1: + argument_parser.error("--limit must be positive") + if command_arguments.post_timeout <= 0: + argument_parser.error("--post-timeout must be positive") + print( + json.dumps( + asyncio.run(_run_post_keyman_backfill(command_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..9413daa6e 100644 --- a/scripts/backfill_thread_group_keys.py +++ b/scripts/backfill_thread_group_keys.py @@ -61,7 +61,9 @@ from backend.app.config import load_settings -async def backfill_thread_group_keys(conn: asyncpg.Connection, *, dry_run: bool) -> dict[str, int]: +async def backfill_thread_group_keys( + database_connection: asyncpg.Connection, *, dry_run: bool +) -> dict[str, int]: """Clear placeholder grouping keys and route project codes to the secondary-key channel. @@ -72,8 +74,8 @@ async def backfill_thread_group_keys(conn: asyncpg.Connection, *, dry_run: bool) resolved live against `thread_group_key` on every read, not frozen in its snapshot. """ - async with conn.transaction(): - anchored_runs = await conn.fetch( + async with database_connection.transaction(): + anchored_analysis_runs = await database_connection.fetch( """ select scope.analysis_run_id, scope.scope_key from analysis_run_scope scope @@ -85,18 +87,21 @@ async def backfill_thread_group_keys(conn: asyncpg.Connection, *, dry_run: bool) ) """ ) - if anchored_runs: - run_ids = ", ".join(str(row["analysis_run_id"]) for row in anchored_runs) + if anchored_analysis_runs: + analysis_run_ids = ", ".join( + str(anchored_run["analysis_run_id"]) + for anchored_run in anchored_analysis_runs + ) raise RuntimeError( "refusing to rewrite thread_group_key: existing " - f"analysis_scope_thread_group run(s) [{run_ids}] resolve their " + f"analysis_scope_thread_group run(s) [{analysis_run_ids}] resolve their " "scope against values this backfill would change. Retire or " "re-scope those runs first." ) # Only rows carrying the placeholder signature -- a thread key equal # to the row's own record key groups nothing and can only be import # damage; a seeded or genuinely-mapped key never self-references. - rows = await conn.fetch( + updated_post_records = await database_connection.fetch( """ update source_post set source_thread_group_key = coalesce( @@ -112,55 +117,81 @@ async def backfill_thread_group_keys(conn: asyncpg.Connection, *, dry_run: bool) returning (nullif(btrim(source_project_code), '') is not null) as had_project_code """ ) - project_evidence = sum(1 for row in rows if row["had_project_code"]) - cleared = len(rows) + project_evidence_post_count = sum( + 1 + for updated_post_record in updated_post_records + if updated_post_record["had_project_code"] + ) + cleared_post_count = len(updated_post_records) if dry_run: - raise _RollbackDryRun(project_evidence, cleared) + raise _RollbackDryRun( + project_evidence_post_count, + cleared_post_count, + ) return { - "cleared_placeholder_posts": cleared, - "project_secondary_evidence_posts": project_evidence, + "cleared_placeholder_posts": cleared_post_count, + "project_secondary_evidence_posts": project_evidence_post_count, } class _RollbackDryRun(Exception): """Raised inside the transaction to force a rollback for --dry-run.""" - def __init__(self, project_evidence: int, cleared: int) -> None: + def __init__( + self, + project_evidence_post_count: int, + cleared_post_count: int, + ) -> None: """Retain the aggregate counts that the rolled-back operator run reports.""" super().__init__("dry run -- rolled back") - self.project_evidence = project_evidence - self.cleared = cleared + self.project_evidence_post_count = project_evidence_post_count + self.cleared_post_count = cleared_post_count -async def _run(args: argparse.Namespace) -> dict[str, object]: +async def _run_thread_group_key_backfill( + command_arguments: argparse.Namespace, +) -> dict[str, object]: """Execute one pooled backfill and convert dry-run rollback into counts.""" - settings = load_settings() - pool = await asyncpg.create_pool(settings.database_url, min_size=1, max_size=1) + runtime_settings = load_settings() + database_pool = await asyncpg.create_pool( + runtime_settings.database_url, min_size=1, max_size=1 + ) try: - async with pool.acquire() as conn: + async with database_pool.acquire() as database_connection: try: - counts = await backfill_thread_group_keys(conn, dry_run=args.dry_run) - return {**counts, "dry_run": False} + backfill_counts = await backfill_thread_group_keys( + database_connection, + dry_run=command_arguments.dry_run, + ) + return {**backfill_counts, "dry_run": False} except _RollbackDryRun as rolled_back: return { - "cleared_placeholder_posts": rolled_back.cleared, - "project_secondary_evidence_posts": rolled_back.project_evidence, + "cleared_placeholder_posts": rolled_back.cleared_post_count, + "project_secondary_evidence_posts": ( + rolled_back.project_evidence_post_count + ), "dry_run": True, } finally: - await pool.close() + await database_pool.close() def main() -> None: """Parse operator arguments and print aggregate, non-identifying evidence.""" - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument( + argument_parser = argparse.ArgumentParser(description=__doc__) + argument_parser.add_argument( "--dry-run", 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)) + command_arguments = argument_parser.parse_args() + print( + json.dumps( + asyncio.run(_run_thread_group_key_backfill(command_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..317cb2517 100644 --- a/scripts/estimate_channel_weights.py +++ b/scripts/estimate_channel_weights.py @@ -58,9 +58,9 @@ def estimator_version() -> str: """The installed fast-mlsirm version, for the persisted provenance.""" from importlib.metadata import PackageNotFoundError, version - for name in ("fast-mlsirm", "fast_mlsirm"): + for distribution_name in ("fast-mlsirm", "fast_mlsirm"): try: - return version(name) + return version(distribution_name) except PackageNotFoundError: continue import fast_mlsirm @@ -68,21 +68,22 @@ def estimator_version() -> str: return str(getattr(fast_mlsirm, "__version__", "unknown")) -def source_snapshot_digest(rows: list) -> str: +def source_snapshot_digest(source_post_rows: list) -> str: """Reproducible SHA-256 over the ordered sampled (post_id, created_at). Two runs that sampled the same posts in the same order produce the same digest, so the provenance row names exactly which corpus slice supported the estimate without storing any post content. """ - material = "\n".join( - f"{row['post_id']}\t{row['created_at'].isoformat()}" for row in rows + digest_material = "\n".join( + f"{source_post_row['post_id']}\t{source_post_row['created_at'].isoformat()}" + for source_post_row in source_post_rows ) - return hashlib.sha256(material.encode("utf-8")).hexdigest() + return hashlib.sha256(digest_material.encode("utf-8")).hexdigest() def sample_pair_scores( - records: list, *, window: int = DEFAULT_CANDIDATE_WINDOW + lineage_records: list, *, candidate_window: int = DEFAULT_CANDIDATE_WINDOW ) -> tuple[list[dict[str, float]], list[int], list[tuple[str, str]]]: """Score every in-window candidate pair, grouped as reconstruct groups. @@ -93,45 +94,54 @@ def sample_pair_scores( (candidate_label, record_label) so the queued llm judging pass can score the same candidate geometry without re-deriving it. """ - groups: dict[str, list] = {} - for record in records: - groups.setdefault(record.group_key, []).append(record) + channel_groups: dict[str, list] = {} + for lineage_record in lineage_records: + channel_groups.setdefault(lineage_record.group_key, []).append(lineage_record) pair_scores: list[dict[str, float]] = [] group_ids: list[int] = [] pair_labels: list[tuple[str, str]] = [] - for group_index, group_records in enumerate(groups.values()): - ordered = sorted(group_records, key=lambda r: r.occurred_at) - for index, record in enumerate(ordered): - for candidate in ordered[max(0, index - window) : index]: + for group_index, group_records in enumerate(channel_groups.values()): + ordered_records = sorted( + group_records, key=lambda grouped_record: grouped_record.occurred_at + ) + for record_index, lineage_record in enumerate(ordered_records): + for candidate_record in ordered_records[ + max(0, record_index - candidate_window) : record_index + ]: pair_scores.append( { - "temporal": temporal_score(candidate, record), - "secondary_key": secondary_key_match_score(candidate, record), - "text": text_similarity_score(candidate, record), + "temporal": temporal_score(candidate_record, lineage_record), + "secondary_key": secondary_key_match_score( + candidate_record, lineage_record + ), + "text": text_similarity_score(candidate_record, lineage_record), } ) group_ids.append(group_index) - pair_labels.append((candidate.label, record.label)) + pair_labels.append((candidate_record.label, lineage_record.label)) return pair_scores, group_ids, pair_labels -def subsample_stride(total: int, limit: int) -> list[int]: +def subsample_stride(sample_pair_total: int, sample_pair_limit: int) -> list[int]: """Deterministic, evenly-spread pair indices for the bounded llm pass. A stride subsample keeps every reconstruction group represented in proportion (pairs are ordered group-by-group) without any randomness that would make re-runs incomparable. """ - if total <= limit: - return list(range(total)) - stride = total / limit - return [min(int(index * stride), total - 1) for index in range(limit)] + if sample_pair_total <= sample_pair_limit: + return list(range(sample_pair_total)) + sample_stride = sample_pair_total / sample_pair_limit + return [ + min(int(sample_index * sample_stride), sample_pair_total - 1) + for sample_index in range(sample_pair_limit) + ] async def persist_estimate( - conn: asyncpg.Connection, - estimate: ChannelWeightEstimate, + database_connection: asyncpg.Connection, + channel_weight_estimate: ChannelWeightEstimate, *, channel_set_code: str, snapshot_sha256: str, @@ -142,14 +152,14 @@ async def persist_estimate( Returns the estimation run id stamped on every row of the set. """ estimation_run_id = str(uuid.uuid4()) - version = estimator_version() - async with conn.transaction(): - await conn.execute( + installed_estimator_version = estimator_version() + async with database_connection.transaction(): + await database_connection.execute( "delete from lineage_channel_weight where channel_set_code = $1", channel_set_code, ) - for channel, weight in estimate.weights.items(): - await conn.execute( + for channel_code, weight_value in channel_weight_estimate.weights.items(): + await database_connection.execute( """ insert into lineage_channel_weight (channel_set_code, channel_code, weight_value, @@ -160,45 +170,49 @@ async def persist_estimate( values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) """, channel_set_code, - channel, - weight, + channel_code, + weight_value, estimation_run_id, - estimate.estimation_method_code, - version, + channel_weight_estimate.estimation_method_code, + installed_estimator_version, UNANCHORED_METHOD_CODE, snapshot_sha256, - estimate.sample_pair_count, + channel_weight_estimate.sample_pair_count, knowledge_cutoff, ) return estimation_run_id -async def _run(args: argparse.Namespace) -> dict[str, object]: - settings = load_settings() +async def _run_channel_weight_estimation( + command_arguments: argparse.Namespace, +) -> dict[str, object]: + runtime_settings = load_settings() # Short-lived fetch connection; nothing stays open while fitting. - conn = await asyncpg.connect(settings.database_url) + database_connection = await asyncpg.connect(runtime_settings.database_url) try: - rows = await conn.fetch( + source_post_rows = await database_connection.fetch( "select post_id, post_title, voc_type_code, created_at, " "corporate_entity_id, process_unit_id, thread_group_key, " "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, + command_arguments.post_limit, ) finally: - await conn.close() - if not rows: + await database_connection.close() + if not source_post_rows: raise RuntimeError( "no eligible source posts exist; import a corpus before estimating" ) - snapshot_sha256 = source_snapshot_digest(rows) - knowledge_cutoff = max(row["created_at"] for row in rows) - records = records_from_source_posts(rows) - pair_scores, group_ids, _pair_labels = sample_pair_scores(records) + snapshot_sha256 = source_snapshot_digest(source_post_rows) + knowledge_cutoff = max( + source_post_row["created_at"] for source_post_row in source_post_rows + ) + lineage_records = records_from_source_posts(source_post_rows) + pair_scores, group_ids, _pair_labels = sample_pair_scores(lineage_records) - estimate = estimate_channel_weights(pair_scores, group_ids) - if estimate is None: + channel_weight_estimate = estimate_channel_weights(pair_scores, group_ids) + if channel_weight_estimate is None: raise RuntimeError( "no grounded estimate was produced (fast_mlsirm unavailable, " "sample too small, a channel degenerate, or the fit did not " @@ -206,28 +220,28 @@ async def _run(args: argparse.Namespace) -> dict[str, object]: "named condition" ) estimation_run_id = None - if not args.dry_run: - conn = await asyncpg.connect(settings.database_url) + if not command_arguments.dry_run: + database_connection = await asyncpg.connect(runtime_settings.database_url) try: estimation_run_id = await persist_estimate( - conn, - estimate, + database_connection, + channel_weight_estimate, channel_set_code=DETERMINISTIC_SET_CODE, snapshot_sha256=snapshot_sha256, knowledge_cutoff=knowledge_cutoff, ) finally: - await conn.close() + await database_connection.close() return { - "weights": estimate.weights, + "weights": channel_weight_estimate.weights, "channel_set_code": DETERMINISTIC_SET_CODE, - "sample_pair_count": estimate.sample_pair_count, - "estimation_method_code": estimate.estimation_method_code, + "sample_pair_count": channel_weight_estimate.sample_pair_count, + "estimation_method_code": channel_weight_estimate.estimation_method_code, "anchor_method_code": UNANCHORED_METHOD_CODE, "estimation_run_id": estimation_run_id, "source_snapshot_sha256": snapshot_sha256, "knowledge_cutoff": knowledge_cutoff.isoformat(), - "persisted": not args.dry_run, + "persisted": not command_arguments.dry_run, "activation": ( "blocked_until_anchor_authorized (ADR 0200 point 3): the " "product loader refuses every anchor method today, so these " @@ -238,22 +252,28 @@ async def _run(args: argparse.Namespace) -> dict[str, object]: def main() -> None: """Validate operator inputs and run the estimation.""" - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument( + argument_parser = argparse.ArgumentParser(description=__doc__) + argument_parser.add_argument( "--post-limit", type=int, default=5000, help="Maximum eligible posts to sample pairs from (default: 5000)", ) - parser.add_argument( + argument_parser.add_argument( "--dry-run", action="store_true", help="Estimate and report, but persist nothing", ) - args = parser.parse_args() - if args.post_limit < 1: - parser.error("--post-limit must be positive") - print(json.dumps(asyncio.run(_run(args)), ensure_ascii=False, sort_keys=True)) + command_arguments = argument_parser.parse_args() + if command_arguments.post_limit < 1: + argument_parser.error("--post-limit must be positive") + print( + json.dumps( + asyncio.run(_run_channel_weight_estimation(command_arguments)), + ensure_ascii=False, + sort_keys=True, + ) + ) if __name__ == "__main__": diff --git a/scripts/estimate_llm_channel_weights.py b/scripts/estimate_llm_channel_weights.py index 1613ceb90..1beef15f1 100644 --- a/scripts/estimate_llm_channel_weights.py +++ b/scripts/estimate_llm_channel_weights.py @@ -55,32 +55,38 @@ def _orchestrator_config() -> tuple[str, str]: """Base URL and bearer key for the batch routing API, from the environment.""" - base_url = next( + orchestrator_base_url = next( ( - os.environ[name].strip() - for name in ("ORCHESTRATOR_BASE_URL", "LLM_GATEWAY_API_URL") - if os.environ.get(name, "").strip() + os.environ[environment_variable_name].strip() + for environment_variable_name in ( + "ORCHESTRATOR_BASE_URL", + "LLM_GATEWAY_API_URL", + ) + if os.environ.get(environment_variable_name, "").strip() ), "", ) - api_key = next( + orchestrator_api_key = next( ( - os.environ[name].strip() - for name in ("ORCHESTRATOR_API_KEY", "CONTEXTUAL_ORCHESTRATOR_TOKEN") - if os.environ.get(name, "").strip() + os.environ[environment_variable_name].strip() + for environment_variable_name in ( + "ORCHESTRATOR_API_KEY", + "CONTEXTUAL_ORCHESTRATOR_TOKEN", + ) + if os.environ.get(environment_variable_name, "").strip() ), "", ) - if not base_url or not api_key: + if not orchestrator_base_url or not orchestrator_api_key: raise RuntimeError( "set ORCHESTRATOR_BASE_URL and ORCHESTRATOR_API_KEY (or " "CONTEXTUAL_ORCHESTRATOR_TOKEN) to reach the batch routing API" ) - return base_url.rstrip("/"), api_key + return orchestrator_base_url.rstrip("/"), orchestrator_api_key def batch_requests_for_pairs( - chosen: list[int], pair_labels: list[tuple[str, str]] + chosen_pair_ordinals: list[int], candidate_pair_labels: list[tuple[str, str]] ) -> list[dict[str, object]]: """One batch request per chosen pair, keyed by its ordinal. @@ -91,60 +97,70 @@ def batch_requests_for_pairs( """ return [ { - "custom_id": f"pair-{ordinal}", + "custom_id": f"pair-{pair_ordinal}", "mode": "auto", "messages": [ { "role": "user", - "content": judge_prompt(*pair_labels[ordinal]), + "content": judge_prompt(*candidate_pair_labels[pair_ordinal]), } ], } - for ordinal in chosen + for pair_ordinal in chosen_pair_ordinals ] -async def _submit(args: argparse.Namespace) -> dict[str, object]: +async def _submit_batch_estimation( + command_arguments: argparse.Namespace, +) -> dict[str, object]: """Sample, submit one batch job, persist the run ledger. Never waits.""" - base_url, api_key = _orchestrator_config() - settings = load_settings() - conn = await asyncpg.connect(settings.database_url) + orchestrator_base_url, orchestrator_api_key = _orchestrator_config() + runtime_settings = load_settings() + database_connection = await asyncpg.connect(runtime_settings.database_url) try: - rows = await conn.fetch( + source_post_rows = await database_connection.fetch( "select post_id, post_title, voc_type_code, created_at, " "corporate_entity_id, process_unit_id, thread_group_key, " "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, + command_arguments.post_limit, ) finally: - await conn.close() - if not rows: + await database_connection.close() + if not source_post_rows: raise RuntimeError( "no eligible source posts exist; import a corpus before estimating" ) - snapshot_sha256 = source_snapshot_digest(rows) - knowledge_cutoff = max(row["created_at"] for row in rows) - pair_scores, group_ids, pair_labels = sample_pair_scores( - records_from_source_posts(rows) + source_snapshot_sha256 = source_snapshot_digest(source_post_rows) + knowledge_cutoff = max( + source_post_row["created_at"] for source_post_row in source_post_rows + ) + candidate_pair_scores, reconstruction_group_ids, candidate_pair_labels = ( + sample_pair_scores(records_from_source_posts(source_post_rows)) ) - chosen = subsample_stride(len(pair_scores), args.pair_limit) - if not chosen: + chosen_pair_ordinals = subsample_stride( + len(candidate_pair_scores), command_arguments.pair_limit + ) + if not chosen_pair_ordinals: raise RuntimeError("the corpus produced no candidate pairs to judge") - submitted = post_json( - f"{base_url}/api/v1/batch_routing_jobs", - {"requests": batch_requests_for_pairs(chosen, pair_labels)}, - headers={"authorization": f"Bearer {api_key}"}, + batch_submission = post_json( + f"{orchestrator_base_url}/api/v1/batch_routing_jobs", + { + "requests": batch_requests_for_pairs( + chosen_pair_ordinals, candidate_pair_labels + ) + }, + headers={"authorization": f"Bearer {orchestrator_api_key}"}, timeout=_BATCH_TIMEOUT_SECONDS, ) - batch_job_id = str(submitted["job_id"]) + batch_job_id = str(batch_submission["job_id"]) - conn = await asyncpg.connect(settings.database_url) + database_connection = await asyncpg.connect(runtime_settings.database_url) try: - async with conn.transaction(): - estimation_run_id = await conn.fetchval( + async with database_connection.transaction(): + estimation_run_id = await database_connection.fetchval( """ insert into lineage_weight_estimation_run (estimation_run_id, channel_set_code, run_status_code, @@ -155,14 +171,14 @@ async def _submit(args: argparse.Namespace) -> dict[str, object]: """, WITH_LLM_SET_CODE, batch_job_id, - snapshot_sha256, + source_snapshot_sha256, knowledge_cutoff, - len(chosen), + len(chosen_pair_ordinals), ) - for ordinal in chosen: - scores = pair_scores[ordinal] - candidate_label, record_label = pair_labels[ordinal] - await conn.execute( + for pair_ordinal in chosen_pair_ordinals: + channel_scores = candidate_pair_scores[pair_ordinal] + candidate_label, record_label = candidate_pair_labels[pair_ordinal] + await database_connection.execute( """ insert into lineage_pair_judgment (estimation_run_id, pair_ordinal, group_ordinal, @@ -171,39 +187,42 @@ async def _submit(args: argparse.Namespace) -> dict[str, object]: values ($1, $2, $3, $4, $5, $6, $7, $8) """, estimation_run_id, - ordinal, - group_ids[ordinal], + pair_ordinal, + reconstruction_group_ids[pair_ordinal], candidate_label, record_label, - scores["temporal"], - scores["secondary_key"], - scores["text"], + channel_scores["temporal"], + channel_scores["secondary_key"], + channel_scores["text"], ) - except Exception as exc: + except Exception as submission_error: raise RuntimeError( f"batch job {batch_job_id} was submitted but the run ledger " "could not be persisted; re-run submit (the orphaned job only " "costs its provider spend, no state references it)" - ) from exc + ) from submission_error finally: - await conn.close() + await database_connection.close() return { "estimation_run_id": str(estimation_run_id), "batch_job_id": batch_job_id, - "sampled_pair_count": len(chosen), + "sampled_pair_count": len(chosen_pair_ordinals), "next_action": "run collect once the batch job completes", } -def _is_complete(polled: dict[str, object]) -> bool: +def _is_complete(batch_status_payload: dict[str, object]) -> bool: """True when the batch backend reports a terminal successful state.""" - if polled.get("is_complete") is True: + if batch_status_payload.get("is_complete") is True: return True - return str(polled.get("status", "")).lower() in {"completed", "succeeded"} + return str(batch_status_payload.get("status", "")).lower() in { + "completed", + "succeeded", + } def judgment_updates_from_results( - results: list[dict[str, object]], + batch_result_records: list[dict[str, object]], ) -> list[tuple[int, float]]: """Map batch results onto (pair_ordinal, llm_score) updates. @@ -212,36 +231,38 @@ def judgment_updates_from_results( errored request must stay unjudged rather than become a confident 0.0 ("definitely unrelated") verdict the judge never gave. """ - updates: list[tuple[int, float]] = [] - for item in results: - custom_id = str(item.get("custom_id", "")) + judgment_updates: list[tuple[int, float]] = [] + for batch_result_record in batch_result_records: + custom_id = str(batch_result_record.get("custom_id", "")) if not custom_id.startswith("pair-"): continue try: - ordinal = int(custom_id.removeprefix("pair-")) + pair_ordinal = int(custom_id.removeprefix("pair-")) except ValueError: continue - score = parse_confidence_or_none(str(item.get("answer", ""))) - if score is None: + llm_score = parse_confidence_or_none(str(batch_result_record.get("answer", ""))) + if llm_score is None: continue - updates.append((ordinal, score)) - return updates + judgment_updates.append((pair_ordinal, llm_score)) + return judgment_updates -async def _collect(args: argparse.Namespace) -> dict[str, object]: +async def _collect_batch_estimation( + command_arguments: argparse.Namespace, +) -> dict[str, object]: """Collect one completed batch into the ledger; fit when the run is whole. No database connection is held across the HTTP calls or the model fit (an idle-reaped connection killed an earlier estimation run): each phase opens its own short-lived connection. """ - base_url, api_key = _orchestrator_config() - settings = load_settings() + orchestrator_base_url, orchestrator_api_key = _orchestrator_config() + runtime_settings = load_settings() - conn = await asyncpg.connect(settings.database_url) + database_connection = await asyncpg.connect(runtime_settings.database_url) try: - if args.run_id: - run = await conn.fetchrow( + if command_arguments.run_id: + estimation_run_record = await database_connection.fetchrow( """ select estimation_run_id, batch_job_id, run_status_code, source_snapshot_sha256, knowledge_cutoff, sampled_pair_count @@ -249,10 +270,10 @@ async def _collect(args: argparse.Namespace) -> dict[str, object]: where estimation_run_id = $1::uuid and run_status_code in ('run_submitted', 'run_collecting') """, - args.run_id, + command_arguments.run_id, ) else: - run = await conn.fetchrow( + estimation_run_record = await database_connection.fetchrow( """ select estimation_run_id, batch_job_id, run_status_code, source_snapshot_sha256, knowledge_cutoff, sampled_pair_count @@ -263,52 +284,56 @@ async def _collect(args: argparse.Namespace) -> dict[str, object]: """ ) finally: - await conn.close() - if run is None: + await database_connection.close() + if estimation_run_record is None: raise RuntimeError( "no submitted run awaits collection; run submit first " "(or pass --run-id for an older run)" ) - polled = get_json( - f"{base_url}/api/v1/batch_routing_jobs/{run['batch_job_id']}", - headers={"authorization": f"Bearer {api_key}"}, + batch_status_payload = get_json( + f"{orchestrator_base_url}/api/v1/batch_routing_jobs/" + f"{estimation_run_record['batch_job_id']}", + headers={"authorization": f"Bearer {orchestrator_api_key}"}, timeout=_BATCH_TIMEOUT_SECONDS, service_peer_name="contextual-orchestrator", ) - if not _is_complete(polled): + if not _is_complete(batch_status_payload): return { - "estimation_run_id": str(run["estimation_run_id"]), - "batch_job_id": run["batch_job_id"], - "batch_status": polled.get("status"), + "estimation_run_id": str(estimation_run_record["estimation_run_id"]), + "batch_job_id": estimation_run_record["batch_job_id"], + "batch_status": batch_status_payload.get("status"), "next_action": "batch not complete yet; run collect again later", } - retrieved = post_json( - f"{base_url}/api/v1/batch_routing_jobs/{run['batch_job_id']}/results", + batch_results_payload = post_json( + f"{orchestrator_base_url}/api/v1/batch_routing_jobs/" + f"{estimation_run_record['batch_job_id']}/results", {}, - headers={"authorization": f"Bearer {api_key}"}, + headers={"authorization": f"Bearer {orchestrator_api_key}"}, timeout=_BATCH_TIMEOUT_SECONDS, ) - updates = judgment_updates_from_results(retrieved.get("results", [])) + judgment_updates = judgment_updates_from_results( + batch_results_payload.get("results", []) + ) judged_at = datetime.now(timezone.utc) - conn = await asyncpg.connect(settings.database_url) + database_connection = await asyncpg.connect(runtime_settings.database_url) try: - async with conn.transaction(): - for ordinal, score in updates: - await conn.execute( + async with database_connection.transaction(): + for pair_ordinal, llm_score in judgment_updates: + await database_connection.execute( """ update lineage_pair_judgment set llm_score = $3, judged_at = $4 where estimation_run_id = $1 and pair_ordinal = $2 """, - run["estimation_run_id"], - ordinal, - score, + estimation_run_record["estimation_run_id"], + pair_ordinal, + llm_score, judged_at, ) - await conn.execute( + await database_connection.execute( """ update lineage_weight_estimation_run set run_status_code = 'run_collecting', @@ -318,9 +343,9 @@ async def _collect(args: argparse.Namespace) -> dict[str, object]: ) where estimation_run_id = $1 """, - run["estimation_run_id"], + estimation_run_record["estimation_run_id"], ) - pairs = await conn.fetch( + pair_judgment_records = await database_connection.fetch( """ select group_ordinal, temporal_score, secondary_key_score, text_score, llm_score @@ -328,46 +353,53 @@ async def _collect(args: argparse.Namespace) -> dict[str, object]: where estimation_run_id = $1 order by pair_ordinal """, - run["estimation_run_id"], + estimation_run_record["estimation_run_id"], ) finally: - await conn.close() + await database_connection.close() - unjudged = sum(1 for row in pairs if row["llm_score"] is None) - if unjudged: + unjudged_pair_count = sum( + 1 + for pair_judgment_record in pair_judgment_records + if pair_judgment_record["llm_score"] is None + ) + if unjudged_pair_count: return { - "estimation_run_id": str(run["estimation_run_id"]), - "judged_pair_count": len(pairs) - unjudged, - "sampled_pair_count": len(pairs), + "estimation_run_id": str(estimation_run_record["estimation_run_id"]), + "judged_pair_count": len(pair_judgment_records) - unjudged_pair_count, + "sampled_pair_count": len(pair_judgment_records), "next_action": ( - f"{unjudged} pairs have no parseable judgment yet; run " + f"{unjudged_pair_count} pairs have no parseable judgment yet; run " "collect again once the batch delivers them, or re-submit " "if the provider errored them permanently" ), } # The fit can take minutes; no connection is open while it runs. - estimate = estimate_channel_weights( + channel_weight_estimate = estimate_channel_weights( [ { - "temporal": row["temporal_score"], - "secondary_key": row["secondary_key_score"], - "text": row["text_score"], - "llm": row["llm_score"], + "temporal": pair_judgment_record["temporal_score"], + "secondary_key": pair_judgment_record["secondary_key_score"], + "text": pair_judgment_record["text_score"], + "llm": pair_judgment_record["llm_score"], } - for row in pairs + for pair_judgment_record in pair_judgment_records + ], + [ + int(pair_judgment_record["group_ordinal"]) + for pair_judgment_record in pair_judgment_records ], - [int(row["group_ordinal"]) for row in pairs], ) - conn = await asyncpg.connect(settings.database_url) + database_connection = await asyncpg.connect(runtime_settings.database_url) try: - if estimate is None: - await conn.execute( + if channel_weight_estimate is None: + await database_connection.execute( "update lineage_weight_estimation_run " "set run_status_code = 'run_failed', completed_at = now() " "where estimation_run_id = $1", - run["estimation_run_id"], + estimation_run_record["estimation_run_id"], ) raise RuntimeError( "no grounded estimate was produced over the judged pairs " @@ -376,26 +408,26 @@ async def _collect(args: argparse.Namespace) -> dict[str, object]: "marked run_failed; nothing was written to the weight table" ) await persist_estimate( - conn, - estimate, + database_connection, + channel_weight_estimate, channel_set_code=WITH_LLM_SET_CODE, - snapshot_sha256=run["source_snapshot_sha256"], - knowledge_cutoff=run["knowledge_cutoff"], + snapshot_sha256=estimation_run_record["source_snapshot_sha256"], + knowledge_cutoff=estimation_run_record["knowledge_cutoff"], ) - await conn.execute( + await database_connection.execute( "update lineage_weight_estimation_run " "set run_status_code = 'run_fitted', completed_at = now() " "where estimation_run_id = $1", - run["estimation_run_id"], + estimation_run_record["estimation_run_id"], ) finally: - await conn.close() + await database_connection.close() return { - "estimation_run_id": str(run["estimation_run_id"]), - "weights": estimate.weights, + "estimation_run_id": str(estimation_run_record["estimation_run_id"]), + "weights": channel_weight_estimate.weights, "channel_set_code": WITH_LLM_SET_CODE, - "sample_pair_count": estimate.sample_pair_count, - "estimation_method_code": estimate.estimation_method_code, + "sample_pair_count": channel_weight_estimate.sample_pair_count, + "estimation_method_code": channel_weight_estimate.estimation_method_code, "activation": ( "blocked_until_anchor_authorized (ADR 0200 point 3): the " "product loader refuses every anchor method today" @@ -405,29 +437,31 @@ async def _collect(args: argparse.Namespace) -> dict[str, object]: def main() -> None: """Validate operator inputs and run the chosen phase.""" - parser = argparse.ArgumentParser(description=__doc__) - subcommands = parser.add_subparsers(dest="phase", required=True) - submit = subcommands.add_parser("submit", help="sample pairs and submit one batch job") - submit.add_argument("--post-limit", type=int, default=5000) - submit.add_argument("--pair-limit", type=int, default=400) - collect = subcommands.add_parser( + argument_parser = argparse.ArgumentParser(description=__doc__) + phase_subparsers = argument_parser.add_subparsers(dest="phase", required=True) + submit_parser = phase_subparsers.add_parser( + "submit", help="sample pairs and submit one batch job" + ) + submit_parser.add_argument("--post-limit", type=int, default=5000) + submit_parser.add_argument("--pair-limit", type=int, default=400) + collect_parser = phase_subparsers.add_parser( "collect", help="collect results; fit when the run is whole" ) - collect.add_argument( + collect_parser.add_argument( "--run-id", default="", help="collect a specific estimation run (default: the newest awaiting one)", ) - args = parser.parse_args() - if args.phase == "submit": - if args.post_limit < 1: - parser.error("--post-limit must be positive") - if args.pair_limit < 1: - parser.error("--pair-limit must be positive") - result = asyncio.run(_submit(args)) + command_arguments = argument_parser.parse_args() + if command_arguments.phase == "submit": + if command_arguments.post_limit < 1: + argument_parser.error("--post-limit must be positive") + if command_arguments.pair_limit < 1: + argument_parser.error("--pair-limit must be positive") + command_result = asyncio.run(_submit_batch_estimation(command_arguments)) else: - result = asyncio.run(_collect(args)) - print(json.dumps(result, ensure_ascii=False, sort_keys=True, default=str)) + command_result = asyncio.run(_collect_batch_estimation(command_arguments)) + print(json.dumps(command_result, ensure_ascii=False, sort_keys=True, default=str)) if __name__ == "__main__": diff --git a/scripts/queue_post_content_backfill.py b/scripts/queue_post_content_backfill.py index bac966ddc..e481a2946 100644 --- a/scripts/queue_post_content_backfill.py +++ b/scripts/queue_post_content_backfill.py @@ -24,45 +24,54 @@ from backend.app.config import load_settings # noqa: E402 -def _parser() -> argparse.ArgumentParser: - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument( +def _queue_backfill_parser() -> argparse.ArgumentParser: + """Build the post-content queue backfill command parser.""" + argument_parser = argparse.ArgumentParser(description=__doc__) + argument_parser.add_argument( "--target-dsn", default=os.environ.get( "DATABASE_URL", "postgresql://lineageweave:lineageweave_dev_only@localhost:15432/lineageweave", ), ) - parser.add_argument( + argument_parser.add_argument( "--valkey-url", default=os.environ.get("VALKEY_URL", "redis://localhost:16379/0"), ) - parser.add_argument("--limit", type=int, default=100) - parser.add_argument("--all", action="store_true", help="scan the complete real corpus") - return parser + argument_parser.add_argument("--limit", type=int, default=100) + argument_parser.add_argument( + "--all", action="store_true", help="scan the complete real corpus" + ) + return argument_parser async def queue_post_content_backfill( target_dsn: str, valkey_url: str, *, - limit: int | None, + post_limit: int | None, ) -> dict[str, int]: - if limit is not None and limit < 1: + """Queue incomplete post-content work and return aggregate counts.""" + if post_limit is not None and post_limit < 1: raise ValueError("limit must be positive") - settings = load_settings() + runtime_settings = load_settings() require_orchestrator_evidence = bool( - settings.orchestrator_base_url and settings.orchestrator_api_key + runtime_settings.orchestrator_base_url and runtime_settings.orchestrator_api_key ) - connection = await asyncpg.connect(target_dsn) - client = redis.from_url(valkey_url, decode_responses=True) - result = {"scanned_posts": 0, "already_complete": 0, "queued_posts": 0, "published_events": 0} + database_connection = await asyncpg.connect(target_dsn) + valkey_client = redis.from_url(valkey_url, decode_responses=True) + backfill_summary = { + "scanned_posts": 0, + "already_complete": 0, + "queued_posts": 0, + "published_events": 0, + } try: - rows = await connection.fetch( + source_post_records = await database_connection.fetch( """ select post_id, post_body - from source_post post + from source_post source_record where nullif(btrim(source_draft_code), '') is null and nullif(btrim(source_deleted_flag), '') is null and ( @@ -82,95 +91,98 @@ async def queue_post_content_backfill( and ( not exists ( select 1 - from post_content_unit unit - where unit.post_id = post.post_id + from post_content_unit content_unit + where content_unit.post_id = source_record.post_id ) or ($1::boolean and exists ( select 1 - from post_content_unit unit - left join post_content_embedding embedding - on embedding.post_content_unit_id = unit.post_content_unit_id - where unit.post_id = post.post_id - and embedding.post_content_embedding_id is null + from post_content_unit content_unit + left join post_content_embedding content_embedding + on content_embedding.post_content_unit_id = content_unit.post_content_unit_id + where content_unit.post_id = source_record.post_id + and content_embedding.post_content_embedding_id is null )) or ($1::boolean and exists ( select 1 - from post_content_unit unit - join post_content_image image - on image.post_content_unit_id = unit.post_content_unit_id - join post_content_image_region region - on region.post_content_image_id = image.post_content_image_id - left join post_content_image_region_embedding embedding - on embedding.post_content_image_region_id = region.post_content_image_region_id - where unit.post_id = post.post_id - and region.description_status_code = 'described' - and embedding.post_content_image_region_embedding_id is null + from post_content_unit content_unit + join post_content_image content_image + on content_image.post_content_unit_id = content_unit.post_content_unit_id + join post_content_image_region image_region + on image_region.post_content_image_id = content_image.post_content_image_id + left join post_content_image_region_embedding region_embedding + on region_embedding.post_content_image_region_id = image_region.post_content_image_region_id + where content_unit.post_id = source_record.post_id + and image_region.description_status_code = 'described' + and region_embedding.post_content_image_region_embedding_id is null )) or ($2::boolean and exists ( select 1 - from post_content_unit unit - left join post_content_unit_structure structure - on structure.post_content_unit_id = unit.post_content_unit_id - where unit.post_id = post.post_id - and unit.unit_kind_code <> 'image' + from post_content_unit content_unit + left join post_content_unit_structure unit_structure + on unit_structure.post_content_unit_id = content_unit.post_content_unit_id + where content_unit.post_id = source_record.post_id + and content_unit.unit_kind_code <> 'image' and ( - structure.post_content_unit_structure_id is null - or structure.decision_source_code = 'unresolved' + unit_structure.post_content_unit_structure_id is null + or unit_structure.decision_source_code = 'unresolved' ) )) ) - order by post.created_at, post.post_id + order by source_record.created_at, source_record.post_id limit $3::bigint """, require_orchestrator_evidence, require_orchestrator_evidence, - limit if limit is not None else 9223372036854775807, + post_limit if post_limit is not None else 9223372036854775807, ) - for row in rows: - result["scanned_posts"] += 1 - post_id = str(row["post_id"]) - async with connection.transaction(): - complete = await post_content_is_complete( - connection, + for source_post_record in source_post_records: + backfill_summary["scanned_posts"] += 1 + post_id = str(source_post_record["post_id"]) + async with database_connection.transaction(): + post_content_complete = await post_content_is_complete( + database_connection, post_id, require_embedding=require_orchestrator_evidence, require_structure=require_orchestrator_evidence, ) - request = await ensure_post_content_job( - connection, + post_content_job_request = await ensure_post_content_job( + database_connection, post_id, - str(row["post_body"] or ""), - content_complete=complete, + str(source_post_record["post_body"] or ""), + content_complete=post_content_complete, ) - if complete and not request.should_publish: - result["already_complete"] += 1 + if post_content_complete and not post_content_job_request.should_publish: + backfill_summary["already_complete"] += 1 continue - if request.should_publish: + if post_content_job_request.should_publish: entry_id = await publish_post_content_event( - client, + valkey_client, post_id=post_id, - source_body_digest=request.source_body_sha256, + source_body_digest=post_content_job_request.source_body_sha256, ) if entry_id is None: - raise RuntimeError(f"Valkey did not publish post-content job {post_id}") - result["published_events"] += 1 - result["queued_posts"] += 1 - return result + raise RuntimeError( + f"Valkey did not publish post-content job {post_id}" + ) + backfill_summary["published_events"] += 1 + backfill_summary["queued_posts"] += 1 + return backfill_summary finally: - await connection.close() - await client.aclose() + await database_connection.close() + await valkey_client.aclose() def main() -> None: - args = _parser().parse_args() - result = asyncio.run( + """Run the post-content queue backfill command.""" + command_arguments = _queue_backfill_parser().parse_args() + backfill_summary = asyncio.run( queue_post_content_backfill( - args.target_dsn, - args.valkey_url, - limit=None if args.all else args.limit, + command_arguments.target_dsn, + command_arguments.valkey_url, + post_limit=None if command_arguments.all else command_arguments.limit, ) ) - print(result) + print(backfill_summary) if __name__ == "__main__": diff --git a/scripts/sync_occupational_construct_catalog.py b/scripts/sync_occupational_construct_catalog.py index 412d8d484..4ac62f8cd 100644 --- a/scripts/sync_occupational_construct_catalog.py +++ b/scripts/sync_occupational_construct_catalog.py @@ -21,18 +21,18 @@ ) -def _parser() -> argparse.ArgumentParser: +def _catalog_sync_parser() -> argparse.ArgumentParser: """Build the operator-only catalog synchronization parser.""" - parser = argparse.ArgumentParser( + catalog_sync_parser = argparse.ArgumentParser( description="Synchronize the governed O*NET occupational construct catalog." ) - parser.add_argument("--target-dsn") - return parser + catalog_sync_parser.add_argument("--target-dsn") + return catalog_sync_parser -async def synchronize_catalog(target_dsn: str) -> int: +async def synchronize_occupational_construct_catalog(target_dsn: str) -> int: """Download the fixed release and persist it without exposing credentials.""" - payload = await asyncio.to_thread( + catalog_payload = await asyncio.to_thread( get_json, ONET_CONTENT_MODEL_URL, timeout=30.0, @@ -40,19 +40,23 @@ async def synchronize_catalog(target_dsn: str) -> int: maximum_response_bytes=8 * 1024 * 1024, expected_response_media_type="application/json", ) - conn = await asyncpg.connect(target_dsn) + database_connection = await asyncpg.connect(target_dsn) try: - return await sync_onet_construct_catalog(conn, payload) + return await sync_onet_construct_catalog(database_connection, catalog_payload) finally: - await conn.close() + await database_connection.close() def main() -> None: """Parse configuration, synchronize the catalog, and print only its count.""" - args = _parser().parse_args() - settings = load_settings() - count = asyncio.run(synchronize_catalog(args.target_dsn or settings.database_url)) - print({"release": "31.0", "construct_count": count}) + command_arguments = _catalog_sync_parser().parse_args() + runtime_settings = load_settings() + synchronized_construct_count = asyncio.run( + synchronize_occupational_construct_catalog( + command_arguments.target_dsn or runtime_settings.database_url + ) + ) + print({"release": "31.0", "construct_count": synchronized_construct_count}) if __name__ == "__main__": diff --git a/tests/test_backfill_post_content_semantic_identifiers.py b/tests/test_backfill_post_content_semantic_identifiers.py new file mode 100644 index 000000000..9b7bac80d --- /dev/null +++ b/tests/test_backfill_post_content_semantic_identifiers.py @@ -0,0 +1,103 @@ +"""Naming and boundary contracts for the synchronous post-content backfill.""" + +from __future__ import annotations + +import ast +from pathlib import Path + +SCRIPT_PATH = Path("scripts/backfill_post_content.py") + + +def test_post_content_backfill_uses_semantic_owned_identifiers() -> None: + """Keep command, database, record, image, and result names specific.""" + script_source = SCRIPT_PATH.read_text(encoding="utf-8") + syntax_tree = ast.parse(script_source) + owned_identifiers = { + syntax_node.id + for syntax_node in ast.walk(syntax_tree) + if isinstance(syntax_node, ast.Name) + } + owned_identifiers.update( + syntax_node.arg + for syntax_node in ast.walk(syntax_tree) + if isinstance(syntax_node, ast.arg) + ) + owned_identifiers.update( + syntax_node.name + for syntax_node in ast.walk(syntax_tree) + if isinstance(syntax_node, (ast.AsyncFunctionDef, ast.FunctionDef)) + ) + + assert owned_identifiers.isdisjoint( + { + "_parser", + "args", + "conn", + "item", + "limit", + "parser", + "result", + "row", + "rows", + } + ) + assert { + "_post_content_backfill_parser", + "argument_parser", + "backfill_summary", + "command_arguments", + "database_connection", + "described_image_count", + "image_result", + "normalized_post_content", + "post_limit", + "selected_post_record", + "selected_post_records", + "source_post_record", + } <= owned_identifiers + + +def test_post_content_backfill_preserves_operator_contracts() -> None: + """Keep CLI, aggregate output, persistence, and close contracts stable.""" + script_source = SCRIPT_PATH.read_text(encoding="utf-8") + + for contract_literal in ( + '"--target-dsn"', + '"--post-id"', + '"--limit"', + '"--all"', + '"--normalize-only"', + '"requested_posts"', + '"selected_posts"', + '"processed_posts"', + '"described_posts"', + '"described_images"', + '"described_regions"', + '"embedding_rows"', + '"skipped_posts"', + "from source_post", + "post_content_unit", + "post_content_embedding", + "persist_post_content(", + "record_post_content_backfill_success(", + "await database_connection.close()", + ): + assert contract_literal in script_source + + for generic_sql_alias in ( + "source_post post", + "source_post real_post", + "post_content_unit unit", + "post_content_embedding embedding", + "corporate_entity entity", + ): + assert generic_sql_alias not in script_source + + for semantic_sql_alias in ( + "source_post source_record", + "source_post attributed_post", + "post_content_unit content_unit", + "post_content_embedding content_embedding", + "corporate_entity owning_entity", + ): + assert semantic_sql_alias in script_source diff --git a/tests/test_backfill_post_keyman_semantic_identifiers.py b/tests/test_backfill_post_keyman_semantic_identifiers.py new file mode 100644 index 000000000..02dd8832c --- /dev/null +++ b/tests/test_backfill_post_keyman_semantic_identifiers.py @@ -0,0 +1,84 @@ +"""Naming contract for the bounded post-Keyman operator command.""" + +from __future__ import annotations + +import ast +from pathlib import Path + + +SCRIPT_PATH = Path(__file__).parents[1] / "scripts" / "backfill_post_keymen.py" + + +def test_post_keyman_backfill_uses_bounded_context_identifiers() -> None: + """Keep owned command, database, record, and result names semantic.""" + syntax_tree = ast.parse(SCRIPT_PATH.read_text(encoding="utf-8")) + owned_identifiers = { + syntax_node.id + for syntax_node in ast.walk(syntax_tree) + if isinstance(syntax_node, ast.Name) + } + owned_identifiers.update( + syntax_node.arg + for syntax_node in ast.walk(syntax_tree) + if isinstance(syntax_node, ast.arg) + ) + owned_identifiers.update( + syntax_node.name + for syntax_node in ast.walk(syntax_tree) + if isinstance(syntax_node, (ast.AsyncFunctionDef, ast.FunctionDef)) + ) + + forbidden_generic_identifiers = { + "_run", + "args", + "conn", + "exc", + "failures", + "limit", + "mention_count", + "name", + "names", + "normalized", + "parser", + "pool", + "processed", + "row", + "rows", + "selector", + "settings", + } + assert owned_identifiers.isdisjoint(forbidden_generic_identifiers) + assert { + "_run_post_keyman_backfill", + "command_arguments", + "database_connection", + "database_pool", + "post_records", + "processed_post_count", + "runtime_settings", + } <= owned_identifiers + + +def test_post_keyman_backfill_preserves_operator_contract() -> None: + """Keep released CLI flags and JSON result fields at the adapter boundary.""" + script_source = SCRIPT_PATH.read_text(encoding="utf-8") + for contract_literal in ( + '"--post-id"', + '"--all"', + '"--limit"', + '"--post-timeout"', + '"failed_posts"', + '"failure_types"', + '"mentions_persisted"', + '"processed_posts"', + '"requested_posts"', + ): + assert contract_literal in script_source + + syntax_tree = ast.parse(script_source) + asynchronous_entrypoint_calls = { + syntax_node.func.id + for syntax_node in ast.walk(syntax_tree) + if isinstance(syntax_node, ast.Call) and isinstance(syntax_node.func, ast.Name) + } + assert "_run_post_keyman_backfill" in asynchronous_entrypoint_calls diff --git a/tests/test_backfill_thread_group_keys.py b/tests/test_backfill_thread_group_keys.py index 8b83ad5d9..e7139363e 100644 --- a/tests/test_backfill_thread_group_keys.py +++ b/tests/test_backfill_thread_group_keys.py @@ -34,7 +34,9 @@ class _Connection: whose live scope match the rewrite would orphan. """ - def __init__(self, rows: list[bool], anchored_runs: list[str] | None = None) -> None: + def __init__( + self, rows: list[bool], anchored_runs: list[str] | None = None + ) -> None: self._rows = rows self._anchored_runs = anchored_runs or [] self.executed: list[str] = [] @@ -50,7 +52,9 @@ async def fetch(self, query: str, *args: object): {"analysis_run_id": run_id, "scope_key": f"key-{run_id}"} for run_id in self._anchored_runs ] - return [{"had_project_code": had_project_code} for had_project_code in self._rows] + return [ + {"had_project_code": had_project_code} for had_project_code in self._rows + ] def test_backfill_clears_placeholders_and_routes_project_codes_to_secondary() -> None: @@ -72,14 +76,19 @@ def test_backfill_clears_placeholders_and_routes_project_codes_to_secondary() -> # related posts that lack a project code, exactly the links the # reconstruction library exists to find. assert "thread_group_key = ''" in update - assert "secondary_grouping_key = coalesce(nullif(btrim(source_project_code), ''), '')" in update + assert ( + "secondary_grouping_key = coalesce(nullif(btrim(source_project_code), ''), '')" + in update + ) assert "source_thread_group_key = coalesce(" in update assert "source_thread_group_key, thread_group_key" in update assert "source_secondary_grouping_key = coalesce(" in update assert "source_secondary_grouping_key, secondary_grouping_key" in update -def test_backfill_fails_closed_when_a_thread_group_scoped_run_would_be_orphaned() -> None: +def test_backfill_fails_closed_when_a_thread_group_scoped_run_would_be_orphaned() -> ( + None +): # analysis_scope_thread_group runs resolve `thread_group_key = # scope_key` live on every read (ABAC visibility) -- their member # posts are snapshot-frozen but the scope match is not. Rewriting @@ -110,8 +119,8 @@ def test_dry_run_reports_counts_but_raises_to_force_a_rollback() -> None: try: asyncio.run(backfill.backfill_thread_group_keys(conn, dry_run=True)) except backfill._RollbackDryRun as rolled_back: - assert rolled_back.project_evidence == 1 - assert rolled_back.cleared == 2 + assert rolled_back.project_evidence_post_count == 1 + assert rolled_back.cleared_post_count == 2 else: raise AssertionError("expected _RollbackDryRun") @@ -139,12 +148,16 @@ def fake_load_settings(): monkeypatch.setattr(backfill, "load_settings", fake_load_settings) -def test_run_reports_dry_run_counts_without_the_internal_exception_leaking(monkeypatch) -> None: +def test_run_reports_dry_run_counts_without_the_internal_exception_leaking( + monkeypatch, +) -> None: import argparse conn = _Connection([True, True, False]) _patch_pool(monkeypatch, conn) - result = asyncio.run(backfill._run(argparse.Namespace(dry_run=True))) + result = asyncio.run( + backfill._run_thread_group_key_backfill(argparse.Namespace(dry_run=True)) + ) assert result == { "cleared_placeholder_posts": 3, "project_secondary_evidence_posts": 2, @@ -157,7 +170,9 @@ def test_run_reports_write_counts_when_not_a_dry_run(monkeypatch) -> None: conn = _Connection([True, False, False]) _patch_pool(monkeypatch, conn) - result = asyncio.run(backfill._run(argparse.Namespace(dry_run=False))) + result = asyncio.run( + backfill._run_thread_group_key_backfill(argparse.Namespace(dry_run=False)) + ) assert result == { "cleared_placeholder_posts": 3, "project_secondary_evidence_posts": 1, diff --git a/tests/test_channel_weight_estimation_semantic_identifiers.py b/tests/test_channel_weight_estimation_semantic_identifiers.py new file mode 100644 index 000000000..108fdadb8 --- /dev/null +++ b/tests/test_channel_weight_estimation_semantic_identifiers.py @@ -0,0 +1,122 @@ +"""Naming and boundary contracts for deterministic channel-weight estimation.""" + +from __future__ import annotations + +import ast +from pathlib import Path + + +SCRIPT_PATH = Path("scripts/estimate_channel_weights.py") + + +def _function_identifiers(function_name: str) -> set[str]: + source_tree = ast.parse(SCRIPT_PATH.read_text(encoding="utf-8")) + function_node = next( + node + for node in ast.walk(source_tree) + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + and node.name == function_name + ) + return { + identifier + for node in ast.walk(function_node) + for identifier in ( + [node.id] + if isinstance(node, ast.Name) + else [node.arg] + if isinstance(node, ast.arg) + else [] + ) + } + + +def test_owned_estimation_identifiers_are_semantic() -> None: + expected_identifiers = { + "source_snapshot_digest": { + "source_post_rows", + "source_post_row", + "digest_material", + }, + "sample_pair_scores": { + "lineage_records", + "candidate_window", + "channel_groups", + "lineage_record", + "ordered_records", + "candidate_record", + }, + "subsample_stride": { + "sample_pair_total", + "sample_pair_limit", + "sample_stride", + }, + "persist_estimate": { + "database_connection", + "channel_weight_estimate", + "installed_estimator_version", + "channel_code", + "weight_value", + }, + "_run_channel_weight_estimation": { + "command_arguments", + "runtime_settings", + "database_connection", + "source_post_rows", + "lineage_records", + "channel_weight_estimate", + }, + } + forbidden_identifiers = { + "args", + "candidate", + "channel", + "conn", + "estimate", + "groups", + "limit", + "material", + "record", + "records", + "row", + "rows", + "settings", + "stride", + "total", + "version", + "weight", + "window", + } + + for function_name, required_identifiers in expected_identifiers.items(): + function_identifiers = _function_identifiers(function_name) + assert required_identifiers <= function_identifiers + assert function_identifiers.isdisjoint(forbidden_identifiers) + + +def test_external_cli_json_and_persistence_contracts_are_unchanged() -> None: + script_source = SCRIPT_PATH.read_text(encoding="utf-8") + + assert '"--post-limit"' in script_source + assert '"--dry-run"' in script_source + for result_key in ( + "weights", + "channel_set_code", + "sample_pair_count", + "estimation_method_code", + "anchor_method_code", + "estimation_run_id", + "source_snapshot_sha256", + "knowledge_cutoff", + "persisted", + "activation", + ): + assert f'"{result_key}"' in script_source + assert "insert into lineage_channel_weight" in script_source + assert ( + "delete from lineage_channel_weight where channel_set_code = $1" + in script_source + ) + assert ( + "asyncio.run(_run_channel_weight_estimation(command_arguments))" + in script_source + ) diff --git a/tests/test_estimate_channel_weights_script.py b/tests/test_estimate_channel_weights_script.py index 3b086524d..19e46ab48 100644 --- a/tests/test_estimate_channel_weights_script.py +++ b/tests/test_estimate_channel_weights_script.py @@ -33,12 +33,14 @@ def _record(record_id: str, group: str, minute: int, secondary: str = "") -> Rec def test_sampling_stays_within_groups_and_window() -> None: - records = [ + lineage_records = [ _record("a1", "g-a", 0), _record("a2", "g-a", 1), _record("b1", "g-b", 2), ] - pair_scores, group_ids, pair_labels = script.sample_pair_scores(records, window=50) + pair_scores, group_ids, pair_labels = script.sample_pair_scores( + lineage_records, candidate_window=50 + ) # Only a1->a2 pairs up; b1 is alone in its group and never crosses. assert len(pair_scores) == 1 assert group_ids == [0] @@ -49,10 +51,12 @@ def test_sampling_stays_within_groups_and_window() -> None: def test_sampling_window_bounds_candidates_like_reconstruct() -> None: - records = [_record(f"r{index}", "g", index) for index in range(5)] - _, unbounded_ids, _ = script.sample_pair_scores(records, window=50) + lineage_records = [_record(f"r{index}", "g", index) for index in range(5)] + _, unbounded_ids, _ = script.sample_pair_scores( + lineage_records, candidate_window=50 + ) assert len(unbounded_ids) == 4 + 3 + 2 + 1 - pair_scores, _, _ = script.sample_pair_scores(records, window=2) + pair_scores, _, _ = script.sample_pair_scores(lineage_records, candidate_window=2) # Each record sees at most its two immediate predecessors. assert len(pair_scores) == 1 + 2 + 2 + 2 @@ -71,13 +75,13 @@ def test_llm_subsample_stride_is_deterministic_and_spread() -> None: def test_snapshot_digest_is_reproducible_and_order_sensitive() -> None: - rows = [ + source_post_rows = [ {"post_id": "a", "created_at": datetime(2026, 1, 1, tzinfo=timezone.utc)}, {"post_id": "b", "created_at": datetime(2026, 1, 2, tzinfo=timezone.utc)}, ] - first = script.source_snapshot_digest(rows) - assert first == script.source_snapshot_digest(list(rows)) - assert first != script.source_snapshot_digest(list(reversed(rows))) + first = script.source_snapshot_digest(source_post_rows) + assert first == script.source_snapshot_digest(list(source_post_rows)) + assert first != script.source_snapshot_digest(list(reversed(source_post_rows))) assert len(first) == 64 @@ -94,39 +98,44 @@ async def execute(self, query: str, *args: object) -> str: return "OK" -def test_persist_estimate_stamps_full_provenance_on_one_scoped_set() -> None: - conn = _Connection() - estimate = ChannelWeightEstimate( +def test_persist_estimate_stamps_full_provenance_on_one_scoped_set( + monkeypatch, +) -> None: + monkeypatch.setattr(script, "estimator_version", lambda: "0.9.1") + database_connection = _Connection() + channel_weight_estimate = ChannelWeightEstimate( weights={"temporal": 0.25, "text": 0.75}, sample_pair_count=600, estimation_method_code="mls2plm_expected_information", ) - cutoff = datetime(2026, 1, 2, tzinfo=timezone.utc) - run_id = asyncio.run( + knowledge_cutoff = datetime(2026, 1, 2, tzinfo=timezone.utc) + estimation_run_id = asyncio.run( script.persist_estimate( - conn, - estimate, + database_connection, + channel_weight_estimate, channel_set_code=script.DETERMINISTIC_SET_CODE, snapshot_sha256="a" * 64, - knowledge_cutoff=cutoff, + knowledge_cutoff=knowledge_cutoff, ) ) - delete_query, delete_args = conn.executed[0] + delete_query, delete_args = database_connection.executed[0] # Scoped delete: persisting the deterministic set must never wipe # another set -- each active-channel combination owns its own rows. - assert "delete from lineage_channel_weight where channel_set_code = $1" in delete_query + assert ( + "delete from lineage_channel_weight where channel_set_code = $1" in delete_query + ) assert delete_args == (script.DETERMINISTIC_SET_CODE,) - inserted = {call[1][1]: call[1] for call in conn.executed[1:]} + inserted = {call[1][1]: call[1] for call in database_connection.executed[1:]} assert set(inserted) == {"temporal", "text"} - for row in inserted.values(): - assert row[0] == script.DETERMINISTIC_SET_CODE - assert row[3] == run_id - assert row[4] == "mls2plm_expected_information" - assert isinstance(row[5], str) and row[5].strip() - assert row[6] == script.UNANCHORED_METHOD_CODE - assert row[7] == "a" * 64 - assert row[8] == 600 - assert row[9] == cutoff + for persisted_row in inserted.values(): + assert persisted_row[0] == script.DETERMINISTIC_SET_CODE + assert persisted_row[3] == estimation_run_id + assert persisted_row[4] == "mls2plm_expected_information" + assert isinstance(persisted_row[5], str) and persisted_row[5].strip() + assert persisted_row[6] == script.UNANCHORED_METHOD_CODE + assert persisted_row[7] == "a" * 64 + assert persisted_row[8] == 600 + assert persisted_row[9] == knowledge_cutoff assert inserted["text"][2] == 0.75 diff --git a/tests/test_estimate_llm_channel_weights_script.py b/tests/test_estimate_llm_channel_weights_script.py index 210c95d84..44b1bcdb4 100644 --- a/tests/test_estimate_llm_channel_weights_script.py +++ b/tests/test_estimate_llm_channel_weights_script.py @@ -17,21 +17,24 @@ def test_batch_requests_carry_caller_custom_ids_for_every_pair() -> None: - labels = [("a", "b"), ("c", "d"), ("e", "f")] - requests = script.batch_requests_for_pairs([0, 2], labels) - assert [request["custom_id"] for request in requests] == ["pair-0", "pair-2"] + candidate_pair_labels = [("a", "b"), ("c", "d"), ("e", "f")] + batch_requests = script.batch_requests_for_pairs([0, 2], candidate_pair_labels) + assert [batch_request["custom_id"] for batch_request in batch_requests] == [ + "pair-0", + "pair-2", + ] # Never mix caller ids with generated ids in one batch (upstream # guidance on contextual-orchestrator #832): every request has one. - assert all("custom_id" in request for request in requests) - assert requests[0]["messages"][0]["content"] == judge_prompt("a", "b") - assert requests[1]["messages"][0]["content"] == judge_prompt("e", "f") - assert all(request["mode"] == "auto" for request in requests) + assert all("custom_id" in batch_request for batch_request in batch_requests) + assert batch_requests[0]["messages"][0]["content"] == judge_prompt("a", "b") + assert batch_requests[1]["messages"][0]["content"] == judge_prompt("e", "f") + assert all(batch_request["mode"] == "auto" for batch_request in batch_requests) def test_shared_judge_prompt_and_confidence_parse_round_trip() -> None: - prompt = judge_prompt("Record about pricing", "Follow-up record") - assert "Record A: Record about pricing" in prompt - assert "Record B: Follow-up record" in prompt + judgment_prompt = judge_prompt("Record about pricing", "Follow-up record") + assert "Record A: Record about pricing" in judgment_prompt + assert "Record B: Follow-up record" in judgment_prompt assert parse_confidence("0.85") == 0.85 assert parse_confidence("confidence: 0.4 maybe") == 0.4 with pytest.raises(HttpClientError): @@ -44,7 +47,7 @@ def test_errored_judgments_stay_unjudged_instead_of_becoming_zero() -> None: 0.0 -- the pair stays unjudged and the incomplete-run path reports it. Mapping is by custom_id only; foreign or malformed ids are ignored. """ - updates = script.judgment_updates_from_results( + judgment_updates = script.judgment_updates_from_results( [ {"custom_id": "pair-3", "answer": "0.7"}, {"custom_id": "pair-4", "answer": ""}, @@ -54,7 +57,7 @@ def test_errored_judgments_stay_unjudged_instead_of_becoming_zero() -> None: {"custom_id": "pair-not-a-number", "answer": "0.9"}, ] ) - assert updates == [(3, 0.7), (6, 0.0)] + assert judgment_updates == [(3, 0.7), (6, 0.0)] def test_batch_completion_is_detected_from_flag_or_status() -> None: diff --git a/tests/test_llm_channel_weight_estimation_semantic_identifiers.py b/tests/test_llm_channel_weight_estimation_semantic_identifiers.py new file mode 100644 index 000000000..9ceb44358 --- /dev/null +++ b/tests/test_llm_channel_weight_estimation_semantic_identifiers.py @@ -0,0 +1,111 @@ +"""Naming and boundary contracts for queued LLM channel-weight estimation.""" + +from __future__ import annotations + +import ast +from pathlib import Path + + +SCRIPT_PATH = Path("scripts/estimate_llm_channel_weights.py") + + +def test_owned_llm_estimation_identifiers_are_semantic() -> None: + script_source = SCRIPT_PATH.read_text(encoding="utf-8") + syntax_tree = ast.parse(script_source) + owned_identifiers = { + syntax_node.id + for syntax_node in ast.walk(syntax_tree) + if isinstance(syntax_node, ast.Name) + } + owned_identifiers.update( + syntax_node.arg + for syntax_node in ast.walk(syntax_tree) + if isinstance(syntax_node, ast.arg) + ) + owned_identifiers.update( + syntax_node.name + for syntax_node in ast.walk(syntax_tree) + if isinstance(syntax_node, (ast.AsyncFunctionDef, ast.FunctionDef)) + ) + + assert owned_identifiers.isdisjoint( + { + "_collect", + "_submit", + "args", + "chosen", + "collect", + "conn", + "estimate", + "exc", + "item", + "name", + "ordinal", + "pairs", + "parser", + "polled", + "result", + "results", + "retrieved", + "row", + "rows", + "run", + "score", + "scores", + "settings", + "submit", + "submitted", + "subcommands", + "unjudged", + "updates", + } + ) + assert { + "_collect_batch_estimation", + "_submit_batch_estimation", + "batch_result_record", + "batch_result_records", + "batch_results_payload", + "batch_status_payload", + "channel_weight_estimate", + "chosen_pair_ordinals", + "command_arguments", + "database_connection", + "estimation_run_record", + "judgment_updates", + "orchestrator_api_key", + "orchestrator_base_url", + "pair_judgment_records", + "pair_ordinal", + "runtime_settings", + "source_post_rows", + } <= owned_identifiers + + +def test_llm_provider_cli_json_and_persistence_contracts_are_unchanged() -> None: + script_source = SCRIPT_PATH.read_text(encoding="utf-8") + + for contract_literal in ( + '"custom_id"', + '"mode"', + '"messages"', + '"role"', + '"content"', + '"job_id"', + '"results"', + '"status"', + '"--post-limit"', + '"--pair-limit"', + '"--run-id"', + '"estimation_run_id"', + '"batch_job_id"', + '"sampled_pair_count"', + '"next_action"', + "insert into lineage_weight_estimation_run", + "insert into lineage_pair_judgment", + "update lineage_pair_judgment", + "update lineage_weight_estimation_run", + ): + assert contract_literal in script_source + assert "asyncio.run(_submit_batch_estimation(command_arguments))" in script_source + assert "asyncio.run(_collect_batch_estimation(command_arguments))" in script_source diff --git a/tests/test_queue_post_content_backfill_semantic_identifiers.py b/tests/test_queue_post_content_backfill_semantic_identifiers.py new file mode 100644 index 000000000..f4badde28 --- /dev/null +++ b/tests/test_queue_post_content_backfill_semantic_identifiers.py @@ -0,0 +1,84 @@ +"""Naming and boundary contracts for the post-content queue backfill.""" + +from __future__ import annotations + +import ast +from pathlib import Path + + +SCRIPT_PATH = Path("scripts/queue_post_content_backfill.py") + + +def test_queue_backfill_uses_semantic_owned_identifiers() -> None: + """Keep command, database, queue, record, and result names specific.""" + script_source = SCRIPT_PATH.read_text(encoding="utf-8") + syntax_tree = ast.parse(script_source) + owned_identifiers = { + syntax_node.id + for syntax_node in ast.walk(syntax_tree) + if isinstance(syntax_node, ast.Name) + } + owned_identifiers.update( + syntax_node.arg + for syntax_node in ast.walk(syntax_tree) + if isinstance(syntax_node, ast.arg) + ) + owned_identifiers.update( + syntax_node.name + for syntax_node in ast.walk(syntax_tree) + if isinstance(syntax_node, (ast.AsyncFunctionDef, ast.FunctionDef)) + ) + + assert owned_identifiers.isdisjoint( + { + "_parser", + "args", + "client", + "complete", + "connection", + "limit", + "parser", + "request", + "result", + "row", + "rows", + "settings", + } + ) + assert { + "_queue_backfill_parser", + "backfill_summary", + "command_arguments", + "database_connection", + "post_limit", + "post_content_complete", + "post_content_job_request", + "runtime_settings", + "source_post_record", + "source_post_records", + "valkey_client", + } <= owned_identifiers + + +def test_queue_backfill_preserves_cli_result_sql_and_publish_contracts() -> None: + """Keep public operator inputs, outputs, persistence, and event calls stable.""" + script_source = SCRIPT_PATH.read_text(encoding="utf-8") + + for contract_literal in ( + '"--target-dsn"', + '"--valkey-url"', + '"--limit"', + '"--all"', + '"scanned_posts"', + '"already_complete"', + '"queued_posts"', + '"published_events"', + "from source_post", + "post_content_unit", + "post_content_embedding", + "post_content_image_region_embedding", + "ensure_post_content_job(", + "publish_post_content_event(", + ): + assert contract_literal in script_source + assert "asyncio.run(\n queue_post_content_backfill(" in script_source diff --git a/tests/test_sync_occupational_construct_catalog_naming.py b/tests/test_sync_occupational_construct_catalog_naming.py new file mode 100644 index 000000000..e6a25cd16 --- /dev/null +++ b/tests/test_sync_occupational_construct_catalog_naming.py @@ -0,0 +1,68 @@ +"""Naming and boundary contracts for the occupational catalog synchronizer.""" + +from __future__ import annotations + +import ast +from pathlib import Path + + +SCRIPT_PATH = Path("scripts/sync_occupational_construct_catalog.py") + + +def test_catalog_sync_uses_semantic_owned_identifiers() -> None: + """Keep command, database, payload, and result names domain-specific.""" + script_source = SCRIPT_PATH.read_text(encoding="utf-8") + syntax_tree = ast.parse(script_source) + owned_identifiers = { + syntax_node.id + for syntax_node in ast.walk(syntax_tree) + if isinstance(syntax_node, ast.Name) + } + owned_identifiers.update( + syntax_node.arg + for syntax_node in ast.walk(syntax_tree) + if isinstance(syntax_node, ast.arg) + ) + owned_identifiers.update( + syntax_node.name + for syntax_node in ast.walk(syntax_tree) + if isinstance(syntax_node, (ast.AsyncFunctionDef, ast.FunctionDef)) + ) + + assert owned_identifiers.isdisjoint( + { + "_parser", + "args", + "conn", + "count", + "payload", + "settings", + "synchronize_catalog", + } + ) + assert { + "_catalog_sync_parser", + "catalog_payload", + "command_arguments", + "database_connection", + "runtime_settings", + "synchronize_occupational_construct_catalog", + "synchronized_construct_count", + } <= owned_identifiers + + +def test_catalog_sync_preserves_cli_and_output_contracts() -> None: + """Keep the operator flag, release, and result keys at the boundary.""" + script_source = SCRIPT_PATH.read_text(encoding="utf-8") + + for contract_literal in ( + '"--target-dsn"', + '"release"', + '"31.0"', + '"construct_count"', + ): + assert contract_literal in script_source + assert ( + "asyncio.run(\n" + " synchronize_occupational_construct_catalog(" in script_source + ) diff --git a/tests/test_thread_group_key_backfill_semantic_identifiers.py b/tests/test_thread_group_key_backfill_semantic_identifiers.py new file mode 100644 index 000000000..4b2801958 --- /dev/null +++ b/tests/test_thread_group_key_backfill_semantic_identifiers.py @@ -0,0 +1,75 @@ +"""Naming contract for the bounded thread-group-key backfill command.""" + +from __future__ import annotations + +import ast +from pathlib import Path + + +SCRIPT_PATH = Path(__file__).parents[1] / "scripts" / "backfill_thread_group_keys.py" + + +def test_thread_group_key_backfill_uses_semantic_identifiers() -> None: + """Keep owned command, database, record, and count names semantic.""" + script_source = SCRIPT_PATH.read_text(encoding="utf-8") + syntax_tree = ast.parse(script_source) + owned_identifiers = { + syntax_node.id + for syntax_node in ast.walk(syntax_tree) + if isinstance(syntax_node, ast.Name) + } + owned_identifiers.update( + syntax_node.arg + for syntax_node in ast.walk(syntax_tree) + if isinstance(syntax_node, ast.arg) + ) + owned_identifiers.update( + syntax_node.name + for syntax_node in ast.walk(syntax_tree) + if isinstance(syntax_node, (ast.AsyncFunctionDef, ast.FunctionDef)) + ) + + assert owned_identifiers.isdisjoint( + { + "_run", + "args", + "cleared", + "conn", + "counts", + "parser", + "pool", + "project_evidence", + "row", + "rows", + "settings", + } + ) + assert { + "_run_thread_group_key_backfill", + "analysis_run_ids", + "command_arguments", + "database_connection", + "database_pool", + "runtime_settings", + "updated_post_records", + } <= owned_identifiers + + +def test_thread_group_key_backfill_preserves_operator_contract() -> None: + """Keep the CLI flag and aggregate JSON keys at the adapter boundary.""" + script_source = SCRIPT_PATH.read_text(encoding="utf-8") + for contract_literal in ( + '"--dry-run"', + '"cleared_placeholder_posts"', + '"project_secondary_evidence_posts"', + '"dry_run"', + ): + assert contract_literal in script_source + + syntax_tree = ast.parse(script_source) + called_functions = { + syntax_node.func.id + for syntax_node in ast.walk(syntax_tree) + if isinstance(syntax_node, ast.Call) and isinstance(syntax_node.func, ast.Name) + } + assert "_run_thread_group_key_backfill" in called_functions