feat(models): integrate governed model configuration

backend/database/db_models.py

@@ -459,6 +459,18 @@ class ModelRecord(TableBase):
         String(100), doc="Source of the persisted capacity value. Optional values: operator, profile, provider_candidate, legacy, default, unknown.")
     capability_profile_version = Column(
         String(100), doc="Version of the approved provider/model capability profile used by the request, e.g. openai/gpt-4o@1.")
+    canonical_model_id = Column(
+        String(512), doc="Versioned canonical model identity used by profile matchers.")
+    capacity_field_metadata = Column(
+        JSONB, doc="Versioned field-level capacity provenance metadata without secrets.")
+    model_identity_metadata = Column(
+        JSONB, doc="Versioned canonical identity and capacity-match evidence.")
+    tokenizer_match_metadata = Column(
+        JSONB, doc="Versioned independent tokenizer match and conformance state.")
+    token_count_probe_metadata = Column(
+        JSONB, doc="Versioned sanitized Provider token-count probe state.")
+    feature_capability_metadata = Column(
+        JSONB, doc="Versioned reasoning and prompt-cache capability resolution without secrets.")
 
 
 class ModelMonitoringRecord(SimpleTableBase):
@@ -571,6 +583,9 @@ class ModelMonitoringRecord(SimpleTableBase):
     budget_warnings = Column(
         JSONB, doc="Structured W2 budget warnings active for this request"
     )
+    context_budget_evidence = Column(
+        JSONB, doc="Content-free P3 final request, compaction, overflow and recovery evidence"
+    )
     generation_rate = Column(
         Float, doc="Token generation rate (tokens per second)")
     is_streaming = Column(

backend/database/model_management_db.py

@@ -122,6 +122,55 @@ def delete_model_record(model_id: int, user_id: str, tenant_id: str) -> bool:
         return result.rowcount > 0
 
 
+def apply_model_mutations(
+    *,
+    creates: List[Dict[str, Any]],
+    updates: List[tuple[int, Dict[str, Any]]],
+    deletes: List[int],
+    user_id: str,
+    tenant_id: str,
+) -> None:
+    """Apply a validated model batch in one database transaction."""
+    with get_db_session() as session:
+        for model_data in creates:
+            cleaned = db_client.clean_string_values(model_data)
+            cleaned["create_time"] = func.current_timestamp()
+            cleaned["tenant_id"] = tenant_id
+            if user_id:
+                cleaned = add_creation_tracking(cleaned, user_id)
+            session.execute(insert(ModelRecord).values(cleaned))
+
+        for model_id, update_data in updates:
+            cleaned = db_client.clean_string_values(update_data)
+            cleaned["update_time"] = func.current_timestamp()
+            if user_id:
+                cleaned = add_update_tracking(cleaned, user_id)
+            session.execute(
+                update(ModelRecord)
+                .where(
+                    ModelRecord.model_id == model_id,
+                    ModelRecord.tenant_id == tenant_id,
+                )
+                .values(cleaned)
+            )
+
+        for model_id in deletes:
+            cleaned = {
+                "delete_flag": "Y",
+                "update_time": func.current_timestamp(),
+            }
+            if user_id:
+                cleaned = add_update_tracking(cleaned, user_id)
+            session.execute(
+                update(ModelRecord)
+                .where(
+                    ModelRecord.model_id == model_id,
+                    ModelRecord.tenant_id == tenant_id,
+                )
+                .values(cleaned)
+            )
+
+
 def get_model_records(filters: Optional[Dict[str, Any]], tenant_id: str) -> List[Dict[str, Any]]:
     """
     Get a list of model records