fix(dbt): align staging contributor types with their silver union targets

src/ingestion/connectors/ai/claude-enterprise/dbt/claude_enterprise__ai_dev_usage.sql

@@ -56,14 +56,14 @@ SELECT
     toUInt32(coalesce(code_session_count, 0))                          AS session_count,
     toUInt32OrNull(toString(code_session_count))                       AS conversation_count,
     toUInt32(coalesce(code_lines_added, 0))                            AS lines_added,
-    toUInt32(coalesce(code_lines_removed, 0))                          AS lines_removed,
+    toNullable(toUInt32(coalesce(code_lines_removed, 0)))              AS lines_removed,
     -- Enterprise reports AI-accepted lines only — no view of total user keystrokes.
     -- ai_loc_share_pct downstream filters to tool='cursor' precisely because of this gap.
     CAST(NULL AS Nullable(UInt32))                                     AS total_lines_added,
     CAST(NULL AS Nullable(UInt32))                                     AS total_lines_removed,
-    toUInt32(coalesce(code_tool_accepted_count, 0)
-           + coalesce(code_tool_rejected_count, 0))                    AS tool_use_offered,
-    toUInt32(coalesce(code_tool_accepted_count, 0))                    AS tool_use_accepted,
+    toNullable(toUInt32(coalesce(code_tool_accepted_count, 0)
+           + coalesce(code_tool_rejected_count, 0)))                   AS tool_use_offered,
+    toNullable(toUInt32(coalesce(code_tool_accepted_count, 0)))        AS tool_use_accepted,
     -- #262: `completions_count` dropped from class_ai_dev_usage — it was
     -- numerically identical to tool_use_accepted (both = code_tool_accepted_count).
     -- agent_sessions := no direct Enterprise equivalent. Enterprise's cowork_dispatch_turn_count
@@ -78,8 +78,8 @@ SELECT
     -- Cost is not surfaced per-user in Enterprise; tied to org subscription, not consumption.
     CAST(NULL AS Nullable(UInt32))                                     AS cost_cents,
     -- Enterprise exposes commit and PR counts per user per day via core_metrics.
-    toUInt32(coalesce(code_commit_count, 0))                           AS commits_count,
-    toUInt32(coalesce(code_pull_request_count, 0))                     AS pull_requests_count,
+    toNullable(toUInt32(coalesce(code_commit_count, 0)))               AS commits_count,
+    toNullable(toUInt32(coalesce(code_pull_request_count, 0)))         AS pull_requests_count,
     -- prs_with_cc_count / prs_total_count: Claude Team-only (Anthropic GitHub-app attribution).
     -- Enterprise exposes code_pull_request_count (above, → pull_requests_count) but not the
     -- GitHub-app split between "PRs with CC active" and "total PRs in window".

src/ingestion/connectors/ai/cursor/dbt/cursor__ai_dev_usage.sql

@@ -87,18 +87,18 @@ SELECT
     toUInt32(1)                                     AS session_count,
     CAST(NULL AS Nullable(UInt32))                  AS conversation_count,
     toUInt32(coalesce(d.acceptedLinesAdded, 0))     AS lines_added,
-    toUInt32(coalesce(d.acceptedLinesDeleted, 0))   AS lines_removed,
+    toNullable(toUInt32(coalesce(d.acceptedLinesDeleted, 0)))   AS lines_removed,
     -- total_lines_added/removed = ALL lines the user wrote/deleted that day
     -- (not just AI-accepted ones). Needed by gold metrics like
     -- ai_loc_share = accepted/total to express AI contribution percentage.
-    toUInt32(coalesce(d.totalLinesAdded, 0))        AS total_lines_added,
-    toUInt32(coalesce(d.totalLinesDeleted, 0))      AS total_lines_removed,
+    toNullable(toUInt32(coalesce(d.totalLinesAdded, 0)))        AS total_lines_added,
+    toNullable(toUInt32(coalesce(d.totalLinesDeleted, 0)))      AS total_lines_removed,
     toUInt32OrNull(toString(d.totalTabsShown))      AS tool_use_offered,
     toUInt32OrNull(toString(d.totalTabsAccepted))   AS tool_use_accepted,
     -- #262: `completions_count` was numerically identical to tool_use_accepted
     -- (both = totalTabsAccepted) and dropped from class_ai_dev_usage.
     toUInt32OrNull(toString(d.agentRequests))       AS agent_sessions,
