diff --git a/cypilot/.core/skills/connector/workflows/create.md b/cypilot/.core/skills/connector/workflows/create.md
index c23f1fc2d..8f4fdf26c 100644
--- a/cypilot/.core/skills/connector/workflows/create.md
+++ b/cypilot/.core/skills/connector/workflows/create.md
@@ -503,6 +503,8 @@ If `validate-strict` fails, do NOT proceed. Fix per-path errors first — the Bu
 
 If `validate-strict` passes but `validate` fails, there is a runtime problem — usually a bad Jinja expression, a template reference to an undefined config key, or a `$ref` pointing at a path that does not exist.
 
+Both validators run inside `airbyte/source-declarative-manifest` pinned (in `source.sh` / `Dockerfile`) to the CDK version bundled with the deployed Airbyte platform — see `src/ingestion/tools/declarative-connector/README.md` §"Validator image pin". Never override the image to `:latest`: its strict schema drifts ahead of the deployed Builder UI and produces false failures.
+
 ### 5.3 Check credentials against API
 
 ```bash
diff --git a/src/ingestion/connectors/ai/claude-admin/connector.yaml b/src/ingestion/connectors/ai/claude-admin/connector.yaml
index 86fd34b92..845c049d7 100644
--- a/src/ingestion/connectors/ai/claude-admin/connector.yaml
+++ b/src/ingestion/connectors/ai/claude-admin/connector.yaml
@@ -20,13 +20,17 @@ definitions:
   tenant_id_injection:
     type: AddFields
     fields:
-      - path: [tenant_id]
+      - type: AddedFieldDefinition
+        path: [tenant_id]
         value: "{{ config['insight_tenant_id'] }}"
-      - path: [insight_source_id]
+      - type: AddedFieldDefinition
+        path: [insight_source_id]
         value: "{{ config.get('insight_source_id', '') }}"
-      - path: [collected_at]
+      - type: AddedFieldDefinition
+        path: [collected_at]
         value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%SZ') }}"
-      - path: [data_source]
+      - type: AddedFieldDefinition
+        path: [data_source]
         value: insight_claude_admin
 
   cursor_paginator:
@@ -225,17 +229,23 @@ streams:
       - $ref: "#/definitions/tenant_id_injection"
       - type: AddFields
         fields:
-          - path: [date]
+          - type: AddedFieldDefinition
+            path: [date]
             value: "{{ stream_interval['start_time'][:10] }}"
-          - path: [cache_read_tokens]
+          - type: AddedFieldDefinition
+            path: [cache_read_tokens]
             value: "{{ record.get('cache_read_input_tokens', 0) }}"
-          - path: [cache_creation_5m_tokens]
+          - type: AddedFieldDefinition
+            path: [cache_creation_5m_tokens]
             value: "{{ record.get('cache_creation', {}).get('ephemeral_5m_input_tokens', 0) }}"
-          - path: [cache_creation_1h_tokens]
+          - type: AddedFieldDefinition
+            path: [cache_creation_1h_tokens]
             value: "{{ record.get('cache_creation', {}).get('ephemeral_1h_input_tokens', 0) }}"
-          - path: [web_search_requests]
+          - type: AddedFieldDefinition
+            path: [web_search_requests]
             value: "{{ record.get('server_tool_use', {}).get('web_search_requests', 0) }}"
-          - path: [unique]
+          - type: AddedFieldDefinition
+            path: [unique]
             value: "{{ stream_interval['start_time'][:10] }}|{{ record.get('model') or '' }}|{{ record.get('api_key_id') or '' }}|{{ record.get('workspace_id') or '' }}|{{ record.get('service_tier') or '' }}|{{ record.get('context_window') or '' }}"
     incremental_sync:
       type: DatetimeBasedCursor
@@ -349,9 +359,11 @@ streams:
       - $ref: "#/definitions/tenant_id_injection"
       - type: AddFields
         fields:
-          - path: [date]
+          - type: AddedFieldDefinition
+            path: [date]
             value: "{{ stream_interval['start_time'][:10] }}"
-          - path: [unique]
+          - type: AddedFieldDefinition
+            path: [unique]
             value: "{{ stream_interval['start_time'][:10] }}|{{ record.get('workspace_id') or '' }}|{{ record.get('description') or '' }}"
     incremental_sync:
       type: DatetimeBasedCursor
@@ -469,29 +481,41 @@ streams:
       - $ref: "#/definitions/tenant_id_injection"
       - type: AddFields
         fields:
-          - path: [date]
+          - type: AddedFieldDefinition
+            path: [date]
             value: "{{ record.get('date', '')[:10] }}"
-          - path: [actor_type]
+          - type: AddedFieldDefinition
+            path: [actor_type]
             value: "{{ record.get('actor', {}).get('type') or '' }}"
-          - path: [actor_identifier]
+          - type: AddedFieldDefinition
+            path: [actor_identifier]
             value: "{{ record.get('actor', {}).get('api_key_name') or record.get('actor', {}).get('email') or '' }}"
-          - path: [session_count]
+          - type: AddedFieldDefinition
+            path: [session_count]
             value: "{{ record.get('core_metrics', {}).get('num_sessions', 0) }}"
-          - path: [lines_added]
+          - type: AddedFieldDefinition
+            path: [lines_added]
             value: "{{ record.get('core_metrics', {}).get('lines_of_code', {}).get('added', 0) }}"
-          - path: [lines_removed]
+          - type: AddedFieldDefinition
+            path: [lines_removed]
             value: "{{ record.get('core_metrics', {}).get('lines_of_code', {}).get('removed', 0) }}"
-          - path: [tool_use_accepted]
+          - type: AddedFieldDefinition
+            path: [tool_use_accepted]
             value: "{{ (record.get('tool_actions', {}).get('edit_tool', {}).get('accepted', 0) or 0) + (record.get('tool_actions', {}).get('write_tool', {}).get('accepted', 0) or 0) + (record.get('tool_actions', {}).get('multi_edit_tool', {}).get('accepted', 0) or 0) + (record.get('tool_actions', {}).get('notebook_edit_tool', {}).get('accepted', 0) or 0) }}"
-          - path: [tool_use_rejected]
+          - type: AddedFieldDefinition
+            path: [tool_use_rejected]
             value: "{{ (record.get('tool_actions', {}).get('edit_tool', {}).get('rejected', 0) or 0) + (record.get('tool_actions', {}).get('write_tool', {}).get('rejected', 0) or 0) + (record.get('tool_actions', {}).get('multi_edit_tool', {}).get('rejected', 0) or 0) + (record.get('tool_actions', {}).get('notebook_edit_tool', {}).get('rejected', 0) or 0) }}"
-          - path: [core_metrics_json]
+          - type: AddedFieldDefinition
+            path: [core_metrics_json]
             value: "{{ record.get('core_metrics', {}) | tojson }}"
-          - path: [tool_actions_json]
+          - type: AddedFieldDefinition
+            path: [tool_actions_json]
             value: "{{ record.get('tool_actions', {}) | tojson }}"
-          - path: [model_breakdown_json]
+          - type: AddedFieldDefinition
+            path: [model_breakdown_json]
             value: "{{ record.get('model_breakdown', []) | tojson }}"
-          - path: [unique]
+          - type: AddedFieldDefinition
+            path: [unique]
             value: "{{ record.get('date', '')[:10] }}|{{ record.get('actor', {}).get('type') or '' }}|{{ record.get('actor', {}).get('api_key_name') or record.get('actor', {}).get('email') or '' }}|{{ record.get('terminal_type') or '' }}"
     incremental_sync:
       type: DatetimeBasedCursor
@@ -583,11 +607,14 @@ streams:
       - $ref: "#/definitions/tenant_id_injection"
       - type: AddFields
         fields:
-          - path: [created_by_id]
+          - type: AddedFieldDefinition
+            path: [created_by_id]
             value: "{{ record.get('created_by', {}).get('id', '') }}"
-          - path: [created_by_name]
+          - type: AddedFieldDefinition
+            path: [created_by_name]
             value: "{{ record.get('created_by', {}).get('name', '') }}"
-          - path: [created_by_type]
+          - type: AddedFieldDefinition
+            path: [created_by_type]
             value: "{{ record.get('created_by', {}).get('type', '') }}"
 
   # ── Stream 6: claude_admin_workspaces (full refresh, offset pagination)
@@ -653,7 +680,8 @@ streams:
       - $ref: "#/definitions/tenant_id_injection"
       - type: AddFields
         fields:
-          - path: [data_residency]
+          - type: AddedFieldDefinition
+            path: [data_residency]
             value: "{{ record.get('data_residency', {}) | tojson if record.get('data_residency') else '' }}"
 
   # ── Stream 7: claude_admin_workspace_members (full refresh, substream of workspaces) ──
@@ -750,9 +778,11 @@ streams:
       - $ref: "#/definitions/tenant_id_injection"
       - type: AddFields
         fields:
-          - path: [workspace_id]
+          - type: AddedFieldDefinition
+            path: [workspace_id]
             value: "{{ stream_partition['workspace_id'] }}"
-          - path: [unique]
+          - type: AddedFieldDefinition
+            path: [unique]
             value: "{{ record.get('user_id', '') }}:{{ stream_partition['workspace_id'] }}"
 
   # ── Stream 8: claude_admin_invites (full refresh, offset pagination)
diff --git a/src/ingestion/connectors/ai/cursor/connector.yaml b/src/ingestion/connectors/ai/cursor/connector.yaml
index e8d23f9b9..48c24fe8a 100644
--- a/src/ingestion/connectors/ai/cursor/connector.yaml
+++ b/src/ingestion/connectors/ai/cursor/connector.yaml
@@ -22,9 +22,11 @@ definitions:
   mandatory_fields:
     type: AddFields
     fields:
-      - path: [tenant_id]
+      - type: AddedFieldDefinition
+        path: [tenant_id]
         value: "{{ config['insight_tenant_id'] }}"
-      - path: [source_id]
+      - type: AddedFieldDefinition
+        path: [source_id]
         value: "{{ config['insight_source_id'] }}"
 
   post_paginator:
@@ -111,7 +113,8 @@ streams:
       - $ref: "#/definitions/mandatory_fields"
       - type: AddFields
         fields:
-          - path: [unique_key]
+          - type: AddedFieldDefinition
+            path: [unique_key]
             value: "{{ config['insight_tenant_id'] }}-{{ config['insight_source_id'] }}-{{ record['id'] }}"
 
   # ── Stream 2: cursor_audit_logs (incremental, GET) ──
@@ -171,7 +174,8 @@ streams:
       - $ref: "#/definitions/mandatory_fields"
       - type: AddFields
         fields:
-          - path: [unique_key]
+          - type: AddedFieldDefinition
+            path: [unique_key]
             value: "{{ config['insight_tenant_id'] }}-{{ config['insight_source_id'] }}-{{ record['event_id'] }}"
     incremental_sync:
       type: DatetimeBasedCursor
@@ -276,7 +280,8 @@ streams:
       - $ref: "#/definitions/mandatory_fields"
       - type: AddFields
         fields:
-          - path: [unique_key]
+          - type: AddedFieldDefinition
+            path: [unique_key]
             value: "{{ config['insight_tenant_id'] }}-{{ config['insight_source_id'] }}-{{ record['userEmail'] }}-{{ record['timestamp'] }}"
     incremental_sync:
       type: DatetimeBasedCursor
@@ -372,7 +377,8 @@ streams:
       - $ref: "#/definitions/mandatory_fields"
       - type: AddFields
         fields:
-          - path: [unique_key]
+          - type: AddedFieldDefinition
+            path: [unique_key]
             value: "{{ config['insight_tenant_id'] }}-{{ config['insight_source_id'] }}-{{ record['userEmail'] }}-{{ record['timestamp'] }}"
     incremental_sync:
       type: DatetimeBasedCursor
@@ -490,7 +496,8 @@ streams:
       - $ref: "#/definitions/mandatory_fields"
       - type: AddFields
         fields:
-          - path: [unique_key]
+          - type: AddedFieldDefinition
+            path: [unique_key]
             value: "{{ config['insight_tenant_id'] }}-{{ config['insight_source_id'] }}-{{ record['userId'] }}-{{ record['date'] }}"
     incremental_sync:
       type: DatetimeBasedCursor
diff --git a/src/ingestion/connectors/git/bitbucket-server/connector.yaml b/src/ingestion/connectors/git/bitbucket-server/connector.yaml
index b229af837..30f96cbdd 100644
--- a/src/ingestion/connectors/git/bitbucket-server/connector.yaml
+++ b/src/ingestion/connectors/git/bitbucket-server/connector.yaml
@@ -4,6 +4,7 @@
 # Sync: full refresh (most streams), incremental (pull_requests on updatedDate)
 
 version: "6.44.0"
+type: DeclarativeSource
 
 definitions:
   # --- Shared components ---
@@ -83,13 +84,17 @@ definitions:
   standard_transformations:
     - type: AddFields
       fields:
-        - path: [tenant_id]
+        - type: AddedFieldDefinition
+          path: [tenant_id]
           value: "{{ config['tenant_id'] }}"
-        - path: [source_instance_id]
+        - type: AddedFieldDefinition
+          path: [source_instance_id]
           value: "bitbucket-{{ config['instance_name'] }}"
-        - path: [_source]
+        - type: AddedFieldDefinition
+          path: [_source]
           value: "bitbucket_server"
-        - path: [_extracted_at]
+        - type: AddedFieldDefinition
+          path: [_extracted_at]
           value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%SZ') }}"
 
   # ===================================================================
@@ -113,15 +118,20 @@ definitions:
     transformations:
       - type: AddFields
         fields:
-          - path: [insight_tenant_id]
+          - type: AddedFieldDefinition
+            path: [insight_tenant_id]
             value: "{{ config['tenant_id'] }}"
-          - path: [insight_source_id]
+          - type: AddedFieldDefinition
+            path: [insight_source_id]
             value: "{{ config['insight_source_id'] }}"
-          - path: [source_instance_id]
+          - type: AddedFieldDefinition
+            path: [source_instance_id]
             value: "bitbucket-{{ config['instance_name'] }}"
-          - path: [_source]
+          - type: AddedFieldDefinition
+            path: [_source]
             value: "bitbucket_server"
-          - path: [_extracted_at]
+          - type: AddedFieldDefinition
+            path: [_extracted_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%SZ') }}"
     schema_loader:
       type: InlineSchemaLoader
@@ -191,17 +201,23 @@ definitions:
     transformations:
       - type: AddFields
         fields:
-          - path: [project_key]
+          - type: AddedFieldDefinition
+            path: [project_key]
             value: "{{ stream_partition.project_key }}"
-          - path: [insight_tenant_id]
+          - type: AddedFieldDefinition
+            path: [insight_tenant_id]
             value: "{{ config['tenant_id'] }}"
-          - path: [insight_source_id]
+          - type: AddedFieldDefinition
+            path: [insight_source_id]
             value: "{{ config['insight_source_id'] }}"
-          - path: [source_instance_id]
+          - type: AddedFieldDefinition
+            path: [source_instance_id]
             value: "bitbucket-{{ config['instance_name'] }}"
-          - path: [_source]
+          - type: AddedFieldDefinition
+            path: [_source]
             value: "bitbucket_server"
-          - path: [_extracted_at]
+          - type: AddedFieldDefinition
+            path: [_extracted_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%SZ') }}"
     schema_loader:
       type: InlineSchemaLoader
@@ -290,19 +306,26 @@ definitions:
     transformations:
       - type: AddFields
         fields:
-          - path: [project_key]
+          - type: AddedFieldDefinition
+            path: [project_key]
             value: "{{ stream_partition.project_key }}"
-          - path: [repo_slug]
+          - type: AddedFieldDefinition
+            path: [repo_slug]
             value: "{{ stream_partition.repo_slug }}"
-          - path: [insight_tenant_id]
+          - type: AddedFieldDefinition
+            path: [insight_tenant_id]
             value: "{{ config['tenant_id'] }}"
-          - path: [insight_source_id]
+          - type: AddedFieldDefinition
+            path: [insight_source_id]
             value: "{{ config['insight_source_id'] }}"
-          - path: [source_instance_id]
+          - type: AddedFieldDefinition
+            path: [source_instance_id]
             value: "bitbucket-{{ config['instance_name'] }}"
-          - path: [_source]
+          - type: AddedFieldDefinition
+            path: [_source]
             value: "bitbucket_server"
-          - path: [_extracted_at]
+          - type: AddedFieldDefinition
+            path: [_extracted_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%SZ') }}"
     schema_loader:
       type: InlineSchemaLoader
@@ -384,19 +407,26 @@ definitions:
     transformations:
       - type: AddFields
         fields:
-          - path: [project_key]
+          - type: AddedFieldDefinition
+            path: [project_key]
             value: "{{ stream_partition.project_key }}"
-          - path: [repo_slug]
+          - type: AddedFieldDefinition
+            path: [repo_slug]
             value: "{{ stream_partition.repo_slug }}"
-          - path: [insight_tenant_id]
+          - type: AddedFieldDefinition
+            path: [insight_tenant_id]
             value: "{{ config['tenant_id'] }}"
-          - path: [insight_source_id]
+          - type: AddedFieldDefinition
+            path: [insight_source_id]
             value: "{{ config['insight_source_id'] }}"
-          - path: [source_instance_id]
+          - type: AddedFieldDefinition
+            path: [source_instance_id]
             value: "bitbucket-{{ config['instance_name'] }}"
-          - path: [_source]
+          - type: AddedFieldDefinition
+            path: [_source]
             value: "bitbucket_server"
-          - path: [_extracted_at]
+          - type: AddedFieldDefinition
+            path: [_extracted_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%SZ') }}"
     schema_loader:
       type: InlineSchemaLoader
@@ -503,19 +533,26 @@ definitions:
     transformations:
       - type: AddFields
         fields:
-          - path: [project_key]
+          - type: AddedFieldDefinition
+            path: [project_key]
             value: "{{ stream_partition.project_key }}"
-          - path: [repo_slug]
+          - type: AddedFieldDefinition
+            path: [repo_slug]
             value: "{{ stream_partition.repo_slug }}"
-          - path: [insight_tenant_id]
+          - type: AddedFieldDefinition
+            path: [insight_tenant_id]
             value: "{{ config['tenant_id'] }}"
-          - path: [insight_source_id]
+          - type: AddedFieldDefinition
+            path: [insight_source_id]
             value: "{{ config['insight_source_id'] }}"
-          - path: [source_instance_id]
+          - type: AddedFieldDefinition
+            path: [source_instance_id]
             value: "bitbucket-{{ config['instance_name'] }}"
-          - path: [_source]
+          - type: AddedFieldDefinition
+            path: [_source]
             value: "bitbucket_server"
-          - path: [_extracted_at]
+          - type: AddedFieldDefinition
+            path: [_extracted_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%SZ') }}"
     schema_loader:
       type: InlineSchemaLoader
@@ -638,21 +675,29 @@ definitions:
     transformations:
       - type: AddFields
         fields:
-          - path: [project_key]
+          - type: AddedFieldDefinition
+            path: [project_key]
             value: "{{ stream_partition.project_key }}"
-          - path: [repo_slug]
+          - type: AddedFieldDefinition
+            path: [repo_slug]
             value: "{{ stream_partition.repo_slug }}"
-          - path: [pr_id]
+          - type: AddedFieldDefinition
+            path: [pr_id]
             value: "{{ stream_partition.pr_id }}"
-          - path: [insight_tenant_id]
+          - type: AddedFieldDefinition
+            path: [insight_tenant_id]
             value: "{{ config['tenant_id'] }}"
-          - path: [insight_source_id]
+          - type: AddedFieldDefinition
+            path: [insight_source_id]
             value: "{{ config['insight_source_id'] }}"
-          - path: [source_instance_id]
+          - type: AddedFieldDefinition
+            path: [source_instance_id]
             value: "bitbucket-{{ config['instance_name'] }}"
-          - path: [_source]
+          - type: AddedFieldDefinition
+            path: [_source]
             value: "bitbucket_server"
-          - path: [_extracted_at]
+          - type: AddedFieldDefinition
+            path: [_extracted_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%SZ') }}"
     schema_loader:
       type: InlineSchemaLoader
@@ -718,15 +763,20 @@ definitions:
     transformations:
       - type: AddFields
         fields:
-          - path: [insight_tenant_id]
+          - type: AddedFieldDefinition
+            path: [insight_tenant_id]
             value: "{{ config['tenant_id'] }}"
-          - path: [insight_source_id]
+          - type: AddedFieldDefinition
+            path: [insight_source_id]
             value: "{{ config['insight_source_id'] }}"
-          - path: [source_instance_id]
+          - type: AddedFieldDefinition
+            path: [source_instance_id]
             value: "bitbucket-{{ config['instance_name'] }}"
-          - path: [_source]
+          - type: AddedFieldDefinition
+            path: [_source]
             value: "bitbucket_server"
-          - path: [_extracted_at]
+          - type: AddedFieldDefinition
+            path: [_extracted_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%SZ') }}"
     schema_loader:
       type: InlineSchemaLoader
@@ -812,19 +862,26 @@ definitions:
     transformations:
       - type: AddFields
         fields:
-          - path: [project_key]
+          - type: AddedFieldDefinition
+            path: [project_key]
             value: "{{ stream_partition.project_key }}"
-          - path: [repo_slug]
+          - type: AddedFieldDefinition
+            path: [repo_slug]
             value: "{{ stream_partition.repo_slug }}"
-          - path: [insight_tenant_id]
+          - type: AddedFieldDefinition
+            path: [insight_tenant_id]
             value: "{{ config['tenant_id'] }}"
-          - path: [insight_source_id]
+          - type: AddedFieldDefinition
+            path: [insight_source_id]
             value: "{{ config['insight_source_id'] }}"
-          - path: [source_instance_id]
+          - type: AddedFieldDefinition
+            path: [source_instance_id]
             value: "bitbucket-{{ config['instance_name'] }}"
-          - path: [_source]
+          - type: AddedFieldDefinition
+            path: [_source]
             value: "bitbucket_server"
-          - path: [_extracted_at]
+          - type: AddedFieldDefinition
+            path: [_extracted_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%SZ') }}"
     schema_loader:
       type: InlineSchemaLoader
@@ -914,21 +971,29 @@ definitions:
     transformations:
       - type: AddFields
         fields:
-          - path: [project_key]
+          - type: AddedFieldDefinition
+            path: [project_key]
             value: "{{ stream_partition.project_key }}"
-          - path: [repo_slug]
+          - type: AddedFieldDefinition
+            path: [repo_slug]
             value: "{{ stream_partition.repo_slug }}"
-          - path: [pr_id]
+          - type: AddedFieldDefinition
+            path: [pr_id]
             value: "{{ stream_partition.pr_id }}"
-          - path: [insight_tenant_id]
+          - type: AddedFieldDefinition
+            path: [insight_tenant_id]
             value: "{{ config['tenant_id'] }}"
-          - path: [insight_source_id]
+          - type: AddedFieldDefinition
+            path: [insight_source_id]
             value: "{{ config['insight_source_id'] }}"
-          - path: [source_instance_id]
+          - type: AddedFieldDefinition
+            path: [source_instance_id]
             value: "bitbucket-{{ config['instance_name'] }}"
-          - path: [_source]
+          - type: AddedFieldDefinition
+            path: [_source]
             value: "bitbucket_server"
-          - path: [_extracted_at]
+          - type: AddedFieldDefinition
+            path: [_extracted_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%SZ') }}"
     schema_loader:
       type: InlineSchemaLoader
@@ -1018,21 +1083,29 @@ definitions:
     transformations:
       - type: AddFields
         fields:
-          - path: [project_key]
+          - type: AddedFieldDefinition
+            path: [project_key]
             value: "{{ stream_partition.project_key }}"
-          - path: [repo_slug]
+          - type: AddedFieldDefinition
+            path: [repo_slug]
             value: "{{ stream_partition.repo_slug }}"
-          - path: [pr_id]
+          - type: AddedFieldDefinition
+            path: [pr_id]
             value: "{{ stream_partition.pr_id }}"
-          - path: [insight_tenant_id]
+          - type: AddedFieldDefinition
+            path: [insight_tenant_id]
             value: "{{ config['tenant_id'] }}"
-          - path: [insight_source_id]
+          - type: AddedFieldDefinition
+            path: [insight_source_id]
             value: "{{ config['insight_source_id'] }}"
-          - path: [source_instance_id]
+          - type: AddedFieldDefinition
+            path: [source_instance_id]
             value: "bitbucket-{{ config['instance_name'] }}"
-          - path: [_source]
+          - type: AddedFieldDefinition
+            path: [_source]
             value: "bitbucket_server"
-          - path: [_extracted_at]
+          - type: AddedFieldDefinition
+            path: [_extracted_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%SZ') }}"
     schema_loader:
       type: InlineSchemaLoader
@@ -1124,21 +1197,29 @@ definitions:
     transformations:
       - type: AddFields
         fields:
-          - path: [project_key]
+          - type: AddedFieldDefinition
+            path: [project_key]
             value: "{{ stream_partition.project_key }}"
-          - path: [repo_slug]
+          - type: AddedFieldDefinition
+            path: [repo_slug]
             value: "{{ stream_partition.repo_slug }}"
-          - path: [commit_id]
+          - type: AddedFieldDefinition
+            path: [commit_id]
             value: "{{ stream_partition.commit_id }}"
-          - path: [insight_tenant_id]
+          - type: AddedFieldDefinition
+            path: [insight_tenant_id]
             value: "{{ config['tenant_id'] }}"
-          - path: [insight_source_id]
+          - type: AddedFieldDefinition
+            path: [insight_source_id]
             value: "{{ config['insight_source_id'] }}"
-          - path: [source_instance_id]
+          - type: AddedFieldDefinition
+            path: [source_instance_id]
             value: "bitbucket-{{ config['instance_name'] }}"
-          - path: [_source]
+          - type: AddedFieldDefinition
+            path: [_source]
             value: "bitbucket_server"
-          - path: [_extracted_at]
+          - type: AddedFieldDefinition
+            path: [_extracted_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%SZ') }}"
     schema_loader:
       type: InlineSchemaLoader
@@ -1227,17 +1308,23 @@ definitions:
     transformations:
       - type: AddFields
         fields:
-          - path: [commit_id]
+          - type: AddedFieldDefinition
+            path: [commit_id]
             value: "{{ stream_partition.commit_id }}"
-          - path: [insight_tenant_id]
+          - type: AddedFieldDefinition
+            path: [insight_tenant_id]
             value: "{{ config['tenant_id'] }}"
-          - path: [insight_source_id]
+          - type: AddedFieldDefinition
+            path: [insight_source_id]
             value: "{{ config['insight_source_id'] }}"
-          - path: [source_instance_id]
+          - type: AddedFieldDefinition
+            path: [source_instance_id]
             value: "bitbucket-{{ config['instance_name'] }}"
-          - path: [_source]
+          - type: AddedFieldDefinition
+            path: [_source]
             value: "bitbucket_server"
-          - path: [_extracted_at]
+          - type: AddedFieldDefinition
+            path: [_extracted_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%SZ') }}"
     schema_loader:
       type: InlineSchemaLoader
@@ -1324,21 +1411,29 @@ definitions:
     transformations:
       - type: AddFields
         fields:
-          - path: [project_key]
+          - type: AddedFieldDefinition
+            path: [project_key]
             value: "{{ stream_partition.project_key }}"
-          - path: [repo_slug]
+          - type: AddedFieldDefinition
+            path: [repo_slug]
             value: "{{ stream_partition.repo_slug }}"
-          - path: [pr_id]
+          - type: AddedFieldDefinition
+            path: [pr_id]
             value: "{{ stream_partition.pr_id }}"
-          - path: [insight_tenant_id]
+          - type: AddedFieldDefinition
+            path: [insight_tenant_id]
             value: "{{ config['tenant_id'] }}"
-          - path: [insight_source_id]
+          - type: AddedFieldDefinition
+            path: [insight_source_id]
             value: "{{ config['insight_source_id'] }}"
-          - path: [source_instance_id]
+          - type: AddedFieldDefinition
+            path: [source_instance_id]
             value: "bitbucket-{{ config['instance_name'] }}"
-          - path: [_source]
+          - type: AddedFieldDefinition
+            path: [_source]
             value: "bitbucket_server"
-          - path: [_extracted_at]
+          - type: AddedFieldDefinition
+            path: [_extracted_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%SZ') }}"
     schema_loader:
       type: InlineSchemaLoader
@@ -1399,15 +1494,20 @@ definitions:
     transformations:
       - type: AddFields
         fields:
-          - path: [insight_tenant_id]
+          - type: AddedFieldDefinition
+            path: [insight_tenant_id]
             value: "{{ config['tenant_id'] }}"
-          - path: [insight_source_id]
+          - type: AddedFieldDefinition
+            path: [insight_source_id]
             value: "{{ config['insight_source_id'] }}"
-          - path: [source_instance_id]
+          - type: AddedFieldDefinition
+            path: [source_instance_id]
             value: "bitbucket-{{ config['instance_name'] }}"
-          - path: [_source]
+          - type: AddedFieldDefinition
+            path: [_source]
             value: "bitbucket_server"
-          - path: [_extracted_at]
+          - type: AddedFieldDefinition
+            path: [_extracted_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%SZ') }}"
     schema_loader:
       type: InlineSchemaLoader
@@ -1463,17 +1563,23 @@ definitions:
     transformations:
       - type: AddFields
         fields:
-          - path: [group_name]
+          - type: AddedFieldDefinition
+            path: [group_name]
             value: "{{ stream_partition.group_name }}"
-          - path: [insight_tenant_id]
+          - type: AddedFieldDefinition
+            path: [insight_tenant_id]
             value: "{{ config['tenant_id'] }}"
-          - path: [insight_source_id]
+          - type: AddedFieldDefinition
+            path: [insight_source_id]
             value: "{{ config['insight_source_id'] }}"
-          - path: [source_instance_id]
+          - type: AddedFieldDefinition
+            path: [source_instance_id]
             value: "bitbucket-{{ config['instance_name'] }}"
-          - path: [_source]
+          - type: AddedFieldDefinition
+            path: [_source]
             value: "bitbucket_server"
-          - path: [_extracted_at]
+          - type: AddedFieldDefinition
+            path: [_extracted_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%SZ') }}"
     schema_loader:
       type: InlineSchemaLoader
@@ -1541,19 +1647,26 @@ definitions:
     transformations:
       - type: AddFields
         fields:
-          - path: [project_key]
+          - type: AddedFieldDefinition
+            path: [project_key]
             value: "{{ stream_partition.project_key }}"
-          - path: [user_id]
+          - type: AddedFieldDefinition
+            path: [user_id]
             value: "{{ record.get('user', {}).get('id', '') }}"
-          - path: [insight_tenant_id]
+          - type: AddedFieldDefinition
+            path: [insight_tenant_id]
             value: "{{ config['tenant_id'] }}"
-          - path: [insight_source_id]
+          - type: AddedFieldDefinition
+            path: [insight_source_id]
             value: "{{ config['insight_source_id'] }}"
-          - path: [source_instance_id]
+          - type: AddedFieldDefinition
+            path: [source_instance_id]
             value: "bitbucket-{{ config['instance_name'] }}"
-          - path: [_source]
+          - type: AddedFieldDefinition
+            path: [_source]
             value: "bitbucket_server"
-          - path: [_extracted_at]
+          - type: AddedFieldDefinition
+            path: [_extracted_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%SZ') }}"
     schema_loader:
       type: InlineSchemaLoader
@@ -1616,19 +1729,26 @@ definitions:
     transformations:
       - type: AddFields
         fields:
-          - path: [project_key]
+          - type: AddedFieldDefinition
+            path: [project_key]
             value: "{{ stream_partition.project_key }}"
-          - path: [group_name]
+          - type: AddedFieldDefinition
+            path: [group_name]
             value: "{{ record.get('group', {}).get('name', '') }}"
-          - path: [insight_tenant_id]
+          - type: AddedFieldDefinition
+            path: [insight_tenant_id]
             value: "{{ config['tenant_id'] }}"
-          - path: [insight_source_id]
+          - type: AddedFieldDefinition
+            path: [insight_source_id]
             value: "{{ config['insight_source_id'] }}"
-          - path: [source_instance_id]
+          - type: AddedFieldDefinition
+            path: [source_instance_id]
             value: "bitbucket-{{ config['instance_name'] }}"
-          - path: [_source]
+          - type: AddedFieldDefinition
+            path: [_source]
             value: "bitbucket_server"
-          - path: [_extracted_at]
+          - type: AddedFieldDefinition
+            path: [_extracted_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%SZ') }}"
     schema_loader:
       type: InlineSchemaLoader
@@ -1699,21 +1819,29 @@ definitions:
     transformations:
       - type: AddFields
         fields:
-          - path: [project_key]
+          - type: AddedFieldDefinition
+            path: [project_key]
             value: "{{ stream_partition.project_key }}"
-          - path: [repo_slug]
+          - type: AddedFieldDefinition
+            path: [repo_slug]
             value: "{{ stream_partition.repo_slug }}"
-          - path: [user_id]
+          - type: AddedFieldDefinition
+            path: [user_id]
             value: "{{ record.get('user', {}).get('id', '') }}"
-          - path: [insight_tenant_id]
+          - type: AddedFieldDefinition
+            path: [insight_tenant_id]
             value: "{{ config['tenant_id'] }}"
-          - path: [insight_source_id]
+          - type: AddedFieldDefinition
+            path: [insight_source_id]
             value: "{{ config['insight_source_id'] }}"
-          - path: [source_instance_id]
+          - type: AddedFieldDefinition
+            path: [source_instance_id]
             value: "bitbucket-{{ config['instance_name'] }}"
-          - path: [_source]
+          - type: AddedFieldDefinition
+            path: [_source]
             value: "bitbucket_server"
-          - path: [_extracted_at]
+          - type: AddedFieldDefinition
+            path: [_extracted_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%SZ') }}"
     schema_loader:
       type: InlineSchemaLoader
@@ -1786,21 +1914,29 @@ definitions:
     transformations:
       - type: AddFields
         fields:
-          - path: [project_key]
+          - type: AddedFieldDefinition
+            path: [project_key]
             value: "{{ stream_partition.project_key }}"
-          - path: [repo_slug]
+          - type: AddedFieldDefinition
+            path: [repo_slug]
             value: "{{ stream_partition.repo_slug }}"
-          - path: [group_name]
+          - type: AddedFieldDefinition
+            path: [group_name]
             value: "{{ record.get('group', {}).get('name', '') }}"
-          - path: [insight_tenant_id]
+          - type: AddedFieldDefinition
+            path: [insight_tenant_id]
             value: "{{ config['tenant_id'] }}"
-          - path: [insight_source_id]
+          - type: AddedFieldDefinition
+            path: [insight_source_id]
             value: "{{ config['insight_source_id'] }}"
-          - path: [source_instance_id]
+          - type: AddedFieldDefinition
+            path: [source_instance_id]
             value: "bitbucket-{{ config['instance_name'] }}"
-          - path: [_source]
+          - type: AddedFieldDefinition
+            path: [_source]
             value: "bitbucket_server"
-          - path: [_extracted_at]
+          - type: AddedFieldDefinition
+            path: [_extracted_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%SZ') }}"
     schema_loader:
       type: InlineSchemaLoader
@@ -1856,17 +1992,23 @@ definitions:
     transformations:
       - type: AddFields
         fields:
-          - path: [user_id]
+          - type: AddedFieldDefinition
+            path: [user_id]
             value: "{{ record.get('user', {}).get('id', '') }}"
-          - path: [insight_tenant_id]
+          - type: AddedFieldDefinition
+            path: [insight_tenant_id]
             value: "{{ config['tenant_id'] }}"
-          - path: [insight_source_id]
+          - type: AddedFieldDefinition
+            path: [insight_source_id]
             value: "{{ config['insight_source_id'] }}"
-          - path: [source_instance_id]
+          - type: AddedFieldDefinition
+            path: [source_instance_id]
             value: "bitbucket-{{ config['instance_name'] }}"
-          - path: [_source]
+          - type: AddedFieldDefinition
+            path: [_source]
             value: "bitbucket_server"
-          - path: [_extracted_at]
+          - type: AddedFieldDefinition
+            path: [_extracted_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%SZ') }}"
     schema_loader:
       type: InlineSchemaLoader
@@ -1918,17 +2060,23 @@ definitions:
     transformations:
       - type: AddFields
         fields:
-          - path: [group_name]
+          - type: AddedFieldDefinition
+            path: [group_name]
             value: "{{ record.get('group', {}).get('name', '') }}"
-          - path: [insight_tenant_id]
+          - type: AddedFieldDefinition
+            path: [insight_tenant_id]
             value: "{{ config['tenant_id'] }}"
-          - path: [insight_source_id]
+          - type: AddedFieldDefinition
+            path: [insight_source_id]
             value: "{{ config['insight_source_id'] }}"
-          - path: [source_instance_id]
+          - type: AddedFieldDefinition
+            path: [source_instance_id]
             value: "bitbucket-{{ config['instance_name'] }}"
-          - path: [_source]
+          - type: AddedFieldDefinition
+            path: [_source]
             value: "bitbucket_server"
-          - path: [_extracted_at]
+          - type: AddedFieldDefinition
+            path: [_extracted_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%SZ') }}"
     schema_loader:
       type: InlineSchemaLoader
@@ -1997,19 +2145,26 @@ definitions:
     transformations:
       - type: AddFields
         fields:
-          - path: [project_key]
+          - type: AddedFieldDefinition
+            path: [project_key]
             value: "{{ stream_partition.project_key }}"
-          - path: [repo_slug]
+          - type: AddedFieldDefinition
+            path: [repo_slug]
             value: "{{ stream_partition.repo_slug }}"
-          - path: [insight_tenant_id]
+          - type: AddedFieldDefinition
+            path: [insight_tenant_id]
             value: "{{ config['tenant_id'] }}"
-          - path: [insight_source_id]
+          - type: AddedFieldDefinition
+            path: [insight_source_id]
             value: "{{ config['insight_source_id'] }}"
-          - path: [source_instance_id]
+          - type: AddedFieldDefinition
+            path: [source_instance_id]
             value: "bitbucket-{{ config['instance_name'] }}"
-          - path: [_source]
+          - type: AddedFieldDefinition
+            path: [_source]
             value: "bitbucket_server"
-          - path: [_extracted_at]
+          - type: AddedFieldDefinition
+            path: [_extracted_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%SZ') }}"
     schema_loader:
       type: InlineSchemaLoader
@@ -2094,19 +2249,26 @@ definitions:
     transformations:
       - type: AddFields
         fields:
-          - path: [project_key]
+          - type: AddedFieldDefinition
+            path: [project_key]
             value: "{{ stream_partition.project_key }}"
-          - path: [repo_slug]
+          - type: AddedFieldDefinition
+            path: [repo_slug]
             value: "{{ stream_partition.repo_slug }}"
-          - path: [insight_tenant_id]
+          - type: AddedFieldDefinition
+            path: [insight_tenant_id]
             value: "{{ config['tenant_id'] }}"
-          - path: [insight_source_id]
+          - type: AddedFieldDefinition
+            path: [insight_source_id]
             value: "{{ config['insight_source_id'] }}"
-          - path: [source_instance_id]
+          - type: AddedFieldDefinition
+            path: [source_instance_id]
             value: "bitbucket-{{ config['instance_name'] }}"
-          - path: [_source]
+          - type: AddedFieldDefinition
+            path: [_source]
             value: "bitbucket_server"
-          - path: [_extracted_at]
+          - type: AddedFieldDefinition
+            path: [_extracted_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%SZ') }}"
     schema_loader:
       type: InlineSchemaLoader
@@ -2190,19 +2352,26 @@ definitions:
     transformations:
       - type: AddFields
         fields:
-          - path: [project_key]
+          - type: AddedFieldDefinition
+            path: [project_key]
             value: "{{ stream_partition.project_key }}"
-          - path: [repo_slug]
+          - type: AddedFieldDefinition
+            path: [repo_slug]
             value: "{{ stream_partition.repo_slug }}"
-          - path: [insight_tenant_id]
+          - type: AddedFieldDefinition
+            path: [insight_tenant_id]
             value: "{{ config['tenant_id'] }}"
-          - path: [insight_source_id]
+          - type: AddedFieldDefinition
+            path: [insight_source_id]
             value: "{{ config['insight_source_id'] }}"
-          - path: [source_instance_id]
+          - type: AddedFieldDefinition
+            path: [source_instance_id]
             value: "bitbucket-{{ config['instance_name'] }}"
-          - path: [_source]
+          - type: AddedFieldDefinition
+            path: [_source]
             value: "bitbucket_server"
-          - path: [_extracted_at]
+          - type: AddedFieldDefinition
+            path: [_extracted_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%SZ') }}"
     schema_loader:
       type: InlineSchemaLoader
@@ -2287,19 +2456,26 @@ definitions:
     transformations:
       - type: AddFields
         fields:
-          - path: [project_key]
+          - type: AddedFieldDefinition
+            path: [project_key]
             value: "{{ stream_partition.project_key }}"
-          - path: [repo_slug]
+          - type: AddedFieldDefinition
+            path: [repo_slug]
             value: "{{ stream_partition.repo_slug }}"
-          - path: [insight_tenant_id]
+          - type: AddedFieldDefinition
+            path: [insight_tenant_id]
             value: "{{ config['tenant_id'] }}"
-          - path: [insight_source_id]
+          - type: AddedFieldDefinition
+            path: [insight_source_id]
             value: "{{ config['insight_source_id'] }}"
-          - path: [source_instance_id]
+          - type: AddedFieldDefinition
+            path: [source_instance_id]
             value: "bitbucket-{{ config['instance_name'] }}"
-          - path: [_source]
+          - type: AddedFieldDefinition
+            path: [_source]
             value: "bitbucket_server"
-          - path: [_extracted_at]
+          - type: AddedFieldDefinition
+            path: [_extracted_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%SZ') }}"
     schema_loader:
       type: InlineSchemaLoader
@@ -2379,19 +2555,26 @@ definitions:
     transformations:
       - type: AddFields
         fields:
-          - path: [project_key]
+          - type: AddedFieldDefinition
+            path: [project_key]
             value: "{{ stream_partition.project_key }}"
-          - path: [repo_slug]
+          - type: AddedFieldDefinition
+            path: [repo_slug]
             value: "{{ stream_partition.repo_slug }}"
-          - path: [insight_tenant_id]
+          - type: AddedFieldDefinition
+            path: [insight_tenant_id]
             value: "{{ config['tenant_id'] }}"
-          - path: [insight_source_id]
+          - type: AddedFieldDefinition
+            path: [insight_source_id]
             value: "{{ config['insight_source_id'] }}"
-          - path: [source_instance_id]
+          - type: AddedFieldDefinition
+            path: [source_instance_id]
             value: "bitbucket-{{ config['instance_name'] }}"
-          - path: [_source]
+          - type: AddedFieldDefinition
+            path: [_source]
             value: "bitbucket_server"
-          - path: [_extracted_at]
+          - type: AddedFieldDefinition
+            path: [_extracted_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%SZ') }}"
     schema_loader:
       type: InlineSchemaLoader
@@ -2450,15 +2633,20 @@ definitions:
     transformations:
       - type: AddFields
         fields:
-          - path: [insight_tenant_id]
+          - type: AddedFieldDefinition
+            path: [insight_tenant_id]
             value: "{{ config['tenant_id'] }}"
-          - path: [insight_source_id]
+          - type: AddedFieldDefinition
+            path: [insight_source_id]
             value: "{{ config['insight_source_id'] }}"
-          - path: [source_instance_id]
+          - type: AddedFieldDefinition
+            path: [source_instance_id]
             value: "bitbucket-{{ config['instance_name'] }}"
-          - path: [_source]
+          - type: AddedFieldDefinition
+            path: [_source]
             value: "bitbucket_server"
-          - path: [_extracted_at]
+          - type: AddedFieldDefinition
+            path: [_extracted_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%SZ') }}"
     schema_loader:
       type: InlineSchemaLoader
diff --git a/src/ingestion/connectors/wiki/confluence/connector.yaml b/src/ingestion/connectors/wiki/confluence/connector.yaml
index 67067465f..8761bada0 100644
--- a/src/ingestion/connectors/wiki/confluence/connector.yaml
+++ b/src/ingestion/connectors/wiki/confluence/connector.yaml
@@ -52,11 +52,14 @@ definitions:
   add_fields:
     type: AddFields
     fields:
-      - path: [tenant_id]
+      - type: AddedFieldDefinition
+        path: [tenant_id]
         value: "{{ config['insight_tenant_id'] }}"
-      - path: [source_id]
+      - type: AddedFieldDefinition
+        path: [source_id]
         value: "{{ config['insight_source_id'] }}"
-      - path: [data_source]
+      - type: AddedFieldDefinition
+        path: [data_source]
         value: "insight_confluence"
 
   confluence_paginator:
@@ -260,26 +263,35 @@ streams:
       - $ref: "#/definitions/add_fields"
       - type: AddFields
         fields:
-          - path: [space_id]
+          - type: AddedFieldDefinition
+            path: [space_id]
             value: "{{ record['id'] }}"
-          - path: [name]
+          - type: AddedFieldDefinition
+            path: [name]
             value: "{{ record.get('name', '') }}"
-          - path: [description]
+          - type: AddedFieldDefinition
+            path: [description]
             value: >-
               {{ (record.get('description', {}) or {}).get('plain', {}).get('value', '') }}
-          - path: [space_type]
+          - type: AddedFieldDefinition
+            path: [space_type]
             value: "{{ record['type'] }}"
-          - path: [status]
+          - type: AddedFieldDefinition
+            path: [status]
             value: >-
               {{ 'active' if record.get('status') == 'current' else record.get('status', '') }}
-          - path: [url]
+          - type: AddedFieldDefinition
+            path: [url]
             value: >-
               {{ record.get('_links', {}).get('webui', '') if record.get('_links', {}).get('webui', '').startswith('http') else config['confluence_instance_url'] + record.get('_links', {}).get('webui', '') }}
-          - path: [created_at]
+          - type: AddedFieldDefinition
+            path: [created_at]
             value: "{{ record.get('createdAt', '') }}"
-          - path: [collected_at]
+          - type: AddedFieldDefinition
+            path: [collected_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%S.000Z') }}"
-          - path: [unique_key]
+          - type: AddedFieldDefinition
+            path: [unique_key]
             value: >-
               {{ config['insight_tenant_id'] }}-{{ config['insight_source_id']
               }}-{{ record['id'] }}
@@ -333,25 +345,35 @@ streams:
       - $ref: "#/definitions/add_fields"
       - type: AddFields
         fields:
-          - path: [page_id]
+          - type: AddedFieldDefinition
+            path: [page_id]
             value: "{{ record['id'] }}"
-          - path: [space_id]
+          - type: AddedFieldDefinition
+            path: [space_id]
             value: "{{ record['spaceId'] }}"
-          - path: [author_id]
+          - type: AddedFieldDefinition
+            path: [author_id]
             value: "{{ record.get('authorId', '') }}"
-          - path: [last_editor_id]
+          - type: AddedFieldDefinition
+            path: [last_editor_id]
             value: "{{ record.get('version', {}).get('authorId', '') }}"
-          - path: [created_at]
+          - type: AddedFieldDefinition
+            path: [created_at]
             value: "{{ record.get('createdAt', '') }}"
-          - path: [updated_at]
+          - type: AddedFieldDefinition
+            path: [updated_at]
             value: "{{ record.get('version', {}).get('createdAt', '') }}"
-          - path: [version_number]
+          - type: AddedFieldDefinition
+            path: [version_number]
             value: "{{ record.get('version', {}).get('number', '') }}"
-          - path: [parent_page_id]
+          - type: AddedFieldDefinition
+            path: [parent_page_id]
             value: "{{ record.get('parentId', '') }}"
-          - path: [collected_at]
+          - type: AddedFieldDefinition
+            path: [collected_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%S.000Z') }}"
-          - path: [unique_key]
+          - type: AddedFieldDefinition
+            path: [unique_key]
             value: >-
               {{ config['insight_tenant_id'] }}-{{ config['insight_source_id']
               }}-{{ record['id'] }}
@@ -393,21 +415,29 @@ streams:
       - $ref: "#/definitions/add_fields"
       - type: AddFields
         fields:
-          - path: [page_id]
+          - type: AddedFieldDefinition
+            path: [page_id]
             value: "{{ stream_partition.page_id }}"
-          - path: [version_number]
+          - type: AddedFieldDefinition
+            path: [version_number]
             value: "{{ record.get('number', '') }}"
-          - path: [author_id]
+          - type: AddedFieldDefinition
+            path: [author_id]
             value: "{{ record.get('authorId', '') }}"
-          - path: [created_at]
+          - type: AddedFieldDefinition
+            path: [created_at]
             value: "{{ record.get('createdAt', '') }}"
-          - path: [message]
+          - type: AddedFieldDefinition
+            path: [message]
             value: "{{ record.get('message', '') }}"
-          - path: [minor_edit]
+          - type: AddedFieldDefinition
+            path: [minor_edit]
             value: "{{ record.get('minorEdit', false) }}"
-          - path: [collected_at]
+          - type: AddedFieldDefinition
+            path: [collected_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%S.000Z') }}"
-          - path: [unique_key]
+          - type: AddedFieldDefinition
+            path: [unique_key]
             value: >-
               {{ config['insight_tenant_id'] }}-{{ config['insight_source_id']
               }}-{{ stream_partition.page_id }}-{{ record['number'] }}
@@ -457,22 +487,30 @@ streams:
       - $ref: "#/definitions/add_fields"
       - type: AddFields
         fields:
-          - path: [comment_id]
+          - type: AddedFieldDefinition
+            path: [comment_id]
             value: "{{ record['id'] }}"
-          - path: [page_id]
+          - type: AddedFieldDefinition
+            path: [page_id]
             value: "{{ stream_partition.page_id }}"
-          - path: [author_id]
+          - type: AddedFieldDefinition
+            path: [author_id]
             value: "{{ (record.get('version', {}) or {}).get('authorId', '') }}"
-          - path: [created_at]
+          - type: AddedFieldDefinition
+            path: [created_at]
             value: "{{ (record.get('version', {}) or {}).get('createdAt', '') }}"
-          - path: [body_storage]
+          - type: AddedFieldDefinition
+            path: [body_storage]
             value: >-
               {{ (record.get('body', {}) or {}).get('storage', {}).get('value', '') }}
-          - path: [resolution_status]
+          - type: AddedFieldDefinition
+            path: [resolution_status]
             value: "{{ record.get('resolutionStatus', '') }}"
-          - path: [collected_at]
+          - type: AddedFieldDefinition
+            path: [collected_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%S.000Z') }}"
-          - path: [unique_key]
+          - type: AddedFieldDefinition
+            path: [unique_key]
             value: >-
               {{ config['insight_tenant_id'] }}-{{ config['insight_source_id']
               }}-{{ record['id'] }}
@@ -515,22 +553,30 @@ streams:
       - $ref: "#/definitions/add_fields"
       - type: AddFields
         fields:
-          - path: [comment_id]
+          - type: AddedFieldDefinition
+            path: [comment_id]
             value: "{{ record['id'] }}"
-          - path: [page_id]
+          - type: AddedFieldDefinition
+            path: [page_id]
             value: "{{ record.get('pageId', '') }}"
-          - path: [parent_comment_id]
+          - type: AddedFieldDefinition
+            path: [parent_comment_id]
             value: "{{ stream_partition.parent_comment_id }}"
-          - path: [author_id]
+          - type: AddedFieldDefinition
+            path: [author_id]
             value: "{{ (record.get('version', {}) or {}).get('authorId', '') }}"
-          - path: [created_at]
+          - type: AddedFieldDefinition
+            path: [created_at]
             value: "{{ (record.get('version', {}) or {}).get('createdAt', '') }}"
-          - path: [body_storage]
+          - type: AddedFieldDefinition
+            path: [body_storage]
             value: >-
               {{ (record.get('body', {}) or {}).get('storage', {}).get('value', '') }}
-          - path: [collected_at]
+          - type: AddedFieldDefinition
+            path: [collected_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%S.000Z') }}"
-          - path: [unique_key]
+          - type: AddedFieldDefinition
+            path: [unique_key]
             value: >-
               {{ config['insight_tenant_id'] }}-{{ config['insight_source_id']
               }}-{{ record['id'] }}
@@ -573,28 +619,38 @@ streams:
       - $ref: "#/definitions/add_fields"
       - type: AddFields
         fields:
-          - path: [comment_id]
+          - type: AddedFieldDefinition
+            path: [comment_id]
             value: "{{ record['id'] }}"
-          - path: [page_id]
+          - type: AddedFieldDefinition
+            path: [page_id]
             value: "{{ stream_partition.page_id }}"
-          - path: [author_id]
+          - type: AddedFieldDefinition
+            path: [author_id]
             value: "{{ (record.get('version', {}) or {}).get('authorId', '') }}"
-          - path: [created_at]
+          - type: AddedFieldDefinition
+            path: [created_at]
             value: "{{ (record.get('version', {}) or {}).get('createdAt', '') }}"
-          - path: [body_storage]
+          - type: AddedFieldDefinition
+            path: [body_storage]
             value: >-
               {{ (record.get('body', {}) or {}).get('storage', {}).get('value', '') }}
-          - path: [resolution_status]
+          - type: AddedFieldDefinition
+            path: [resolution_status]
             value: "{{ record.get('resolutionStatus', '') }}"
-          - path: [inline_marker_ref]
+          - type: AddedFieldDefinition
+            path: [inline_marker_ref]
             value: >-
               {{ (record.get('properties', {}) or {}).get('inlineMarkerRef', '') }}
-          - path: [inline_original_selection]
+          - type: AddedFieldDefinition
+            path: [inline_original_selection]
             value: >-
               {{ (record.get('properties', {}) or {}).get('inlineOriginalSelection', '') }}
-          - path: [collected_at]
+          - type: AddedFieldDefinition
+            path: [collected_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%S.000Z') }}"
-          - path: [unique_key]
+          - type: AddedFieldDefinition
+            path: [unique_key]
             value: >-
               {{ config['insight_tenant_id'] }}-{{ config['insight_source_id']
               }}-{{ record['id'] }}
@@ -637,22 +693,30 @@ streams:
       - $ref: "#/definitions/add_fields"
       - type: AddFields
         fields:
-          - path: [comment_id]
+          - type: AddedFieldDefinition
+            path: [comment_id]
             value: "{{ record['id'] }}"
-          - path: [page_id]
+          - type: AddedFieldDefinition
+            path: [page_id]
             value: "{{ record.get('pageId', '') }}"
-          - path: [parent_comment_id]
+          - type: AddedFieldDefinition
+            path: [parent_comment_id]
             value: "{{ stream_partition.parent_comment_id }}"
-          - path: [author_id]
+          - type: AddedFieldDefinition
+            path: [author_id]
             value: "{{ (record.get('version', {}) or {}).get('authorId', '') }}"
-          - path: [created_at]
+          - type: AddedFieldDefinition
+            path: [created_at]
             value: "{{ (record.get('version', {}) or {}).get('createdAt', '') }}"
-          - path: [body_storage]
+          - type: AddedFieldDefinition
+            path: [body_storage]
             value: >-
               {{ (record.get('body', {}) or {}).get('storage', {}).get('value', '') }}
-          - path: [collected_at]
+          - type: AddedFieldDefinition
+            path: [collected_at]
             value: "{{ now_utc().strftime('%Y-%m-%dT%H:%M:%S.000Z') }}"
-          - path: [unique_key]
+          - type: AddedFieldDefinition
+            path: [unique_key]
             value: >-
               {{ config['insight_tenant_id'] }}-{{ config['insight_source_id']
               }}-{{ record['id'] }}
diff --git a/src/ingestion/tools/declarative-connector/.env.local.example b/src/ingestion/tools/declarative-connector/.env.local.example
index 0ce32a736..6486183fd 100644
--- a/src/ingestion/tools/declarative-connector/.env.local.example
+++ b/src/ingestion/tools/declarative-connector/.env.local.example
@@ -1,6 +1,9 @@
 ## Optional: override default image tags and commands.
 ## Copy to .env.local to customize.
-# AIRBYTE_CONNECTOR_IMAGE=airbyte/source-declarative-manifest:local
-# AIRBYTE_BASE_IMAGE=airbyte/source-declarative-manifest:latest
+## AIRBYTE_BASE_IMAGE defaults to the CDK version bundled with the deployed
+## Airbyte platform (see BASE_IMAGE comment in source.sh) — do NOT use :latest,
+## its strict schema drifts ahead of the deployed Builder UI.
+# AIRBYTE_CONNECTOR_IMAGE=airbyte/source-declarative-manifest:local-6.60.9
+# AIRBYTE_BASE_IMAGE=airbyte/source-declarative-manifest:6.60.9
 # AIRBYTE_COMMAND=source-declarative-manifest
 # AIRBYTE_SECRETS_TMPFS_OPTS=/secrets:rw,mode=1777
diff --git a/src/ingestion/tools/declarative-connector/Dockerfile b/src/ingestion/tools/declarative-connector/Dockerfile
index 690550ab1..1ecc7bb3d 100644
--- a/src/ingestion/tools/declarative-connector/Dockerfile
+++ b/src/ingestion/tools/declarative-connector/Dockerfile
@@ -1,4 +1,6 @@
-ARG BASE_IMAGE=airbyte/source-declarative-manifest:latest
+# Default pin must match the CDK bundled with the deployed Airbyte platform —
+# see the BASE_IMAGE comment in source.sh. Bump together with AIRBYTE_VERSION.
+ARG BASE_IMAGE=airbyte/source-declarative-manifest:6.60.9
 FROM ${BASE_IMAGE}
 ARG AIRBYTE_COMMAND=source-declarative-manifest
 ENV AIRBYTE_COMMAND=${AIRBYTE_COMMAND}
diff --git a/src/ingestion/tools/declarative-connector/README.md b/src/ingestion/tools/declarative-connector/README.md
index d912e4f7f..f795a1241 100644
--- a/src/ingestion/tools/declarative-connector/README.md
+++ b/src/ingestion/tools/declarative-connector/README.md
@@ -6,18 +6,22 @@ Runs Airbyte declarative-manifest connectors in Docker without the full Airbyte
 
 | Command | Needs creds? | When to use |
 |---|---|---|
-| `./source.sh validate <class>/<connector>` | no | CDK-runtime validation. Resolves `$ref` before checking. **Lenient** — passes manifests the Builder UI rejects. |
-| `./source.sh validate-strict <class>/<connector>` | no | **Strict** Builder-UI validation — runs the manifest through `declarative_component_schema.yaml` **without `$ref` resolution**, emitting per-path errors. Run this **before** opening the manifest in the Airbyte Builder UI. |
+| `./source.sh validate <class>/<connector>` | no | CDK-runtime validation. Resolves `$ref` AND normalizes (auto-fills `type:` etc.) before checking. **Lenient** — passes manifests the Builder UI rejects. |
+| `./source.sh validate-strict <class>/<connector>` | no | **Strict** Builder-UI validation — resolves `$ref`s with the CDK resolver, then validates against `declarative_component_schema.yaml` **without normalization**, emitting per-path errors. Run this **before** opening the manifest in the Airbyte Builder UI. |
 | `./source.sh check <class>/<connector> <tenant>` | yes | Manifest + credentials smoke test against the source API. |
 | `./source.sh discover <class>/<connector> <tenant>` | yes | List available streams and their schemas. |
 | `./source.sh read <class>/<connector> <tenant>` | yes | Extract data (Airbyte Protocol JSON on stdout). |
 
 `<class>/<connector>` is relative to `src/ingestion/connectors/`, e.g. `collaboration/m365` or `task-tracking/youtrack`.
 
+## Validator image pin
+
+All commands run inside `airbyte/source-declarative-manifest` pinned in `source.sh` (`BASE_IMAGE`) to the `airbyte-cdk` version bundled with the **deployed** Airbyte platform — the version in `airbyte-connector-builder-server/requirements.in` of the `airbyte-platform` release matching `AIRBYTE_VERSION` in `deploy/scripts/install-airbyte.sh`. Never switch the default to `:latest`: upstream tightens `declarative_component_schema.yaml` over time, so a newer schema rejects manifests the deployed Builder UI happily accepts (false `validate-strict` failures). When bumping `AIRBYTE_VERSION`, bump the pin in `source.sh` + `Dockerfile` in the same change; the local wrapper image tag derives from the pin, so the rebuild happens automatically.
+
 ## Validation ladder — always in this order
 
-1. **`validate-strict`** — first. Catches Builder UI blockers (`$ref` misuse, missing `type: AddedFieldDefinition`, integer-only slots templated by mistake, bad `$schema` URL, etc.) early, before runtime wastes a round trip.
-2. **`validate`** — second. Smoke-checks that the CDK loader accepts the manifest at runtime, after `$ref` resolution.
+1. **`validate-strict`** — first. Catches Builder UI blockers (unresolvable `$ref`, missing `type: AddedFieldDefinition`, integer-only slots templated by mistake, bad `$schema` URL, etc.) early, before runtime wastes a round trip.
+2. **`validate`** — second. Smoke-checks that the CDK loader accepts the manifest at runtime, after `$ref` resolution and normalization.
 3. **`check <tenant>`** — third. Real credentials against the real API. Catches query-syntax errors and auth problems.
 4. **`discover` / `read`** — fourth. Produces real records locally; feeds `generate-schema.sh`.
 
@@ -25,31 +29,11 @@ If any step in the ladder fails, fix the issue and restart from step 1. **Do not
 
 ## Builder-UI compatibility — hard rules
 
-The Airbyte Builder UI validates manifests against `declarative_component_schema.yaml` with **no `$ref` resolution**. A manifest can load fine via the CDK and still be rejected by the Builder. Keep these rules in mind when authoring manifests or when copying from another connector package:
-
-### Rule 1 — No whole-object `$ref`
-
-❌ **Forbidden** (the `task-tracking/jira` manifest uses this pattern — **do not copy from it**):
-
-```yaml
-requester:
-  $ref: "#/definitions/base_requester"
-paginator:
-  $ref: "#/definitions/paginator"
-```
+The Builder UI resolves `$ref`s with the CDK's `ManifestReferenceResolver` and then validates the resolved manifest against `declarative_component_schema.yaml` — but **without** the runtime's `ManifestNormalizer` pass that auto-fills missing `type:` fields and similar. `validate-strict` reproduces exactly that: resolve refs, then raw schema check. A manifest can load fine via the runtime CDK (`validate`) and still be rejected by the Builder. Keep these rules in mind when authoring manifests or when copying from another connector package:
 
-```yaml
-parent_stream_configs:
-  - stream:
-      $ref: "#/streams/4"        # substream-parent-by-ref
-```
-
-```yaml
-transformations:
-  - $ref: "#/definitions/add_fields"   # transformation-by-ref
-```
+### Rule 1 — Prefer field-level `$ref` into `definitions.linked`
 
-✅ **Allowed** — granular, field-level `$ref` into `definitions.linked.<Component>/<field>`:
+Any resolvable `$ref` passes `validate-strict` (the resolver inlines it before the schema check, including `$ref`-with-sibling-keys merge). For **project style**, keep shared values in `definitions.linked.<Component>/<field>` — the Builder's own shared-components store, round-tripped losslessly by the UI:
 
 ```yaml
 definitions:
