@@ -1,6 +1,6 @@
 [project]
 name = "vgi-python"
-version = "0.38.1"
+version = "0.39.0"
 description = "Vector Gateway Interface - Connect DuckDB to external programs via Apache Arrow"
 readme = "README.md"
 keywords = [
 
@@ -55,6 +55,12 @@
     ("vgi.codegen.rust_request_builders", "vgi-protocol/src/generated/request_params.rs", "vgi-rust"),
     ("vgi.codegen.ts_schemas", "src/generated/vgi-protocol-schemas.ts", "vgi-typescript"),
     ("vgi.codegen.ts_client", "src/generated/vgi-client.ts", "vgi-typescript"),
+    ("vgi.codegen.csharp_types", "src/QueryFarm.Vgi/Protocol/Generated/VgiProtocolTypes.g.cs", "vgi-csharp"),
+    (
+        "vgi.codegen.csharp_schemas",
+        "test/QueryFarm.Vgi.Tests/Generated/VgiProtocolSchemas.g.cs",
+        "vgi-csharp",
+    ),
     (
         "vgi.codegen.java_schemas",
         "vgi/src/test/java/farm/query/vgi/generated/VgiProtocolSchemas.java",
 
@@ -0,0 +1,111 @@
+# Copyright 2025, 2026 Query Farm LLC - https://query.farm
+
+"""Drift + behavior tests for `vgi.codegen.csharp_types` and `vgi.codegen.csharp_schemas`.
+
+vgi-csharp checks in two generated files: the protocol's records and enums as C#
+types (its wire schemas are derived from them by reflection), and the protocol's
+schemas as C# Arrow schemas, which its test project compares that derivation
+against. Both must match what the generators emit right now.
+
+The drift tests compare text, not parsed schemas: the types file is more than
+schemas (CLR types, attributes, initializers), and any byte of it changing is a
+change to what vgi-csharp compiles.
+
+Skipped when the `vgi-csharp` checkout is not next to vgi-python;
+``VGI_CSHARP_ROOT`` overrides the location.
+"""
+
+from __future__ import annotations
+
+import io
+import os
+from pathlib import Path
+
+import pytest
+
+from vgi.codegen import csharp_schemas, csharp_types
+
+_TYPES = Path("src/QueryFarm.Vgi/Protocol/Generated/VgiProtocolTypes.g.cs")
+_SCHEMAS = Path("test/QueryFarm.Vgi.Tests/Generated/VgiProtocolSchemas.g.cs")
+
+_REGEN_HINT = (
+    "To regenerate, run:\n"
+    "  uv run --project ~/Development/vgi-python python scripts/regen_generated.py\n"
+    "\n"
+    "Do not redirect a generator into its destination with `>`: the shell\n"
+    "truncates the file before the generator runs, so any failure destroys\n"
+    "the checked-in artifact. The script renders to memory first."
+)
+
+
+def _vgi_csharp_root() -> Path:
+    override = os.environ.get("VGI_CSHARP_ROOT")
+    if override:
+        return Path(override)
+    return Path(__file__).resolve().parents[2] / "vgi-csharp"
+
+
+def _render(module: object) -> str:
+    buf = io.StringIO()
+    module.emit(buf)  # type: ignore[attr-defined]
+    return buf.getvalue()
+
+
+@pytest.mark.parametrize("module", [csharp_types, csharp_schemas], ids=["types", "schemas"])
+def test_generator_is_deterministic(module: object) -> None:
+    """Calling emit() twice produces byte-identical output."""
+    assert _render(module) == _render(module)
+
+
+@pytest.mark.parametrize(
+    ("module", "relative"),
+    [(csharp_types, _TYPES), (csharp_schemas, _SCHEMAS)],
+    ids=["types", "schemas"],
+)
+def test_checked_in_file_matches_generator(module: object, relative: Path) -> None:
+    """Drift check: the checked-in C# matches what the generator produces."""
+    path = _vgi_csharp_root() / relative
+    if not path.exists():
+        pytest.skip(f"{path} not found; set VGI_CSHARP_ROOT or check out vgi-csharp")
+    assert path.read_text() == _render(module), f"{path} is stale.\n{_REGEN_HINT}"
+
+
+def test_every_csharp_rename_is_used() -> None:
+    """A rename whose Python type no longer exists is dead, and would hide the next rename it needs."""
+    model = csharp_types.build_model()
+    emitted = set(model.records) | set(model.enums)
+    reachable_python_names = {py for py, cs in csharp_types.CSHARP_NAMES.items() if cs in emitted}
+    stale = set(csharp_types.CSHARP_NAMES) - reachable_python_names
+    assert not stale, f"CSHARP_NAMES entries map to nothing emitted: {sorted(stale)}"
+
+
+def test_merged_items_responses_share_one_type() -> None:
+    """The catalog items responses collapse into one C# type, built from all of them."""
+    record = csharp_types.build_model().records["ItemsResponse"]
+    methods = {o.removeprefix("method '").split("'")[0] for o in record.origins}
+    assert {"catalog_catalogs", "catalog_schemas", "catalog_table_get", "catalog_index_get"} <= methods
+    assert [p.wire_name for p in record.props] == ["items"]
+
+
+def test_large_binary_list_carries_large_width() -> None:
+    """`list<large_binary>` is declared on the property; without it the derivation says `list<binary>`."""
+    props = {p.wire_name: p for p in csharp_types.build_model().records["InitRequest"].props}
+    assert "[global::QueryFarm.VgiRpc.Reflection.LargeWidth]" in props["join_keys"].attrs
+    assert props["join_keys"].clr_type == "global::System.Collections.Generic.List<byte[]>?"
+    assert "[global::QueryFarm.VgiRpc.Reflection.LargeWidth]" in props["pushdown_filters"].attrs
+
+
+def test_appended_nullable_flag_is_wire_optional() -> None:
+    """A nullable column typed as a plain value with a default is an appended column."""
+    props = {p.wire_name: p for p in csharp_types.build_model().records["AttachOptionSpec"].props}
+    assert props["required"].clr_type == "bool"
+    assert "[global::QueryFarm.Vgi.Internal.WireOptional]" in props["required"].attrs
+
+
+def test_enum_members_round_trip_to_their_wire_names() -> None:
+    """C# derives an enum member's wire name as UPPER_SNAKE of its name; every member must survive that."""
+    for e in csharp_types.build_model().enums.values():
+        for m in e.members:
+            derived = csharp_types._to_snake_case(m.name).upper()
+            rendered = csharp_types._render_enum(e)
+            assert derived == m.wire_name or f'RpcName("{m.wire_name}")' in rendered, (e.name, m.name)
@@ -0,0 +1,203 @@
+# Copyright 2025, 2026 Query Farm LLC - https://query.farm
+
+r"""Emit the VGI protocol's Arrow schemas as C#, for vgi-csharp's test project.
+
+The companion of ``vgi.codegen.csharp_types``. That module emits the C# record
+types vgi-csharp derives its wire schemas from; this one emits the schemas the
+derivation must produce, taken from the protocol rather than from C#, so a test
+can compare the two field for field — the same "second description" vgi-java's
+``WireRecordSchemaConformanceTest`` relies on.
+
+Generating the types makes them agree with the protocol only as far as the
+generator's CLR mapping agrees with vgi-rpc-csharp's ``SchemaDerivation``; this
+is what checks that. It also covers what is still written by hand: the flat
+parameters of ``IVgiService``'s methods, and which record each method returns.
+
+Three maps, all keyed by protocol names:
+
+- ``Records`` — every record ``csharp_types`` emits, by C# type name.
+- ``MethodResults`` — each unary method's result record schema (the record
+  inside the ``result`` binary), by wire method name; methods returning nothing
+  or raw IPC are absent.
+- ``MethodParams`` — each method's params batch schema, by wire method name.
+
+.. code-block:: bash
+
+   uv run --project ~/Development/vgi-python python scripts/regen_generated.py
+"""
+
+from __future__ import annotations
+
+import argparse
+import io
+import sys
+from typing import TYPE_CHECKING, Any
+
+import pyarrow as pa
+from vgi_rpc.rpc._types import MethodType, rpc_methods  # type: ignore[attr-defined]
+
+from vgi.codegen._common import GeneratorError, _resolve_inner_schema, provenance_comment
+from vgi.codegen.csharp_types import build_model
+from vgi.protocol import VgiProtocol
+
+if TYPE_CHECKING:
+    from typing import TextIO
+
+
+GENERATOR_VERSION = "1"
+
+DEFAULT_NAMESPACE = "QueryFarm.Vgi.Tests.Generated"
+
+_SCALARS: list[tuple[pa.DataType, str]] = [
+    (pa.bool_(), "BooleanType.Default"),
+    (pa.int8(), "Int8Type.Default"),
+    (pa.int16(), "Int16Type.Default"),
+    (pa.int32(), "Int32Type.Default"),
+    (pa.int64(), "Int64Type.Default"),
+    (pa.uint8(), "UInt8Type.Default"),
+    (pa.uint16(), "UInt16Type.Default"),
+    (pa.uint32(), "UInt32Type.Default"),
+    (pa.uint64(), "UInt64Type.Default"),
+    (pa.float32(), "FloatType.Default"),
+    (pa.float64(), "DoubleType.Default"),
+    (pa.string(), "StringType.Default"),
+    (pa.large_string(), "LargeStringType.Default"),
+    (pa.binary(), "BinaryType.Default"),
+    (pa.large_binary(), "LargeBinaryType.Default"),
+]
+
+_UNITS = {"s": "Second", "ms": "Millisecond", "us": "Microsecond", "ns": "Nanosecond"}
+
+
+def _type(dtype: pa.DataType, origin: str, indent: str) -> str:
+    for proto, expr in _SCALARS:
+        if dtype.equals(proto):
+            return expr
+    if pa.types.is_dictionary(dtype):
+        index = _type(dtype.index_type, f"{origin}[dict index]", indent)
+        value = _type(dtype.value_type, f"{origin}[dict value]", indent)
+        return f"new DictionaryType({index}, {value}, ordered: {str(dtype.ordered).lower()})"
+    if pa.types.is_fixed_size_binary(dtype):
+        return f"new FixedSizeBinaryType({dtype.byte_width})"
+    if pa.types.is_timestamp(dtype):
+        tz = "(string?)null" if dtype.tz is None else f'"{dtype.tz}"'
+        return f"new TimestampType(TimeUnit.{_UNITS[dtype.unit]}, {tz})"
+    if pa.types.is_list(dtype):
+        return f"new ListType({_field(dtype.value_field, origin, indent)})"
+    if pa.types.is_map(dtype):
+        key = _field(dtype.key_field.with_name("key"), origin, indent)
+        value = _field(dtype.item_field.with_name("value"), origin, indent)
+        return f"new MapType({key}, {value})"
+    if pa.types.is_struct(dtype):
+        inner = indent + "    "
+        children = ",\n".join(inner + _field(dtype.field(i), origin, inner) for i in range(dtype.num_fields))
+        return f"new StructType(\n{indent}[\n{children},\n{indent}])"
+    raise GeneratorError(f"vgi.codegen.csharp_schemas: unsupported Arrow type {dtype} at {origin}.")
+
+
+def _field(f: pa.Field[Any], origin: str, indent: str) -> str:
+    return f'F("{f.name}", {_type(f.type, f"{origin}.{f.name}", indent)}, {str(f.nullable).lower()})'
+
+
+def _schema(schema: pa.Schema, origin: str) -> str:
+    if len(schema) == 0:
+        return "S()"
+    indent = "            "
+    fields = ",\n".join(indent + _field(f, origin, indent) for f in schema)
+    return f"S(\n{fields})"
+
+
+def _map(name: str, doc: str, entries: list[tuple[str, pa.Schema]]) -> str:
+    lines = [
+        f"    /// <summary>{doc}</summary>",
+        f"    public static IReadOnlyDictionary<string, Schema> {name} {{ get; }} = new Dictionary<string, Schema>",
+        "    {",
+    ]
+    for key, schema in entries:
+        lines.append(f'        ["{key}"] = {_schema(schema, key)},')
+    lines.append("    };")
+    return "\n".join(lines) + "\n"
+
+
+def emit(out: TextIO, *, namespace: str = DEFAULT_NAMESPACE) -> None:
+    """Emit the generated C# schema class to *out*."""
+    model = build_model()
+    records = [(name, model.records[name].schema) for name in sorted(model.records)]
+
+    methods = rpc_methods(VgiProtocol)
+    results: list[tuple[str, pa.Schema]] = []
+    params: list[tuple[str, pa.Schema]] = []
+    for method_name in sorted(methods):
+        info = methods[method_name]
+        params.append((method_name, info.params_schema))
+        if info.method_type == MethodType.UNARY and info.has_return:
+            inner = _resolve_inner_schema(info.result_type, method_name)
+            if inner is not None:
+                results.append((method_name, inner))
+
+    body = io.StringIO()
+    body.write("// Copyright 2025, 2026 Query Farm LLC - https://query.farm\n")
+    body.write("// <auto-generated/>\n")
+    body.write("\n")
+    body.write("#nullable enable\n")
+    body.write("\n")
+    body.write("using Apache.Arrow;\n")
+    body.write("using Apache.Arrow.Types;\n")
+    body.write("\n")
+    body.write(f"namespace {namespace};\n")
+    body.write("\n")
+    body.write(
+        "/// <summary>\n"
+        "/// The VGI protocol's Arrow schemas, generated from the protocol itself (vgi-python) rather\n"
+        "/// than derived from this port's CLR types, so a test can compare the two. Dictionary ids and\n"
+        "/// metadata are not part of the comparison.\n"
+        "/// </summary>\n"
+    )
+    body.write("public static class VgiProtocolSchemas\n{\n")
+    body.write(
+        "    private static Schema S(params Field[] fields) => new(fields, metadata: null);\n\n"
+        "    private static Field F(string name, IArrowType type, bool nullable) => new(name, type, nullable);\n\n"
+    )
+    body.write(_map("Records", "Every protocol record's schema, by its C# type name.", records))
+    body.write("\n")
+    body.write(
+        _map(
+            "MethodResults",
+            "Each unary method's result record schema, by wire method name. Methods that return "
+            "nothing or raw IPC are absent.",
+            results,
+        )
+    )
+    body.write("\n")
+    body.write(_map("MethodParams", "Each method's params batch schema, by wire method name.", params))
+    body.write("}\n")
+
+    out.write(
+        provenance_comment(
+            generator_module="vgi.codegen.csharp_schemas",
+            generator_command="python -m vgi.codegen.csharp_schemas",
+            generator_version=GENERATOR_VERSION,
+            regen_command_lines=[
+                "uv run --project ~/Development/vgi-python python scripts/regen_generated.py",
+            ],
+            body=body.getvalue(),
+        )
+    )
+    out.write("\n")
+    out.write(body.getvalue())
+
+
+def main() -> None:
+    """Console-script entrypoint — write the C# schema class to stdout."""
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--namespace", default=DEFAULT_NAMESPACE, help="C# namespace of the emitted class")
+    args = parser.parse_args()
+    try:
+        emit(sys.stdout, namespace=args.namespace)
+    except GeneratorError as e:
+        print(f"\nerror: {e}\n", file=sys.stderr)
+        sys.exit(2)
+
+
+if __name__ == "__main__":
+    main()