Add attach_id_required field and update CLI to use --attach-id/--catalog options

- Add attach_id_required field to CatalogAttachResult for stateful/stateless
  catalog indication
- Update ReadOnlyCatalogInterface to return attach_id_required=False
- Replace positional attach_id arguments with --attach-id and --catalog options
  across all catalog CLI commands (schema, table, view, transaction, version)
- Add get_attach_id_from_options() helper for auto-attach workflow
- Show warning when using --catalog with stateful catalogs
- Update and add CLI tests for new option syntax and validation

🤖 Generated with [Claude Code](https://claude.com/claude-code)

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

Author: rustyconover

tests/catalog/test_serialization.py

@@ -346,12 +346,13 @@ class TestArrowSchemaCorrectness:
     def test_catalog_attach_result_schema(self) -> None:
         """Verify CatalogAttachResult Arrow schema."""
         schema = CatalogAttachResult.ARROW_SCHEMA
-        assert len(schema) == 5
+        assert len(schema) == 6
         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_()
         assert schema.field("catalog_version_frozen").type == pa.bool_()
         assert schema.field("catalog_version").type == pa.int64()
+        assert schema.field("attach_id_required").type == pa.bool_()
 
     def test_schema_info_schema(self) -> None:
         """Verify SchemaInfo Arrow schema."""

tests/client/test_cli_catalog_functions.py

@@ -54,6 +54,8 @@ def test_catalog_attach_returns_attach_id(self, example_worker: str) -> None:
         assert len(attach_result["attach_id"]) > 0
         assert attach_result["supports_transactions"] is False
         assert attach_result["catalog_version_frozen"] is True
+        # ReadOnlyCatalogInterface returns attach_id_required=False
+        assert attach_result["attach_id_required"] is False
 
 
 class TestCLISchemaContents:
@@ -72,15 +74,47 @@ def test_schema_contents_lists_functions(self, example_worker: str) -> None:
         attach_data = json.loads(attach_result.output)
         attach_id = attach_data["attach_id"]
 
-        # List schema contents (now nested under catalog)
+        # List schema contents using --attach-id option
         contents_result = runner.invoke(
             cli,
             [
                 "catalog",
                 "schema",
                 "contents",
+                "main",
+                "--attach-id",
                 attach_id,
+                "--server",
+                example_worker,
+            ],
+        )
+        assert contents_result.exit_code == 0
+
+        # Parse all output lines as JSON
+        lines = contents_result.output.strip().split("\n")
+        items = [json.loads(line) for line in lines if line.strip()]
+
+        # Should have items
+        assert len(items) > 0
+
+        # All items should be functions
+        for item in items:
+            assert item["type"] == "function"
+
+    def test_schema_contents_with_catalog_option(self, example_worker: str) -> None:
+        """Schema contents works with --catalog option for auto-attach."""
+        runner = CliRunner()
+
+        # List schema contents using --catalog option (auto-attach)
+        contents_result = runner.invoke(
+            cli,
+            [
+                "catalog",
+                "schema",
+                "contents",
                 "main",
+                "--catalog",
+                "example",
                 "--server",
                 example_worker,
             ],
@@ -110,15 +144,16 @@ def test_all_example_functions_present(self, example_worker: str) -> None:
         assert attach_result.exit_code == 0
         attach_id = json.loads(attach_result.output)["attach_id"]
 
-        # Get schema contents (now nested under catalog)
+        # Get schema contents using --attach-id option
         contents_result = runner.invoke(
             cli,
             [
                 "catalog",
                 "schema",
                 "contents",
-                attach_id,
                 "main",
+                "--attach-id",
+                attach_id,
                 "--server",
                 example_worker,
             ],
@@ -158,8 +193,9 @@ def test_function_info_has_correct_structure(self, example_worker: str) -> None:
                 "catalog",
                 "schema",
                 "contents",
-                attach_id,
                 "main",
+                "--attach-id",
+                attach_id,
                 "--server",
                 example_worker,
             ],
@@ -195,8 +231,9 @@ def test_scalar_functions_have_correct_type(self, example_worker: str) -> None:
                 "catalog",
                 "schema",
                 "contents",
-                attach_id,
                 "main",
+                "--attach-id",
+                attach_id,
                 "--server",
                 example_worker,
             ],
@@ -229,8 +266,9 @@ def test_table_functions_have_correct_type(self, example_worker: str) -> None:
                 "catalog",
                 "schema",
                 "contents",
-                attach_id,
                 "main",
+                "--attach-id",
+                attach_id,
                 "--server",
                 example_worker,
             ],