-    toUInt32(coalesce(d.chatRequests, 0) + coalesce(d.composerRequests, 0))
+    toNullable(toUInt32(coalesce(d.chatRequests, 0) + coalesce(d.composerRequests, 0)))
                                                     AS chat_requests,
     -- Rank guard: cost is aggregated per (tenant, source, email, day) because
     -- Cursor's events carry only userEmail, while a row is identified by userId.

src/ingestion/connectors/ai/github-copilot/dbt/copilot__ai_dev_usage.sql

@@ -110,15 +110,15 @@ SELECT
     toUInt32(1)                                                            AS session_count,
     CAST(NULL AS Nullable(UInt32))                                         AS conversation_count,
     toUInt32(coalesce(m.loc_added_sum, 0))                                 AS lines_added,
-    toUInt32(coalesce(m.loc_deleted_sum, 0))                               AS lines_removed,
+    toNullable(toUInt32(coalesce(m.loc_deleted_sum, 0)))                   AS lines_removed,
     -- See header comment — Copilot reports AI-accepted lines only.
     CAST(NULL AS Nullable(UInt32))                                         AS total_lines_added,
     CAST(NULL AS Nullable(UInt32))                                         AS total_lines_removed,
     -- code_generation_activity_count = number of code generation events
     -- (a proxy for "suggestions offered"). The API does not split offered
     -- from rejected, so this is the closest mappable signal.
-    toUInt32(coalesce(m.code_generation_activity_count, 0))                AS tool_use_offered,
-    toUInt32(coalesce(m.code_acceptance_activity_count, 0))                AS tool_use_accepted,
+    toNullable(toUInt32(coalesce(m.code_generation_activity_count, 0)))    AS tool_use_offered,
+    toNullable(toUInt32(coalesce(m.code_acceptance_activity_count, 0)))    AS tool_use_accepted,
     -- #262: `completions_count` dropped from class_ai_dev_usage — it was numerically identical
     -- to tool_use_accepted (code_acceptance_activity_count). Same drop applied to cursor and
     -- claude_enterprise in PR #262; copilot now aligned.
@@ -144,12 +144,12 @@ SELECT
     -- column slot) and re-surfaces the chat/agent flags for downstream
     -- consumers that want raw boolean state rather than the marker
     -- counts above.
-    concat(
+    toNullable(concat(
         '{"used_chat":',  if(coalesce(m.used_chat,  false), 'true', 'false'),
         ',"used_agent":', if(coalesce(m.used_agent, false), 'true', 'false'),
         ',"used_cli":',   if(coalesce(m.used_cli,   false), 'true', 'false'),
         '}'
-    )                                                                      AS tool_action_breakdown_json,
+    ))                                                                     AS tool_action_breakdown_json,
     'copilot'                                                              AS source,
     'insight_github_copilot'                                               AS data_source,
     parseDateTime64BestEffortOrNull(coalesce(m.collected_at, ''), 3)       AS collected_at,

src/ingestion/connectors/collaboration/m365/dbt/m365__collab_document_activity_sharepoint.sql

@@ -28,7 +28,7 @@ SELECT
     syncedFileCount AS synced_count,
     sharedInternallyFileCount AS shared_internally_count,
     sharedExternallyFileCount AS shared_externally_count,
-    visitedPageCount AS visited_page_count,
+    CAST(visitedPageCount AS Nullable(Int64)) AS visited_page_count,
     reportPeriod AS report_period,
     now() AS collected_at,
     'insight_m365' AS data_source,

src/ingestion/connectors/collaboration/m365/dbt/m365__collab_meeting_activity.sql

@@ -23,18 +23,18 @@ SELECT
     -- when both feeders UNION ALL into silver.class_collab_meeting_activity.
     -- Bronze stores them as Decimal(38, 9) / Float64; CH 25.3 refuses
     -- Int64 ∪ Decimal/Float with NO_COMMON_TYPE.
