fix(sink): promote the physical column type when a contract widens

Symptom: batch 5 of billing_transactions died with
  _duckdb.ConversionException: Conversion Error: Type INT64 with value
  2216450393 can't be cast because the value is out of range for the
  destination type INT32 when casting from source column account_id
The classifier had already returned widening_safe for account_id, the BACKWARD
policy said EVOLVE, and contract v3 on disk said BIGINT. The load still failed.

Root cause: contract evolution and physical table evolution were two separate
things. LandingZone.evolve_table only issued ADD COLUMN for fields missing from
the table, so ddl.build_alter_type had no callers at all. The registry said
BIGINT while DuckDB still said INTEGER, which made the framework's central
promise ("widening is safe, we handle it") false at the storage layer.

Fix: evolve_table now reads the physical types out of information_schema, maps
them back into the logical lattice with duckdb_type_to_spec, and issues
ALTER COLUMN ... SET DATA TYPE for any field whose contract type is a strict
widening of what the table has. The promotion is gated on is_widening so this
path can never narrow a column and truncate landed history.

Author: sandeep0430

src/schema_guard/sink.py

@@ -9,9 +9,10 @@
 
 * Landing is idempotent. A batch that lands twice produces the same table, because rows
   are merged on the contract primary key rather than appended. Replay depends on this.
-* Evolution only adds. ``evolve_table`` issues ADD COLUMN, never DROP COLUMN. A field
-  removed from a contract keeps its physical column, so a report reading history under an
-  older contract version still resolves.
+* Evolution never destroys. ``evolve_table`` issues ADD COLUMN for new fields and
+  ALTER COLUMN for a widened type, and never DROP COLUMN. A field removed from a contract
+  keeps its physical column, so a report reading history under an older contract version
+  still resolves.
 """
 
 from __future__ import annotations
@@ -22,12 +23,57 @@
 import pandas as pd
 
 from .contracts import Contract
-from .ddl import build_add_column, build_create_table, build_quarantine_table
+from .ddl import (
+    build_add_column,
+    build_alter_type,
+    build_create_table,
+    build_quarantine_table,
+)
 from .logging_setup import get_logger
 from .transforms import AUDIT_COLUMNS
+from .types import (
+    BOOLEAN,
+    DATE,
+    DECIMAL,
+    FLOAT64,
+    INT32,
+    INT64,
+    STRING,
+    TIMESTAMP,
+    TypeSpec,
+    is_widening,
+)
 
 log = get_logger("sink")
 
+# DuckDB reports column types through information_schema as plain strings. Mapping them
+# back into the logical lattice is what lets evolve_table decide whether a contract type
+# is a widening of what is physically there.
+_DUCKDB_TO_LOGICAL = {
+    "TINYINT": INT32,
+    "SMALLINT": INT32,
+    "INTEGER": INT32,
+    "BIGINT": INT64,
+    "HUGEINT": INT64,
+    "FLOAT": FLOAT64,
+    "DOUBLE": FLOAT64,
+    "VARCHAR": STRING,
+    "BOOLEAN": BOOLEAN,
+    "DATE": DATE,
+    "TIMESTAMP": TIMESTAMP,
+}
+
+
+def duckdb_type_to_spec(data_type: str) -> TypeSpec | None:
+    """Parse a DuckDB information_schema data_type into a TypeSpec, or None if unknown."""
+    upper = data_type.strip().upper()
+    if upper.startswith("DECIMAL"):
+        inner = upper[upper.find("(") + 1 : upper.find(")")]
+        precision, _, scale = inner.partition(",")
+        return TypeSpec(DECIMAL, precision=int(precision), scale=int(scale or 0))
+    logical = _DUCKDB_TO_LOGICAL.get(upper)
+    return None if logical is None else TypeSpec(logical)
+
 
 class LandingZone:
     """Owns the DuckDB connection and every write into the landing schema."""
@@ -67,8 +113,16 @@ def columns(self, name: str) -> list[str]:
         ).fetchall()
         return [r[0] for r in rows]
 
+    def column_types(self, name: str) -> dict[str, str]:
+        rows = self.connection.execute(
+            "SELECT column_name, data_type FROM information_schema.columns "
+            "WHERE table_schema = ? AND table_name = ? ORDER BY ordinal_position",
+            [self.schema, name],
+        ).fetchall()
+        return {r[0]: r[1] for r in rows}
+
     def ensure_table(self, contract: Contract) -> None:
-        """Create the landing table if absent, otherwise add any new contract columns."""
+        """Create the landing table if absent, otherwise bring it up to the contract."""
         if not self.table_exists(contract.dataset):
             self.connection.execute(build_create_table(contract, self.schema))
             log.info(
@@ -78,21 +132,43 @@ def ensure_table(self, contract: Contract) -> None:
             return
         self.evolve_table(contract)
 
-    def evolve_table(self, contract: Contract) -> list[str]:
-        """Add columns present in the contract but not yet in the table. Returns names."""
-        existing = set(self.columns(contract.dataset))
+    def evolve_table(self, contract: Contract) -> dict[str, list[str]]:
+        """Bring the physical table up to the contract: add columns, promote types.
+
+        Both halves are needed. Adding the new columns alone was the first version of
+        this method and it was wrong: when the classifier promoted account_id from
+        INTEGER to BIGINT, the contract on disk said BIGINT while DuckDB still said
+        INTEGER, and the next insert failed with an out of range conversion. A "widening
+        is safe" guarantee that stops at the registry is not a guarantee.
+
+        Promotion is gated on ``is_widening``, so this can never quietly narrow a column
+        and truncate landed history. A narrowing must go through a contract amendment
+        that a person approved.
+        """
+        physical = self.column_types(contract.dataset)
         added: list[str] = []
+        promoted: list[str] = []
         for f in contract.fields:
-            if f.name in existing:
+            if f.name not in physical:
+                self.connection.execute(build_add_column(contract, self.schema, f.name))
+                added.append(f.name)
                 continue
-            self.connection.execute(build_add_column(contract, self.schema, f.name))
-            added.append(f.name)
-        if added:
+            current = duckdb_type_to_spec(physical[f.name])
+            if current is not None and is_widening(current, f.type):
+                self.connection.execute(build_alter_type(contract, self.schema, f.name))
+                promoted.append(f"{f.name}:{current.sql_type()}->{f.type.sql_type()}")
+        if added or promoted:
             log.info(
                 "evolved landing table",
-                extra={"context": {"table": contract.dataset, "added_columns": added}},
+                extra={
+                    "context": {
+                        "table": contract.dataset,
+                        "added_columns": added,
+                        "promoted_columns": promoted,
+                    }
+                },
             )
-        return added
+        return {"added": added, "promoted": promoted}
 
     # ------------------------------------------------------------------ writes