diff --git a/awx/main/scheduler/task_manager.py b/awx/main/scheduler/task_manager.py
index f355554f61..70b2655eb5 100644
--- a/awx/main/scheduler/task_manager.py
+++ b/awx/main/scheduler/task_manager.py
@@ -28,7 +28,7 @@ from ansible_base.lib.utils.db import advisory_lock
 
 # AWX
 from awx.main.dispatch.reaper import reap_job
-from awx.main.utils.validation_bypass_observability import audit_bulk_update_instances
+from ansible_base.lib.utils.bulk_validation_audit import audit_bulk_model_instances
 from awx.main.models import (
     Instance,
     InventorySource,
@@ -667,7 +667,9 @@ class TaskManager(TaskBase):
                     )
             if not found_acceptable_queue:
                 self.task_needs_capacity(task, tasks_to_update_job_explanation)
-        audit_bulk_update_instances(tasks_to_update_job_explanation, ['job_explanation'])
+        audit_bulk_model_instances(
+            tasks_to_update_job_explanation, operation='bulk_update', update_fields=['job_explanation']
+        )
         UnifiedJob.objects.bulk_update(tasks_to_update_job_explanation, ['job_explanation'])
 
     def task_needs_capacity(self, task, tasks_to_update_job_explanation):
diff --git a/awx/main/tests/unit/utils/test_validation_bypass_observability.py b/awx/main/tests/unit/utils/test_validation_bypass_observability.py
index c3581e651d..7948469e61 100644
--- a/awx/main/tests/unit/utils/test_validation_bypass_observability.py
+++ b/awx/main/tests/unit/utils/test_validation_bypass_observability.py
@@ -10,7 +10,6 @@ from ansible_base.lib.utils.bulk_validation_audit import audit_bulk_model_instan
 from awx.main.models import Host, Inventory, Job, Organization, WorkflowJob, WorkflowJobNode
 from awx.main.utils.db import bulk_update_sorted_by_id
 from awx.main.utils.validation_bypass_observability import (
-    audit_bulk_update_instances,
     audit_workflow_job_nodes_for_bulk_create,
     configure_validation_bypass_observability,
 )
@@ -27,9 +26,15 @@ def test_configure_validation_bypass_observability_wires_dab(mocker):
     configure_validation_bypass_observability()
 
     allow.assert_called_once()
-    assert 'awx.api.serializers' in allow.call_args[0][0]
+    prefixes = allow.call_args[0][0]
+    assert 'awx.api.serializers' in prefixes
+    assert 'awx.main.scheduler' in prefixes
+    assert 'awx.main.utils' not in prefixes
     deny.assert_called_once()
-    assert 'awx.main.models' in deny.call_args[0][0]
+    internal = deny.call_args[0][0]
+    assert 'awx.main.models' in internal
+    assert 'awx.main.utils.validation_bypass_observability' in internal
+    assert 'awx.main.utils.db' in internal
 
 
 @pytest.mark.django_db
@@ -64,27 +69,27 @@ def test_audit_workflow_job_nodes_no_op_for_empty_list(caplog):
 
 
 @pytest.mark.django_db
-def test_audit_bulk_update_instances_logs_registered_text_field(caplog):
+def test_audit_bulk_model_instances_logs_registered_text_field_on_bulk_update(caplog):
     job = Job.objects.create(job_explanation='<script>x</script>')
     with caplog.at_level(logging.WARNING, logger=LOGGER):
-        audit_bulk_update_instances([job], ['job_explanation'])
+        audit_bulk_model_instances([job], operation='bulk_update', update_fields=['job_explanation'])
     assert 'ORM bypass (bulk_update)' in caplog.text
     assert 'job_explanation' in caplog.text
 
 
 @pytest.mark.django_db
-def test_audit_bulk_update_instances_skips_unregistered_workflow_job(caplog):
+def test_audit_bulk_model_instances_skips_unregistered_workflow_job_on_bulk_update(caplog):
     wj = WorkflowJob.objects.create(name='wf-job', job_explanation='<script>x</script>')
     with caplog.at_level(logging.WARNING, logger=LOGGER):
-        audit_bulk_update_instances([wj], ['job_explanation'])
+        audit_bulk_model_instances([wj], operation='bulk_update', update_fields=['job_explanation'])
     assert 'ORM bypass (bulk_update)' not in caplog.text
 
 
 @pytest.mark.django_db
-def test_audit_bulk_update_instances_no_op_when_fields_empty(caplog):
+def test_audit_bulk_model_instances_bulk_update_no_op_when_update_fields_empty(caplog):
     job = Job.objects.create(job_explanation='<script>x</script>')
     with caplog.at_level(logging.WARNING, logger=LOGGER):