-    toInt64(coalesce(callCount, 0)) AS calls_count,
-    toInt64(coalesce(meetingsOrganizedCount, 0)) AS meetings_organized,
+    toNullable(toInt64(coalesce(callCount, 0))) AS calls_count,
+    toNullable(toInt64(coalesce(meetingsOrganizedCount, 0))) AS meetings_organized,
     toInt64(coalesce(meetingsAttendedCount, 0)) AS meetings_attended,
-    toInt64(coalesce(adHocMeetingsOrganizedCount, 0)) AS adhoc_meetings_organized,
-    toInt64(coalesce(adHocMeetingsAttendedCount, 0)) AS adhoc_meetings_attended,
-    toInt64(COALESCE(scheduledOneTimeMeetingsOrganizedCount, 0)
-        + COALESCE(scheduledRecurringMeetingsOrganizedCount, 0)) AS scheduled_meetings_organized,
-    toInt64(COALESCE(scheduledOneTimeMeetingsAttendedCount, 0)
-        + COALESCE(scheduledRecurringMeetingsAttendedCount, 0)) AS scheduled_meetings_attended,
-    toInt64({{ iso8601_duration_seconds("ifNull(audioDuration, 'PT0S')") }}) AS audio_duration_seconds,
-    toInt64({{ iso8601_duration_seconds("ifNull(videoDuration, 'PT0S')") }}) AS video_duration_seconds,
-    toInt64({{ iso8601_duration_seconds("ifNull(screenShareDuration, 'PT0S')") }}) AS screen_share_duration_seconds,
+    toNullable(toInt64(coalesce(adHocMeetingsOrganizedCount, 0))) AS adhoc_meetings_organized,
+    toNullable(toInt64(coalesce(adHocMeetingsAttendedCount, 0))) AS adhoc_meetings_attended,
+    toNullable(toInt64(COALESCE(scheduledOneTimeMeetingsOrganizedCount, 0)
+        + COALESCE(scheduledRecurringMeetingsOrganizedCount, 0))) AS scheduled_meetings_organized,
+    toNullable(toInt64(COALESCE(scheduledOneTimeMeetingsAttendedCount, 0)
+        + COALESCE(scheduledRecurringMeetingsAttendedCount, 0))) AS scheduled_meetings_attended,
+    toNullable(toInt64({{ iso8601_duration_seconds("ifNull(audioDuration, 'PT0S')") }})) AS audio_duration_seconds,
+    toNullable(toInt64({{ iso8601_duration_seconds("ifNull(videoDuration, 'PT0S')") }})) AS video_duration_seconds,
+    toNullable(toInt64({{ iso8601_duration_seconds("ifNull(screenShareDuration, 'PT0S')") }})) AS screen_share_duration_seconds,
     reportPeriod AS report_period,
     now() AS collected_at,
     'insight_m365' AS data_source,

src/ingestion/connectors/collaboration/slack/dbt/slack__collab_chat_activity.sql

@@ -38,19 +38,19 @@ SELECT
         toString(toDate(parseDateTimeBestEffortOrNull(u.date)))
     )) AS unique_key,
     u.user_id,
-    coalesce(u.email_address, '') AS user_name,
-    coalesce(u.email_address, '') AS email,
+    toNullable(coalesce(u.email_address, '')) AS user_name,
+    toNullable(coalesce(u.email_address, '')) AS email,
     if(coalesce(u.email_address, '') != '',
        lower(u.email_address),
        lower(u.user_id)) AS person_key,
     toDate(parseDateTimeBestEffortOrNull(u.date)) AS date,
     CAST(NULL AS Nullable(Int64)) AS direct_messages,
     CAST(NULL AS Nullable(Int64)) AS group_chat_messages,
     -- #266: total - channel = DMs + MPIMs (everything posted outside channels).
