Use vgi. prefix for log metadata keys for consistency

All logging metadata keys now use the vgi. prefix to match the existing
vgi.status key pattern. This ensures protocol consistency.

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

Author: rustyconover

docs/protocol.md

@@ -194,6 +194,7 @@ Input and output data are standard Arrow RecordBatches matching the declared sch
 - `vgi.status`: One of `NEED_MORE_INPUT`, `HAVE_MORE_OUTPUT`, `FINISHED`
 - `vgi.log_level`: Log level if log message present (optional)
 - `vgi.log_message`: Log message text (optional)
+- `vgi.log_extra`: JSON-encoded additional context (optional)
 
 ---
 

vgi/client/client.py

@@ -180,18 +180,18 @@ def _handle_log_message(
     ) -> bool:
         """Handle a log message from the worker if present.
 
-        Detects log messages by checking for zero-row batches with log_level and
-        log_message metadata keys. When detected, logs the message using structlog
-        at the appropriate level. If the log level is "exception", raises a
-        ClientError with the message and traceback.
+        Detects log messages by checking for zero-row batches with vgi.log_level
+        and vgi.log_message metadata keys. When detected, logs the message using
+        structlog at the appropriate level. If the log level is "exception",
+        raises a ClientError with the message and traceback.
 
         Args:
             output_batch: The output batch from the worker. A log message is
                 detected when this has zero rows.
             output_metadata: Custom metadata dictionary from the batch. Must
-                contain b"log_level" and b"log_message" keys for the batch
-                to be treated as a log message. May optionally contain
-                b"log_extra" with JSON-encoded additional context.
+                contain b"vgi.log_level" and b"vgi.log_message" keys for the
+                batch to be treated as a log message. May optionally contain
+                b"vgi.log_extra" with JSON-encoded additional context.
 
         Returns:
             True if this was a log message and the caller should continue to
@@ -207,31 +207,31 @@ def _handle_log_message(
 
         if not (
             output_batch.num_rows == 0
-            and output_metadata.get(b"log_level") is not None
-            and output_metadata.get(b"log_message") is not None
+            and output_metadata.get(b"vgi.log_level") is not None
+            and output_metadata.get(b"vgi.log_message") is not None
         ):
             return False
 
         extra: dict[str, Any] = {}
-        if output_metadata.get(b"log_extra") is not None:
+        if output_metadata.get(b"vgi.log_extra") is not None:
             try:
-                extra = json.loads(output_metadata[b"log_extra"].decode())
+                extra = json.loads(output_metadata[b"vgi.log_extra"].decode())
             except json.JSONDecodeError as e:
                 log.error(
                     "failed_to_decode_log_extra",
                     error=str(e),
-                    raw=output_metadata[b"log_extra"],
+                    raw=output_metadata[b"vgi.log_extra"],
                 )
 
-        level_name = output_metadata[b"log_level"].decode().lower()
+        level_name = output_metadata[b"vgi.log_level"].decode().lower()
         worker_log._proxy_to_logger(
             level_name,
-            output_metadata[b"log_message"].decode(),
+            output_metadata[b"vgi.log_message"].decode(),
             **extra,
         )
 
         if level_name == "exception":
-            message = output_metadata[b"log_message"].decode()
+            message = output_metadata[b"vgi.log_message"].decode()
             traceback = extra.get("traceback", "")
             full_message = f"Worker Exception: {message}\n{traceback}"
             raise ClientError(full_message)
@@ -261,6 +261,11 @@ def _parse_bind_result(self, batch: pa.RecordBatch) -> _BindResult:
                 - raw_batch: The original RecordBatch for reference
 
         """
+        if batch.num_rows != 1:
+            raise ClientError(
+                "Expected single-row RecordBatch for bind result,"
+                f" got {batch.num_rows} rows"
+            )
         max_processes_array = batch.column(
             batch.schema.get_field_index("max_processes")
         )

vgi/log.py

@@ -203,14 +203,14 @@ def add_to_metadata(
 
         Returns:
             New dict containing original entries plus:
-            - log_level: The Level value (e.g., "INFO", "EXCEPTION")
-            - log_message: The human-readable message text
-            - log_extra: JSON string with {correlation_id, invocation_id,
+            - vgi.log_level: The Level value (e.g., "INFO", "EXCEPTION")
+            - vgi.log_message: The human-readable message text
+            - vgi.log_extra: JSON string with {correlation_id, invocation_id,
                 pid, ...extra kwargs}
 
         """
         result = dict(metadata) if metadata else {}
-        result["log_level"] = self.level.value
+        result["vgi.log_level"] = self.level.value
         log_data: dict[str, Any] = {
             "correlation_id": invocation.correlation_id,
             "invocation_id": invocation.invocation_id.hex()
@@ -220,8 +220,8 @@ def add_to_metadata(
         }
         if self.extra:
             log_data.update(self.extra)
-        result["log_message"] = self.message
-        result["log_extra"] = json.dumps(log_data)
+        result["vgi.log_message"] = self.message
+        result["vgi.log_extra"] = json.dumps(log_data)
         return result
 
     @classmethod