@@ -77,7 +61,7 @@ streams:
         path: /widgets
 ```
 
-For anything that cannot be expressed as a leaf-field `$ref` (full requester, paginator, retriever, stream), **inline the full definition** or duplicate it across streams. Repetition is the price of Builder compatibility.
+Whole-object `$ref`s (full requester, paginator, stream — the `task-tracking/jira` style, also what the Builder export itself emits as `streams: [{$ref: "#/definitions/streams/<name>"}]`) resolve fine and pass the gate; for new in-house manifests still prefer linked field refs or inlining — resolved whole-object refs lose their sharing on a Builder round-trip.
 
 ### Rule 2 — `type: AddedFieldDefinition` on every `AddFields.fields[]` item
 
@@ -233,10 +217,10 @@ Keep both `%ms` (for live record values) and `%Y-%m-%dT%H:%M:%S` (for persisted
 → `streams[0].transformations[0].fields[3]` is missing `type: AddedFieldDefinition`.
 
 ```
-[1] streams/0/retriever/requester: 'type' is a required property
+STRICT VALIDATION FAILED — unresolvable $ref: Undefined reference #/definitions/base_requester
 ```
 
-→ The `requester` is a `$ref` (opaque to the Builder validator). Inline or use granular leaf-field `$ref`.
+→ A `$ref` points at a path that does not exist in the manifest. Fix the path or define the target. (Resolvable `$ref`s are inlined before validation — error paths always refer to the **resolved** manifest.)
 
 ```
 [1] concurrency_level/default_concurrency: "{{ config.get('x_concurrency', 1) }}" is not of type 'integer'