-    toInt64(greatest(
+    toNullable(toInt64(greatest(
         coalesce(u.messages_posted_count, 0) - coalesce(u.channel_messages_posted_count, 0),
         0
-    )) AS direct_and_group_messages,
+    ))) AS direct_and_group_messages,
     coalesce(u.messages_posted_count, 0) AS total_chat_messages,
     u.channel_messages_posted_count AS channel_posts,
     CAST(NULL AS Nullable(Int64)) AS channel_replies,

src/ingestion/connectors/collaboration/zoom/dbt/zoom__collab_meeting_activity.sql

@@ -87,7 +87,7 @@ SELECT
     -- this, GROUP BY would split them and produce two rows with identical
     -- unique_key — the staging model's `unique_key` is keyed on
     -- (tenant, source, lower(email), date), so user_name is non-keying.
-    coalesce(any(p.user_name), '') AS user_name,
+    toNullable(coalesce(any(p.user_name), '')) AS user_name,
     p.email AS email,
     lower(p.email) AS person_key,
     toDate(parseDateTimeBestEffortOrNull(p.join_time)) AS date,

src/ingestion/connectors/collaboration/zulip-proxy/dbt/zulip_proxy__collab_chat_activity.sql

@@ -40,7 +40,7 @@ SELECT
         toString(toDate(parseDateTimeBestEffortOrNull(m.created_at)))
     )) AS unique_key,
     lower(u.email) AS user_id,
-    coalesce(any(u.full_name), '') AS user_name,
+    toNullable(coalesce(any(u.full_name), '')) AS user_name,
     lower(u.email) AS email,
     lower(u.email) AS person_key,
     toDate(parseDateTimeBestEffortOrNull(m.created_at)) AS date,

src/ingestion/connectors/crm/hubspot/dbt/hubspot__crm_deals.sql

@@ -49,7 +49,7 @@ WITH src AS (
         -- "2025-10-23T08:49:39Z"). `toDateOrNull` only handles
         -- YYYY-MM-DD; we parse via `parseDateTime64BestEffortOrNull`
         -- first then truncate to Date.
-        toDate(parseDateTime64BestEffortOrNull(properties_closedate)) AS close_date,
+        toDate32(parseDateTime64BestEffortOrNull(properties_closedate)) AS close_date,
         properties_hubspot_owner_id                     AS owner_id,
         -- Rep who logged / created the deal — distinct from owner_id, which
         -- can be the contact owner. Resolves to silver.class_crm_users
@@ -58,8 +58,8 @@ WITH src AS (
         nullIf(arrayElement(
             JSONExtract(coalesce(associations_companies, '[]'), 'Array(String)'), 1
         ), '')                                          AS account_id,
-        toInt64(coalesce(properties_hs_is_closed, 'false') = 'true')     AS is_closed,
-        toInt64(coalesce(properties_hs_is_closed_won, 'false') = 'true') AS is_won,
+        toNullable(toInt64(coalesce(properties_hs_is_closed, 'false') = 'true'))     AS is_closed,
+        toNullable(toInt64(coalesce(properties_hs_is_closed_won, 'false') = 'true')) AS is_won,
         properties_hs_analytics_source                  AS lead_source,
         toFloat64OrNull(properties_hs_deal_stage_probability) AS probability,
         properties_dealtype                             AS deal_type,

src/ingestion/connectors/crm/hubspot/dbt/hubspot__crm_users.sql

@@ -49,7 +49,7 @@ WITH src AS (
         -- the columns to exist so emit explicit NULLs.
         CAST(NULL AS Nullable(String))                  AS title,
         CAST(NULL AS Nullable(String))                  AS department,
-        toInt64(NOT coalesce(archived, false))          AS is_active,
+        toNullable(toInt64(NOT coalesce(archived, false)))          AS is_active,
         toJSONString(map(
             'userId',   coalesce(toString(userId), ''),
             'archived', toString(coalesce(archived, false))

src/ingestion/connectors/crm/salesforce/dbt/salesforce__crm_accounts.sql

@@ -29,7 +29,7 @@ WITH src AS (
             'AnnualRevenue',     coalesce(toString(AnnualRevenue), ''),
             'IsDeleted',         toString(coalesce(IsDeleted, false))
         ))                                              AS metadata,
-        custom_fields,
+        coalesce(custom_fields, '{}')                   AS custom_fields,
         CreatedDate                                     AS created_at,
         LastModifiedDate                                AS updated_at,
         data_source,

src/ingestion/connectors/crm/salesforce/dbt/salesforce__crm_activities.sql

@@ -48,7 +48,7 @@ WITH tasks AS (
             'CallType',    coalesce(toString(CallType), ''),
             'IsDeleted',   toString(coalesce(IsDeleted, false))
         ))                                          AS metadata,
-        custom_fields,
+        coalesce(custom_fields, '{}')                   AS custom_fields,
         CreatedDate                                 AS created_at,
         data_source,
         coalesce(toUnixTimestamp64Milli(SystemModstamp), 0) AS _version
@@ -93,7 +93,7 @@ events AS (
             'EventSubtype', coalesce(toString(EventSubtype), ''),
             'IsDeleted',    toString(coalesce(IsDeleted, false))
         ))                                          AS metadata,
-        custom_fields,
+        coalesce(custom_fields, '{}')                   AS custom_fields,
         CreatedDate                                 AS created_at,
         data_source,
         coalesce(toUnixTimestamp64Milli(SystemModstamp), 0) AS _version

src/ingestion/connectors/crm/salesforce/dbt/salesforce__crm_contacts.sql

@@ -27,7 +27,7 @@ WITH src AS (
             'LeadSource', coalesce(toString(LeadSource), ''),
             'IsDeleted',  toString(coalesce(IsDeleted, false))
         ))                                              AS metadata,