@@ -262,14 +300,96 @@ def test_schema_list_shows_main(self, example_worker: str) -> None:
         )
         attach_id = json.loads(attach_result.output)["attach_id"]
 
-        # List schemas (now nested under catalog)
+        # List schemas using --attach-id option
+        list_result = runner.invoke(
+            cli,
+            [
+                "catalog",
+                "schema",
+                "list",
+                "--attach-id",
+                attach_id,
+                "--server",
+                example_worker,
+            ],
+        )
+        assert list_result.exit_code == 0
+
+        # Parse schema info
+        schema_info = json.loads(list_result.output)
+        assert schema_info["name"] == "main"
+        assert schema_info["is_default"] is True
+
+    def test_schema_list_with_catalog_option(self, example_worker: str) -> None:
+        """Schema list works with --catalog option."""
+        runner = CliRunner()
+
+        # List schemas using --catalog option (auto-attach)
         list_result = runner.invoke(
             cli,
-            ["catalog", "schema", "list", attach_id, "--server", example_worker],
+            [
+                "catalog",
+                "schema",
+                "list",
+                "--catalog",
+                "example",
+                "--server",
+                example_worker,
+            ],
         )
         assert list_result.exit_code == 0
 
         # Parse schema info
         schema_info = json.loads(list_result.output)
         assert schema_info["name"] == "main"
         assert schema_info["is_default"] is True
+
+
+class TestCLIAttachIdCatalogOptions:
+    """Tests for --attach-id and --catalog option validation."""
+
+    def test_mutual_exclusivity(self, example_worker: str) -> None:
+        """Error when both --attach-id and --catalog are specified."""
+        runner = CliRunner()
+
+        # First get an attach_id
+        attach_result = runner.invoke(
+            cli,
+            ["catalog", "attach", "example", "--server", example_worker],
+        )
+        attach_id = json.loads(attach_result.output)["attach_id"]
+
+        # Try to use both options
+        result = runner.invoke(
+            cli,
+            [
+                "catalog",
+                "schema",
+                "list",
+                "--attach-id",
+                attach_id,
+                "--catalog",
+                "example",
+                "--server",
+                example_worker,
+            ],
+        )
+        assert result.exit_code != 0
+        assert "Cannot specify both" in result.output
+
+    def test_requires_attach_id_or_catalog(self, example_worker: str) -> None:
+        """Error when neither --attach-id nor --catalog is specified."""
+        runner = CliRunner()
+
+        result = runner.invoke(
+            cli,
+            [
+                "catalog",
+                "schema",
+                "list",
+                "--server",
+                example_worker,
+            ],
+        )
+        assert result.exit_code != 0
+        assert "Must specify either" in result.output

vgi/catalog/catalog_interface.py

@@ -40,6 +40,10 @@ class CatalogAttachResult:
     # The initial catalog version, it increments when schemas, tables
     # or other objects change.
     catalog_version: int
+    # Indicate if the attach_id must be persisted across commands.
+    # True: Catalog is stateful; attach_id represents a session
+    # False: Catalog is stateless; CLI can auto-attach on each command
+    attach_id_required: bool = True
 
     ARROW_SCHEMA: ClassVar[pa.Schema] = pa.schema(
         [
@@ -48,6 +52,7 @@ class CatalogAttachResult:
             pa.field("supports_time_travel", pa.bool_(), nullable=False),
             pa.field("catalog_version_frozen", pa.bool_(), nullable=False),
             pa.field("catalog_version", pa.int64(), nullable=False),
+            pa.field("attach_id_required", pa.bool_(), nullable=False),
         ]  # type: ignore[arg-type]
     )
 