@@ -251,7 +235,7 @@ docker run --rm \
   -v "$PWD/src/ingestion/connectors/<class>/<connector>:/input:ro" \
   -v "$PWD/src/ingestion/tools/declarative-connector/validate_strict.py:/scripts/validate_strict.py:ro" \
   --entrypoint=/bin/sh \
-  airbyte/source-declarative-manifest:local \
+  airbyte/source-declarative-manifest:local-6.60.9 \
   -c "python3 /scripts/validate_strict.py /input/connector.yaml"
 ```
 
diff --git a/src/ingestion/tools/declarative-connector/source.sh b/src/ingestion/tools/declarative-connector/source.sh
index 6c1e683be..d3ad49377 100755
--- a/src/ingestion/tools/declarative-connector/source.sh
+++ b/src/ingestion/tools/declarative-connector/source.sh
@@ -18,11 +18,13 @@ set -euo pipefail
 #
 # validate vs validate-strict:
 #   - `validate` passes if the CDK loader accepts the manifest at runtime. It is
-#     lenient and resolves `$ref` before validation, so it will happily accept
-#     manifests that the Airbyte Builder UI rejects.
-#   - `validate-strict` runs the manifest through the Airbyte Builder JSON-schema
-#     validator (no `$ref` resolution). Use this before attempting to open the
-#     manifest in the Builder UI. Emits per-path error messages.
+#     lenient: it resolves `$ref` AND normalizes (auto-fills missing `type:`
+#     fields etc.) before validation, so it will happily accept manifests that
+#     the Airbyte Builder UI rejects.
+#   - `validate-strict` resolves `$ref`s with the CDK resolver, then validates
+#     against declarative_component_schema.yaml WITHOUT the normalization pass —
+#     matching the Builder UI. Use this before attempting to open the manifest
+#     in the Builder UI. Emits per-path error messages.
 #
 # Example:
 #   $0 validate        collaboration/m365
@@ -39,8 +41,15 @@ if [[ -f "${GLOBAL_ENV_FILE}" ]]; then
   set -a; source "${GLOBAL_ENV_FILE}"; set +a
 fi
 
-IMAGE="${AIRBYTE_CONNECTOR_IMAGE:-airbyte/source-declarative-manifest:local}"  # RULE-DEFAULTS-OK: see file header
-BASE_IMAGE="${AIRBYTE_BASE_IMAGE:-airbyte/source-declarative-manifest:latest}"  # RULE-DEFAULTS-OK: see file header
+# BASE_IMAGE is pinned to the airbyte-cdk version bundled with the deployed
+# Airbyte platform (AIRBYTE_VERSION in deploy/scripts/install-airbyte.sh ->
+# airbyte-platform/airbyte-connector-builder-server/requirements.in). This
+# keeps validate-strict aligned with the Builder-UI schema of the platform we
+# deploy to; `:latest` drifts and rejects manifests the deployed Builder
+# accepts. Bump this pin together with AIRBYTE_VERSION.
+BASE_IMAGE="${AIRBYTE_BASE_IMAGE:-airbyte/source-declarative-manifest:6.60.9}"  # RULE-DEFAULTS-OK: see file header
+# Wrapper tag derives from the base tag so bumping the pin auto-rebuilds.
+IMAGE="${AIRBYTE_CONNECTOR_IMAGE:-airbyte/source-declarative-manifest:local-${BASE_IMAGE##*:}}"  # RULE-DEFAULTS-OK: see file header
 COMMAND_NAME="${AIRBYTE_COMMAND:-source-declarative-manifest}"  # RULE-DEFAULTS-OK: see file header
 SECRETS_TMPFS_OPTS="${AIRBYTE_SECRETS_TMPFS_OPTS:-/secrets:rw,mode=1777}"  # RULE-DEFAULTS-OK: see file header
 
@@ -55,11 +64,12 @@ Usage:
 
 Commands:
   validate        Runtime validation — passes if the CDK loader accepts the
-                  manifest. Resolves \$ref before checking. Lenient.
-  validate-strict Strict Builder-UI validation — runs the manifest through the
-                  declarative_component_schema.yaml validator WITHOUT \$ref
-                  resolution. Use this before opening the manifest in the
-                  Airbyte Builder UI. Reports per-path errors.
+                  manifest. Resolves \$ref and normalizes before checking.
+                  Lenient.
+  validate-strict Strict Builder-UI validation — resolves \$refs, then runs the
+                  declarative_component_schema.yaml validator WITHOUT the
+                  CDK normalization pass. Use this before opening the manifest
+                  in the Airbyte Builder UI. Reports per-path errors.
   check           Manifest + credentials against source API (smoke test)
   discover        List available streams and their schemas
   read            Extract data (outputs Airbyte Protocol JSON to stdout)
@@ -216,7 +226,7 @@ case "${command}" in
     ;;
 
   validate-strict)
-    # Run the exact jsonschema check the Builder UI performs (no $ref resolution).
+    # Run the jsonschema check the Builder UI performs ($ref resolution, no normalization).
     # Logic lives in validate_strict.py (per the no-inline-Python rule in
     # cypilot/config/rules/code-conventions.md §"No inline scripts"); this case
     # just mounts the script + the connector dir and invokes it inside the
diff --git a/src/ingestion/tools/declarative-connector/validate_strict.py b/src/ingestion/tools/declarative-connector/validate_strict.py
index 90b73f1d2..78969ea1a 100644
--- a/src/ingestion/tools/declarative-connector/validate_strict.py
+++ b/src/ingestion/tools/declarative-connector/validate_strict.py
@@ -1,9 +1,17 @@
 #!/usr/bin/env python3
 """Builder-UI strict validator for declarative Airbyte manifests.
 
-Runs the exact jsonschema check the Airbyte Builder UI performs — no `$ref`
-resolution — and emits per-path errors with the deepest-matching `oneOf` branch
-context so the user can pinpoint bad fields. Exits non-zero on any error.
+Runs the jsonschema check the Airbyte Builder UI performs and emits per-path
+errors with the deepest-matching `oneOf` branch context so the user can
+pinpoint bad fields. Exits non-zero on any error.
+
+`$ref` handling mirrors the Builder UI: the manifest is preprocessed with the
+CDK's own `ManifestReferenceResolver` (the exact resolver the Builder and the
+runtime use, including `$ref`-with-sibling-keys merge semantics) and the
+resolved manifest is then validated against `declarative_component_schema.yaml`
+as-is. Unlike the runtime `validate` command, NO `ManifestNormalizer` /
+type-inference pass runs first — so missing `type:` fields, templated
+integer-only slots, and similar Builder blockers still surface.
 
 Designed to run inside the `airbyte/source-declarative-manifest` Docker image,
 which ships the `airbyte_cdk` package, `pyyaml`, and `jsonschema`. The schema
@@ -22,6 +30,9 @@
 import airbyte_cdk
 import jsonschema
 import yaml
+from airbyte_cdk.sources.declarative.parsers.manifest_reference_resolver import (
+    ManifestReferenceResolver,
+)
 
 
 def _resolve_schema_path() -> Path:
@@ -77,6 +88,11 @@ def main(argv: list[str]) -> int:
         schema = yaml.safe_load(f)
     with manifest_path.open() as f:
         manifest = yaml.safe_load(f)
+    try:
+        manifest = ManifestReferenceResolver().preprocess_manifest(manifest)
+    except Exception as exc:  # noqa: BLE001 — any unresolvable $ref is a hard failure
+        print(f"STRICT VALIDATION FAILED — unresolvable $ref: {exc}", file=sys.stderr)
+        return 1
 
     validator = jsonschema.Draft7Validator(schema)
     errors = list(validator.iter_errors(manifest))