-        custom_fields,
+        coalesce(custom_fields, '{}')                   AS custom_fields,
         CreatedDate                                     AS created_at,
         LastModifiedDate                                AS updated_at,
         data_source,

src/ingestion/connectors/crm/salesforce/dbt/salesforce__crm_deals.sql

@@ -55,7 +55,7 @@ WITH src AS (
             'Type',      coalesce(toString(Type), ''),
             'IsDeleted', if(coalesce(IsDeleted, false), 'true', 'false')
         ))                                              AS metadata,
-        custom_fields,
+        coalesce(custom_fields, '{}')                   AS custom_fields,
         CreatedDate                                     AS created_at,
         LastModifiedDate                                AS updated_at,
         data_source,

src/ingestion/connectors/crm/salesforce/dbt/salesforce__crm_users.sql

@@ -29,7 +29,7 @@ WITH src AS (
             'Username',   coalesce(toString(Username), ''),
             'UserRoleId', coalesce(toString(UserRoleId), '')
         ))                                              AS metadata,
-        custom_fields,
+        coalesce(custom_fields, '{}')                   AS custom_fields,
         collected_at,
         data_source,
         coalesce(toUnixTimestamp64Milli(SystemModstamp), 0) AS _version

src/ingestion/connectors/git/github-v2/dbt/github__commits.sql

@@ -21,9 +21,9 @@ SELECT
     COALESCE(committer_email, '') AS committer_email,
     COALESCE(message, '') AS message,
     parseDateTimeBestEffortOrNull(committed_date) AS date,
-    COALESCE(changed_files, 0) AS files_changed,
-    COALESCE(additions, 0) AS lines_added,
-    COALESCE(deletions, 0) AS lines_removed,
+    toNullable(COALESCE(changed_files, 0)) AS files_changed,
+    toNullable(COALESCE(additions, 0)) AS lines_added,
+    toNullable(COALESCE(deletions, 0)) AS lines_removed,
     -- parent_hashes arrives as a JSON-array string (Airbyte serializes the
     -- connector's array field into the Nullable(String) bronze column), so
     -- count elements with JSONLength — plain length() would count characters

src/ingestion/connectors/git/github-v2/dbt/github__file_changes.sql

@@ -31,8 +31,8 @@ SELECT
         ''
     ) AS file_extension,
     COALESCE(status, '') AS change_type,
-    COALESCE(additions, 0) AS lines_added,
-    COALESCE(deletions, 0) AS lines_removed,
+    toNullable(COALESCE(additions, 0)) AS lines_added,
+    toNullable(COALESCE(deletions, 0)) AS lines_removed,
     COALESCE(source_type, '') AS source_type,
     'insight_github' AS data_source,
     toUnixTimestamp64Milli(now64()) AS _version,

