diff --git a/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/rc06_publication.py b/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/rc06_publication.py index f4eb83f4..ec573f8b 100644 --- a/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/rc06_publication.py +++ b/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/rc06_publication.py @@ -570,7 +570,11 @@ class Rc06PromptPublicationService: if transaction.state != "committed": raise Rc06Conflict(f"Prompt transaction cannot seal from {transaction.state}.") self._verify_committed(session, transaction, expected_runtime, expected_modules) - now = datetime.now(UTC) + now = _database_wall_clock(session) + if now < (_as_utc(transaction.committed_at) or now): + raise Rc06PublicationError( + "RC06 AI database clock moved behind the durable commit boundary." + ) transaction.state = "sealed" transaction.sealed_at = now transaction.updated_at = now diff --git a/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/schema_check.py b/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/schema_check.py index 5aab8771..78ea7a8b 100644 --- a/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/schema_check.py +++ b/yuqei-ai-platform/services/ai-platform-api/src/yuqei_ai_platform_api/schema_check.py @@ -180,6 +180,60 @@ RC06_CONSTRAINTS_BY_TABLE = { "ck_rc06_prompt_module_state", ), } +RC06_PRIMARY_KEYS_BY_TABLE = { + "prompt_templates": ("id",), + "rc06_prompt_publications": ("transaction_id",), + "rc06_prompt_module_publications": ("id",), +} +RC06_UNIQUE_COLUMNS_BY_TABLE = { + "prompt_templates": { + "ux_prompt_templates_name_version": ("name", "version"), + }, + "rc06_prompt_publications": { + "ux_rc06_prompt_publications_idempotency": ("idempotency_key",), + }, + "rc06_prompt_module_publications": { + "ux_rc06_prompt_module_publications_transaction_key": ( + "transaction_id", + "module_key", + ), + }, +} +RC06_ENUM_CHECKS_BY_TABLE = { + "rc06_prompt_publications": { + "ck_rc06_prompt_publications_state": ( + "state", + frozenset({ + "prepared", + "committed", + "sealed", + "compensated", + "compensation_failed", + }), + ), + }, + "rc06_prompt_module_publications": { + "ck_rc06_prompt_module_category": ( + "category", + frozenset({"drafting", "review"}), + ), + "ck_rc06_prompt_module_state": ( + "state", + frozenset({"prepared", "applied", "compensated"}), + ), + }, +} +RC06_FOREIGN_KEYS_BY_TABLE = { + "rc06_prompt_module_publications": { + ( + ("transaction_id",), + "rc06_prompt_publications", + ("transaction_id",), + "__current__", + "NO ACTION", + ), + }, +} class RuntimeSchemaCheck(BaseModel): @@ -662,6 +716,178 @@ def _check_expected_columns( ) +def _strip_outer_sql_parentheses(value: str) -> str: + result = value.strip() + while result.startswith("(") and result.endswith(")"): + depth = 0 + quoted = False + outer_closes_at_end = False + index = 0 + while index < len(result): + character = result[index] + if character == "'": + if quoted and index + 1 < len(result) and result[index + 1] == "'": + index += 2 + continue + quoted = not quoted + elif not quoted: + if character == "(": + depth += 1 + elif character == ")": + depth -= 1 + if depth == 0: + outer_closes_at_end = index == len(result) - 1 + break + index += 1 + if not outer_closes_at_end: + break + result = result[1:-1].strip() + return result + + +def _split_top_level_sql(value: str, operator: str) -> list[str]: + result: list[str] = [] + start = 0 + depth = 0 + quoted = False + index = 0 + lowered = value.lower() + while index < len(value): + character = value[index] + if character == "'": + if quoted and index + 1 < len(value) and value[index + 1] == "'": + index += 2 + continue + quoted = not quoted + index += 1 + continue + if quoted: + index += 1 + continue + if character == "(": + depth += 1 + elif character == ")": + depth -= 1 + elif depth == 0 and lowered.startswith(operator, index): + before = lowered[index - 1] if index else " " + after_index = index + len(operator) + after = lowered[after_index] if after_index < len(lowered) else " " + if not (before.isalnum() or before == "_") and not ( + after.isalnum() or after == "_" + ): + result.append(value[start:index].strip()) + start = after_index + index = after_index + continue + index += 1 + result.append(value[start:].strip()) + return result + + +def _flatten_sql_boolean(value: str, operator: str) -> list[str]: + stripped = _strip_outer_sql_parentheses(value) + parts = _split_top_level_sql(stripped, operator) + if len(parts) == 1: + return [stripped] + flattened: list[str] = [] + for part in parts: + flattened.extend(_flatten_sql_boolean(part, operator)) + return flattened + + +def _normalize_inspected_check(value: str) -> str: + normalized = value.lower().replace("<>", "!=") + normalized = re.sub(r'"([a-z_][a-z0-9_]*)"', r"\1", normalized) + normalized = re.sub(r"`([a-z_][a-z0-9_]*)`", r"\1", normalized) + normalized = re.sub(r"\[([a-z_][a-z0-9_]*)\]", r"\1", normalized) + normalized = re.sub( + r"::\s*(?:text|character\s+varying(?:\s*\(\s*\d+\s*\))?|" + r"integer|bigint)(?:\s*\[\s*\])?", + "", + normalized, + flags=re.IGNORECASE, + ) + return normalized.strip() + + +def _sql_atom(value: str) -> str: + return re.sub(r"[()\s]+", "", _strip_outer_sql_parentheses(value)) + + +def _enum_check_values(value: str, column: str) -> frozenset[str]: + sqltext = _normalize_inspected_check(value) + identifier = rf"\(*\s*{re.escape(column)}\s*\)*" + match = re.fullmatch( + rf"\s*\(*\s*{identifier}\s+in\s*\((?P[^)]*)\)\s*\)*\s*", + sqltext, + flags=re.IGNORECASE, + ) + pg_any = re.fullmatch( + rf"\s*\(*\s*{identifier}\s*=\s*any\s*" + r"\(\s*\(*\s*array\s*\[(?P[^\]]*)\]\s*\)*\s*\)\s*\)*\s*", + sqltext, + flags=re.IGNORECASE, + ) + literal_source = ( + match.group("values") + if match is not None + else pg_any.group("values") + if pg_any is not None + else "" + ) + sql_literal = r"'(?:''|[^'])*'" + if not re.fullmatch( + rf"\s*{sql_literal}(?:\s*,\s*{sql_literal})*\s*", + literal_source, + flags=re.IGNORECASE, + ): + return frozenset() + return frozenset( + item.replace("''", "'") + for item in re.findall(r"'((?:''|[^'])*)'", literal_source) + ) + + +def _valid_compensation_check(value: str) -> bool: + sqltext = _normalize_inspected_check(value) + or_parts = _flatten_sql_boolean(sqltext, "or") + if len(or_parts) != 2 or _sql_atom(or_parts[0]) != "state!='compensated'": + return False + clauses = _flatten_sql_boolean(or_parts[1], "and") + enum_clauses = [ + clause + for clause in clauses + if _enum_check_values(clause, "compensated_from_state") + == frozenset({"prepared", "committed"}) + ] + if len(enum_clauses) != 1: + return False + actual = { + _sql_atom(clause) + for clause in clauses + if clause != enum_clauses[0] + } + expected = { + "compensation_item_results_jsonisnotnull", + "runtime_snapshot_compensation_results_jsonisnotnull", + "compensation_evidence_sha256isnotnull", + "restored_item_countisnotnull", + "restored_item_count>=0", + "removed_item_countisnotnull", + "removed_item_count>=0", + "unchanged_item_countisnotnull", + "unchanged_item_count>=0", + "compensated_atisnotnull", + } + return len(clauses) == 11 and actual == expected + + +def _valid_authority_window_check(value: str) -> bool: + return _sql_atom(_normalize_inspected_check(value)) == ( + "authority_expires_at>created_at" + ) + + def _check_rc06_constraints(inspector, table_names: set[str], *, dialect: str) -> RuntimeSchemaCheck: schema = "public" if dialect == "postgresql" else None missing: list[str] = [] @@ -669,20 +895,70 @@ def _check_rc06_constraints(inspector, table_names: set[str], *, dialect: str) - if table_name not in table_names: missing.extend(f"{table_name}.{name}" for name in expected) continue - actual = { - str(item.get("name")) - for item in ( - inspector.get_unique_constraints(table_name, schema=schema) - + inspector.get_check_constraints(table_name, schema=schema) - + inspector.get_indexes(table_name, schema=schema) - ) + actual_unique = { + str(item.get("name")): tuple(item.get("column_names") or ()) + for item in inspector.get_unique_constraints(table_name, schema=schema) if item.get("name") } - missing.extend( - f"{table_name}.{name}" - for name in expected - if name not in actual + actual_checks = { + str(item.get("name")): str(item.get("sqltext") or "") + for item in inspector.get_check_constraints(table_name, schema=schema) + if item.get("name") + } + actual_primary = tuple( + inspector.get_pk_constraint(table_name, schema=schema).get( + "constrained_columns" + ) + or () ) + if actual_primary != RC06_PRIMARY_KEYS_BY_TABLE[table_name]: + missing.append(f"{table_name}.primary_key") + for name, columns in RC06_UNIQUE_COLUMNS_BY_TABLE.get(table_name, {}).items(): + if actual_unique.get(name) != columns: + missing.append(f"{table_name}.{name}") + for name, (column, values) in RC06_ENUM_CHECKS_BY_TABLE.get( + table_name, + {}, + ).items(): + if _enum_check_values(actual_checks.get(name, ""), column) != values: + missing.append(f"{table_name}.{name}") + if table_name == "rc06_prompt_publications": + if not _valid_compensation_check(actual_checks.get( + "ck_rc06_prompt_publications_compensation_evidence", + "", + )): + missing.append( + f"{table_name}.ck_rc06_prompt_publications_compensation_evidence" + ) + if not _valid_authority_window_check(actual_checks.get( + "ck_rc06_prompt_publications_authority_window", + "", + )): + missing.append( + f"{table_name}.ck_rc06_prompt_publications_authority_window" + ) + expected_foreign = RC06_FOREIGN_KEYS_BY_TABLE.get(table_name, set()) + if expected_foreign: + actual_foreign = { + ( + tuple(item.get("constrained_columns") or ()), + str(item.get("referred_table") or ""), + tuple(item.get("referred_columns") or ()), + ( + "__current__" + if item.get("referred_schema") in {None, "", schema} + else str(item.get("referred_schema")) + ), + str( + (item.get("options") or {}).get("ondelete") + or item.get("ondelete") + or "NO ACTION" + ).upper().replace("_", " "), + ) + for item in inspector.get_foreign_keys(table_name, schema=schema) + } + if actual_foreign != expected_foreign: + missing.append(f"{table_name}.foreign_key") if missing: return RuntimeSchemaCheck( name="rc06_constraints", @@ -694,7 +970,13 @@ def _check_rc06_constraints(inspector, table_names: set[str], *, dialect: str) - name="rc06_constraints", status="ok", message="AI Platform RC06 integrity constraints are present.", - details={"checked_constraints": sum(map(len, RC06_CONSTRAINTS_BY_TABLE.values()))}, + details={ + "checked_constraints": ( + sum(map(len, RC06_CONSTRAINTS_BY_TABLE.values())) + + len(RC06_PRIMARY_KEYS_BY_TABLE) + + sum(map(len, RC06_FOREIGN_KEYS_BY_TABLE.values())) + ) + }, ) diff --git a/yuqei-ai-platform/services/ai-platform-api/tests/test_health.py b/yuqei-ai-platform/services/ai-platform-api/tests/test_health.py index 49222aab..03e9cda2 100644 --- a/yuqei-ai-platform/services/ai-platform-api/tests/test_health.py +++ b/yuqei-ai-platform/services/ai-platform-api/tests/test_health.py @@ -1,12 +1,18 @@ from fastapi.testclient import TestClient +from sqlalchemy import inspect as sqlalchemy_inspect from yuqei_ai_platform_api.config import AiPlatformSettings from yuqei_ai_platform_api.main import create_app -from yuqei_ai_platform_api.repository import InMemoryAiPlatformRepository, ProviderConfigCreate +from yuqei_ai_platform_api.repository import ( + InMemoryAiPlatformRepository, + ProviderConfigCreate, + SqlAlchemyAiPlatformRepository, +) from yuqei_ai_platform_api.schema_check import ( RC06_COLUMNS_BY_TABLE, RC06_CONSTRAINTS_BY_TABLE, _check_provider_runtime, + _check_rc06_constraints, _check_rc06_publication_config, ) @@ -149,6 +155,101 @@ def test_rc06_schema_contract_includes_durable_compensation_evidence() -> None: ) +class _InspectorProxy: + def __init__(self, delegate, overrides): + self._delegate = delegate + self._overrides = overrides + + def __getattr__(self, name): + if name in self._overrides: + return self._overrides[name] + return getattr(self._delegate, name) + + +def test_rc06_schema_rejects_same_named_indexes_that_spoof_constraints(tmp_path) -> None: + repository = SqlAlchemyAiPlatformRepository( + f"sqlite+pysqlite:///{(tmp_path / 'spoofed-constraints.sqlite3').as_posix()}", + auto_create=True, + ) + try: + inspector = sqlalchemy_inspect(repository._engine) + names = set(inspector.get_table_names()) + spoofed_names = { + name + for expected in RC06_CONSTRAINTS_BY_TABLE.values() + for name in expected + } + proxy = _InspectorProxy(inspector, { + "get_unique_constraints": lambda *_args, **_kwargs: [], + "get_check_constraints": lambda *_args, **_kwargs: [], + "get_indexes": lambda *_args, **_kwargs: [ + {"name": name, "column_names": ["unrelated"], "unique": False} + for name in spoofed_names + ], + }) + + result = _check_rc06_constraints(proxy, names, dialect="sqlite") + + assert result.status == "error" + assert "ux_rc06_prompt_publications_idempotency" in result.details[ + "missing_constraints" + ] + assert "ck_rc06_prompt_publications_authority_window" in result.details[ + "missing_constraints" + ] + finally: + repository._engine.dispose() + + +def test_rc06_schema_rejects_tautological_check_and_cascade_fk(tmp_path) -> None: + repository = SqlAlchemyAiPlatformRepository( + f"sqlite+pysqlite:///{(tmp_path / 'drifted-constraints.sqlite3').as_posix()}", + auto_create=True, + ) + try: + inspector = sqlalchemy_inspect(repository._engine) + names = set(inspector.get_table_names()) + + def drifted_checks(table_name, **kwargs): + rows = [dict(item) for item in inspector.get_check_constraints( + table_name, + **kwargs, + )] + for row in rows: + if row.get("name") in { + "ck_rc06_prompt_publications_state", + "ck_rc06_prompt_publications_compensation_evidence", + "ck_rc06_prompt_publications_authority_window", + }: + row["sqltext"] = f"({row['sqltext']}) OR 1 = 1" + return rows + + def cascade_foreign_keys(table_name, **kwargs): + rows = [dict(item) for item in inspector.get_foreign_keys( + table_name, + **kwargs, + )] + for row in rows: + row["options"] = {"ondelete": "CASCADE"} + return rows + + proxy = _InspectorProxy(inspector, { + "get_check_constraints": drifted_checks, + "get_foreign_keys": cascade_foreign_keys, + }) + + result = _check_rc06_constraints(proxy, names, dialect="sqlite") + + assert result.status == "error" + missing = result.details["missing_constraints"] + assert "ck_rc06_prompt_publications_state" in missing + assert "ck_rc06_prompt_publications_compensation_evidence" in missing + assert "ck_rc06_prompt_publications_authority_window" in missing + assert "rc06_prompt_module_publications.foreign_key" in missing + finally: + repository._engine.dispose() + + def test_runtime_schema_check_errors_for_missing_core_tables(tmp_path) -> None: database_path = tmp_path / "empty-ai-platform.sqlite3" app = create_app( diff --git a/yuqei-ai-platform/services/ai-platform-api/tests/test_rc06_publication.py b/yuqei-ai-platform/services/ai-platform-api/tests/test_rc06_publication.py index af421ad7..55859884 100644 --- a/yuqei-ai-platform/services/ai-platform-api/tests/test_rc06_publication.py +++ b/yuqei-ai-platform/services/ai-platform-api/tests/test_rc06_publication.py @@ -203,6 +203,7 @@ def test_seal_after_expiry_recovers_proven_preexpiry_durable_commit( assert sealed.state == "sealed" assert sealed.committed_at == committed.committed_at + assert sealed.sealed_at == request.authority_expires_at + timedelta(days=1) assert service.seal( request.transaction_id, COMPENSATION_CAPABILITY,