Fix review issues from secrets system overhaul

- Fix scalar 0-column/0-row validation: validate output against synthetic
  batch instead of skipping validation when inject_row=True
- Remove dead _validate_required_secrets no-op method and call site
- Deduplicate secret batch parsing: add SecretsAccessor.to_dict(), replace
  all 6 _batch_to_secret_dict() call sites, delete the old function
- Use SecretLookupEntry (now extends ArrowSerializableDataclass) instead of
  untyped dicts for required_secrets throughout metadata and catalog
- Eliminate CatalogSecretRequirement which duplicated SecretLookupEntry
- Remove backward-compat isinstance(entry, str) branch in resolve_metadata
- Add SecretLookupEntry to __all__ in arguments.py
- Add type-ignore explanation comments for PyArrow field metadata limitation
- Fix pre-existing mypy errors (boto3/aiohttp import-not-found suppressions)
- Add tests: SecretsAccessor (get, scoped, required, to_dict),
  BindResponse.secret_scope_request round-trip, TState validation,
  SecretTypeSpec serialization round-trip

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: rustyconover

CLAUDE.md

@@ -37,6 +37,49 @@ uv run ruff check --fix . && uv run ruff format .
 
 When making changes, we don't need to worry about backward compatibility, make the changes and change the import references.
 
+## Integration Testing
+
+Integration tests live in the `vgi` C++ repo (sibling directory) at `test/sql/integration/`.
+They run DuckDB's sqllogictest framework against a real VGI worker. Run from the `vgi` directory.
+
+**Subprocess transport** (worker spawned as a child process):
+```bash
+cd ~/Development/vgi
+VGI_TEST_WORKER="uv run --project ~/Development/vgi-python vgi-example-worker" \
+    ./build/release/test/unittest "test/sql/integration/*"
+```
+
+**HTTP transport** (worker as an HTTP server):
+```bash
+cd ~/Development/vgi
+./test/run_http_integration.sh "test/sql/integration/*"
+```
+
+The HTTP script (`test/run_http_integration.sh`) starts the HTTP server, waits for it to be
+ready, runs the tests, and cleans up. Always use this script instead of manually managing
+the server. Server logs are at `/tmp/vgi-http-test-server.log`.
+
+**Filter by test subset:**
+```bash
+./test/run_http_integration.sh "test/sql/integration/secret/*"
+./test/run_http_integration.sh "test/sql/integration/table/countdown*"
+```
+
+**Running ad-hoc SQL against DuckDB CLI:**
+Use `-f` to supply SQL files to DuckDB, not stdin redirection. Do not redirect stderr to stdout.
+```bash
+# Correct:
+~/Development/vgi/build/debug/duckdb -f /tmp/my_test.sql
+
+# Wrong:
+~/Development/vgi/build/debug/duckdb < /tmp/my_test.sql 2>&1
+```
+
+**Known HTTP limitations** (4 tests fail, not regressions):
+- `logging_generator.test` — inline log streaming not supported over HTTP
+- `partitioned_sequence.test` — partition-local state not preserved across HTTP exchanges
+- `buffer_input/sizes.test`, `buffer_input/scale.test_slow` — input buffering semantics differ
+
 ## Project Overview
 
 VGI (Vector Gateway Interface) provides an Apache Arrow-based protocol for connecting DuckDB to external programs. It enables user-defined functions to run in separate processes, communicating via stdin/stdout using Arrow IPC streaming.

tests/catalog/test_serialization.py

@@ -766,7 +766,7 @@ class TestArrowSchemaCorrectness:
     def test_catalog_attach_result_schema(self) -> None:
         """Verify CatalogAttachResult Arrow schema."""
         schema = CatalogAttachResult.ARROW_SCHEMA
-        assert len(schema) == 8
+        assert len(schema) == 9
         assert schema.field("attach_id").type == pa.binary()
         assert schema.field("supports_transactions").type == pa.bool_()
         assert schema.field("supports_time_travel").type == pa.bool_()
@@ -775,6 +775,7 @@ def test_catalog_attach_result_schema(self) -> None:
         assert schema.field("attach_id_required").type == pa.bool_()
         assert schema.field("default_schema").type == pa.string()
         assert schema.field("settings").type == pa.list_(pa.binary())
+        assert schema.field("secret_types").type == pa.list_(pa.binary())
 
     def test_schema_info_schema(self) -> None:
         """Verify SchemaInfo Arrow schema."""
@@ -816,3 +817,33 @@ def test_scan_function_result_schema(self) -> None:
         assert schema.field("function_name").type == pa.string()
         assert schema.field("arguments").type == pa.binary()
         assert schema.field("required_extensions").type == pa.list_(pa.string())
+
+
+class TestSecretTypeSpecSerialization:
+    """Test SecretTypeSpec serialization round-trip."""
+
+    def test_round_trip(self) -> None:
+        """Test basic serialization and deserialization."""
+        from vgi.catalog.secret_type import SecretTypeSpec
+
+        original = SecretTypeSpec(
+            name="vgi_example",
+            description="Example VGI secret for testing",
+            schema=pa.schema(
+                [
+                    pa.field("secret_string", pa.string(), metadata={"redact": "true"}),
+                    pa.field("api_key", pa.string(), metadata={"redact": "true"}),
+                    pa.field("port", pa.int32()),
+                ]  # type: ignore[arg-type]  # PyArrow field metadata typing limitation
+            ),
+        )
+        serialized = original.serialize()
+        batch, _ = deserialize_record_batch(serialized)
+        restored = SecretTypeSpec.deserialize(batch)
+
+        assert restored.name == original.name
+        assert restored.description == original.description
+        assert restored.schema.names == original.schema.names
+        # Verify field metadata survives round-trip
+        assert restored.schema.field("secret_string").metadata == {b"redact": b"true"}
+        assert restored.schema.field("port").metadata is None

tests/scalar/test_return_secret_value_function.py

@@ -25,7 +25,7 @@ def test_return_secret_value_basic(self, example_worker: str) -> None:
                 client.scalar_function(
                     function_name="return_secret_value",
                     input=iter([batch]),
-                    secrets={"vgi_example_secret": secret_value},
+                    secrets={"vgi_example": secret_value},
                 )
             )
 

tests/test_http_s3_offload_input.py

@@ -113,7 +113,7 @@ def test_http_input_upload_url_then_external_location_scalar_exchange(compressio
     pytest.importorskip("vgi_rpc.s3")
 
     if compression == "zstd":
-        import aiohttp.http_parser as http_parser
+        import aiohttp.http_parser as http_parser  # type: ignore[import-not-found]
 
         if not bool(getattr(http_parser, "HAS_ZSTD", False)):
             pytest.skip("zstd input test requires aiohttp zstd decode support (backports.zstd)")

tests/test_http_s3_offload_output.py

@@ -115,7 +115,7 @@ def test_http_output_offload_random_bytes_large_response(compression: str, expec
     pytest.importorskip("vgi_rpc.http")
     pytest.importorskip("vgi_rpc.s3")
     if compression == "zstd":
-        import aiohttp.http_parser as http_parser
+        import aiohttp.http_parser as http_parser  # type: ignore[import-not-found]
 
         if not bool(getattr(http_parser, "HAS_ZSTD", False)):
             pytest.skip("zstd externalization test requires aiohttp zstd decode support (backports.zstd)")