src/ingestion/connectors/git/github-v2/dbt/github__pull_requests.sql

@@ -26,9 +26,9 @@ SELECT
     parseDateTimeBestEffortOrNull(updated_at) AS updated_on,
     parseDateTimeBestEffortOrNull(closed_at) AS closed_on,
     COALESCE(merge_commit_sha, '') AS merge_commit_hash,
-    COALESCE(changed_files, 0) AS files_changed,
-    COALESCE(additions, 0) AS lines_added,
-    COALESCE(deletions, 0) AS lines_removed,
+    toNullable(COALESCE(changed_files, 0)) AS files_changed,
+    toNullable(COALESCE(additions, 0)) AS lines_added,
+    toNullable(COALESCE(deletions, 0)) AS lines_removed,
     'insight_github' AS data_source,
     toUnixTimestamp64Milli(now64()) AS _version,
     _airbyte_extracted_at

src/ingestion/connectors/git/github-v2/dbt/github__pull_requests_commits.sql

@@ -15,7 +15,7 @@ SELECT
     COALESCE(repo_name, '') AS repo_slug,
     COALESCE(pull_request_id, 0) AS pr_id,
     COALESCE(sha, '') AS commit_hash,
-    0 AS commit_order,
+    toInt64(0) AS commit_order,
     'insight_github' AS data_source,
     toUnixTimestamp64Milli(now64()) AS _version,
     _airbyte_extracted_at

src/ingestion/connectors/git/gitlab/dbt/gitlab__commits.sql

@@ -47,9 +47,9 @@ SELECT
     COALESCE(c.committer_email, '') AS committer_email,
     COALESCE(c.message, '') AS message,
     parseDateTimeBestEffortOrNull(c.committed_date) AS date,
-    toInt64(COALESCE(f.files_changed, 0)) AS files_changed,
-    COALESCE(c.stats_additions, 0) AS lines_added,
-    COALESCE(c.stats_deletions, 0) AS lines_removed,
+    toNullable(toInt64(COALESCE(f.files_changed, 0))) AS files_changed,
+    toNullable(COALESCE(c.stats_additions, 0)) AS lines_added,
+    toNullable(COALESCE(c.stats_deletions, 0)) AS lines_removed,
     if(COALESCE(c.parent_count, 0) > 1, 1, 0) AS is_merge_commit,
     'insight_gitlab' AS data_source,
     toUnixTimestamp64Milli(now64()) AS _version,

src/ingestion/connectors/git/gitlab/dbt/gitlab__file_changes.sql

@@ -39,8 +39,8 @@ SELECT
         fc.renamed_file = true, 'renamed',
         'modified'
     ) AS change_type,
-    COALESCE(fc.lines_added, 0) AS lines_added,
-    COALESCE(fc.lines_removed, 0) AS lines_removed,
+    toNullable(COALESCE(fc.lines_added, 0)) AS lines_added,
+    toNullable(COALESCE(fc.lines_removed, 0)) AS lines_removed,
     '' AS source_type,
     'insight_gitlab' AS data_source,
     toUnixTimestamp64Milli(now64()) AS _version,

src/ingestion/connectors/git/gitlab/dbt/gitlab__pull_requests.sql

@@ -56,9 +56,9 @@ SELECT
     parseDateTimeBestEffortOrNull(mr.updated_at) AS updated_on,
     parseDateTimeBestEffortOrNull(COALESCE(mr.closed_at, mr.merged_at)) AS closed_on,
     COALESCE(mr.merge_commit_sha, '') AS merge_commit_hash,
-    toInt64(0) AS files_changed,
-    toInt64(0) AS lines_added,
-    toInt64(0) AS lines_removed,
+    toNullable(toInt64(0)) AS files_changed,
+    toNullable(toInt64(0)) AS lines_added,
+    toNullable(toInt64(0)) AS lines_removed,
     'insight_gitlab' AS data_source,
     toUnixTimestamp64Milli(now64()) AS _version,
     mr._airbyte_extracted_at