@@ -61,6 +66,7 @@ def serialize(self) -> bytes:
                     "supports_time_travel": self.supports_time_travel,
                     "catalog_version_frozen": self.catalog_version_frozen,
                     "catalog_version": self.catalog_version,
+                    "attach_id_required": self.attach_id_required,
                 }
             ],
             schema=self.ARROW_SCHEMA,
@@ -79,6 +85,7 @@ def deserialize(cls, batch: pa.RecordBatch) -> Self:
                 "supports_time_travel",
                 "catalog_version_frozen",
                 "catalog_version",
+                "attach_id_required",
             ],
         )
         return cls(
@@ -87,6 +94,7 @@ def deserialize(cls, batch: pa.RecordBatch) -> Self:
             supports_time_travel=row["supports_time_travel"],
             catalog_version_frozen=row["catalog_version_frozen"],
             catalog_version=row["catalog_version"],
+            attach_id_required=row["attach_id_required"],
         )
 
 
@@ -959,6 +967,7 @@ def catalog_attach(
             supports_time_travel=False,
             catalog_version_frozen=True,
             catalog_version=1,
+            attach_id_required=False,
         )
 
     def schema_get(

vgi/client/cli_catalog.py

@@ -8,7 +8,7 @@
 ├── detach <attach_id>      # Detach from a catalog
 ├── create <name>           # Create a new catalog
 ├── drop <name>             # Drop a catalog
-├── version <attach_id>     # Get catalog version
+├── version                 # Get catalog version (--attach-id or --catalog)
 ├── schema                  # Schema operations
 │   ├── list/get/create/drop/contents
 ├── table                   # Table operations
@@ -29,7 +29,9 @@
 from vgi.client.cli_table import table
 from vgi.client.cli_transaction import transaction
 from vgi.client.cli_utils import (
+    bytes_to_hex,
     catalog_attach_result_to_dict,
+    get_attach_id_from_options,
     hex_to_attach_id,
     hex_to_transaction_id,
     output_json,
@@ -124,23 +126,37 @@ def catalog_drop(name: str, server: str) -> None:
 
 
 @catalog.command("version")
-@click.argument("attach_id")
+@click.option("--attach-id", help="Hex-encoded attach ID")
+@click.option("--catalog", "catalog_name", help="Catalog name for auto-attach")
+@click.option("--attach-options", default="{}", help="Attach options as JSON")
 @click.option("--server", required=True, help="VGI worker command")
 @click.option("--transaction-id", help="Transaction ID (hex) for transactional read")
-def catalog_version(attach_id: str, server: str, transaction_id: str | None) -> None:
-    """Get the current catalog version.
-
-    ATTACH_ID is the hex-encoded attach ID from catalog attach.
-
-    """
+def catalog_version(
+    attach_id: str | None,
+    catalog_name: str | None,
+    attach_options: str,
+    server: str,
+    transaction_id: str | None,
+) -> None:
+    """Get the current catalog version."""
     client = Client(server)
+    opts = parse_json_option(attach_options, "--attach-options")
+    resolved_attach_id, is_stateful = get_attach_id_from_options(
+        client, attach_id, catalog_name, opts
+    )
+    if is_stateful and catalog_name:
+        click.echo(
+            "Warning: Using --catalog with a stateful catalog. "
+            "Consider using --attach-id for session persistence.",
+            err=True,
+        )
     version = client.catalog_version(
-        attach_id=hex_to_attach_id(attach_id),
+        attach_id=resolved_attach_id,
         transaction_id=(
             hex_to_transaction_id(transaction_id) if transaction_id else None
         ),
     )
-    output_json({"version": version, "attach_id": attach_id})
+    output_json({"version": version, "attach_id": bytes_to_hex(resolved_attach_id)})
 
 
 # Add nested subcommand groups