-        audit_bulk_update_instances([job], [])
+        audit_bulk_model_instances([job], operation='bulk_update', update_fields=[])
     assert 'ORM bypass (bulk_update)' not in caplog.text
 
 
@@ -115,16 +120,16 @@ def test_task_manager_audits_before_job_explanation_bulk_update():
     """Regression: scheduler must not bulk_update job_explanation without audit."""
     task_manager_py = Path(__file__).resolve().parents[3] / 'scheduler' / 'task_manager.py'
     source = task_manager_py.read_text()
-    audit_marker = 'audit_bulk_update_instances(tasks_to_update_job_explanation'
+    audit_marker = "audit_bulk_model_instances(\n            tasks_to_update_job_explanation, operation='bulk_update', update_fields=['job_explanation']"
     bulk_marker = "UnifiedJob.objects.bulk_update(tasks_to_update_job_explanation, ['job_explanation'])"
     assert audit_marker in source
     assert bulk_marker in source
-    assert source.index(audit_marker) < source.index(bulk_marker)
+    assert source.index('audit_bulk_model_instances') < source.index(bulk_marker)
 
 
 def test_db_bulk_update_sorted_by_id_audits_before_orm_bulk_update():
     """Regression: shared bulk_update helper must audit before Django bulk_update."""
     db_py = Path(__file__).resolve().parents[3] / 'utils' / 'db.py'
     source = db_py.read_text()
-    assert 'audit_bulk_update_instances(sorted_objects, fields)' in source
-    assert source.index('audit_bulk_update_instances') < source.index('model.objects.bulk_update')
+    assert "audit_bulk_model_instances(sorted_objects, operation='bulk_update', update_fields=fields)" in source
+    assert source.index('audit_bulk_model_instances') < source.index('model.objects.bulk_update')
diff --git a/awx/main/utils/db.py b/awx/main/utils/db.py
index a4b0741888..893647059a 100644
--- a/awx/main/utils/db.py
+++ b/awx/main/utils/db.py
@@ -10,7 +10,7 @@ from awx import MODE
 from django.conf import settings
 from django.db import connection
 
-from awx.main.utils.validation_bypass_observability import audit_bulk_update_instances
+from ansible_base.lib.utils.bulk_validation_audit import audit_bulk_model_instances
 
 
 def set_connection_name(function):
@@ -30,6 +30,9 @@ def bulk_update_sorted_by_id(model, objects, fields, batch_size=1000):
     which helps avoid the row-level locking contention that can lead to deadlocks
     in PostgreSQL when multiple processes are involved.
 
+    When ``fields`` includes registered Char/Text columns, logs Tier 1/2 ORM bypass
+    violations (warnings only) via django-ansible-base before ``bulk_update``.
+
     Returns:
         int: The number of rows affected by the update.
     """
@@ -38,7 +41,7 @@ def bulk_update_sorted_by_id(model, objects, fields, batch_size=1000):
         return 0  # Return 0 when nothing is updated
 
     sorted_objects = sorted(objects, key=lambda obj: obj.id)
-    audit_bulk_update_instances(sorted_objects, fields)
+    audit_bulk_model_instances(sorted_objects, operation='bulk_update', update_fields=fields)
     return model.objects.bulk_update(sorted_objects, fields, batch_size=batch_size)
 
 
diff --git a/awx/main/utils/validation_bypass_observability.py b/awx/main/utils/validation_bypass_observability.py
index cb91ea880f..7725eb3277 100644
--- a/awx/main/utils/validation_bypass_observability.py
+++ b/awx/main/utils/validation_bypass_observability.py
@@ -1,20 +1,22 @@
-"""Register django-ansible-base ORM bypass caller attribution for AWX."""
+"""Register django-ansible-base ORM bypass observability for AWX.
+
+Startup registers DAB ``post_save`` bypass signals and AWX caller prefixes.
+Bulk ``bulk_create`` / ``bulk_update`` paths that skip ``post_save`` are audited at
+call sites via DAB helpers; workflow job node prompt pseudo-fields are handled here.
+"""
 
 from collections.abc import Iterable
 
 from django.db.models import Model
 
-from ansible_base.lib.utils.bulk_validation_audit import (
-    _log_bulk_violation,
-    audit_bulk_model_instances,
-)
+from ansible_base.lib.utils.bulk_validation_audit import audit_bulk_model_instances
 from ansible_base.lib.utils.validation_signals import (
     _get_caller_info,
-    _get_text_fields,
     _protected_models,
     _validate_field,
     extend_caller_allowlist_prefixes,
     extend_internal_caller_prefixes,
+    log_orm_bypass_violation,
     register_validation_signals,
 )
 
@@ -23,7 +25,14 @@ _WORKFLOW_JOB_NODE_PROMPT_PSEUDO_FIELDS = frozenset({'scm_branch', 'limit', 'job
 
 
 def configure_validation_bypass_observability() -> None:
-    """Call from AppConfig.ready() so ORM bypass logs show AWX entry points."""
+    """Call from AppConfig.ready() so ORM bypass logs show AWX entry points.
+
+    DAB resolves ``[caller: …]`` by walking the stack outward and returning the
+    first module on the caller allowlist (denylist is not applied in that phase).
+    Allowlist only product surfaces (API, serializers, tasks, management, scheduler).
+    List bulk-audit helpers under ``awx.main.utils`` on the internal denylist so
+    logs attribute to the serializer, task, or scheduler frame that invoked them.
+    """
     register_validation_signals()
     extend_caller_allowlist_prefixes(
         [
@@ -31,7 +40,7 @@ def configure_validation_bypass_observability() -> None:
             "awx.api.serializers",
             "awx.main.tasks",
             "awx.main.management",
-            "awx.main.utils",
+            "awx.main.scheduler",
         ]
     )
     extend_internal_caller_prefixes(
@@ -39,6 +48,8 @@ def configure_validation_bypass_observability() -> None:
             "awx.main.models",
             "awx.main.signals",
             "awx.main.dispatch",
+            "awx.main.utils.validation_bypass_observability",
+            "awx.main.utils.db",
         ]
     )
 
@@ -51,17 +62,18 @@ def audit_workflow_job_nodes_for_bulk_create(nodes: Iterable[Model]) -> None:
     ``NullablePromptPseudoField`` and must be read with ``getattr`` after deferred
     attrs are applied on the in-memory instances.
     """
-    audit_bulk_model_instances(nodes, operation='bulk_create')
-    if not nodes:
+    materialized = list(nodes)
+    if not materialized:
         return
-    sample = next(iter(nodes))
+    audit_bulk_model_instances(materialized, operation='bulk_create')
+    sample = materialized[0]
     protected = _protected_models.get(type(sample))
     if protected is None:
         return
     name_fields, excluded_fields = protected
-    caller_info = _get_caller_info()
     resource_type = f"{sample._meta.app_label}.{sample._meta.object_name}"
-    for instance in nodes:
+    caller_info = None
+    for instance in materialized:
         for field_name in _WORKFLOW_JOB_NODE_PROMPT_PSEUDO_FIELDS:
             if field_name in excluded_fields:
                 continue
@@ -71,45 +83,6 @@ def audit_workflow_job_nodes_for_bulk_create(nodes: Iterable[Model]) -> None:
             violation = _validate_field(field_name, value, name_fields)
             if violation:
                 tier, reason = violation
-                _log_bulk_violation('bulk_create', field_name, resource_type, tier, caller_info, reason)
-
-
-def _audited_text_field_names(model: type[Model]) -> frozenset[str]:
-    protected = _protected_models.get(model)
-    if protected is None:
-        return frozenset()
-    _, excluded_fields = protected
-    text_fields, _json_fields = _get_text_fields(model)
-    return frozenset(field_name for field_name in text_fields if field_name not in excluded_fields)
-
-
-def audit_bulk_update_instances(instances: Iterable[Model], fields: Iterable[str]) -> None:
-    """Log Tier 1/2 violations for instances about to be bulk-updated (non-blocking).
-
-    Only fields named in ``fields`` that are registered Char/Text columns are checked,
-    so callers updating JSON or non-text columns (e.g. Host ``ansible_facts``) are not
-    scanned for unrelated text on the in-memory instance.
-    """
-    update_fields = frozenset(fields)
-    if not update_fields or not instances:
-        return
-    caller_info = _get_caller_info()
-    for instance in instances:
-        target_fields = update_fields & _audited_text_field_names(type(instance))
-        if not target_fields:
-            continue
-        protected = _protected_models.get(type(instance))
-        if protected is None:
-            continue
-        name_fields, excluded_fields = protected
-        resource_type = f"{instance._meta.app_label}.{instance._meta.object_name}"
-        for field_name in target_fields:
-            if field_name in excluded_fields:
-                continue
-            value = getattr(instance, field_name, None)
-            if value is None or not isinstance(value, str):
-                continue
-            violation = _validate_field(field_name, value, name_fields)
-            if violation:
-                tier, reason = violation
-                _log_bulk_violation('bulk_update', field_name, resource_type, tier, caller_info, reason)
+                if caller_info is None:
+                    caller_info = _get_caller_info()
+                log_orm_bypass_violation('bulk_create', field_name, resource_type, tier, caller_info, reason)