feat(catalog): worker-settable per-schema tags

The vgi-lint metadata linter expects schema-level tags
(vgi.description_llm / vgi.description_md, surfaced via
duckdb_schemas().tags), but the worker SDK had no way to set them:
catalog_schemas hard-coded SchemaInfo.tags to Map.of().

(The catalog source_url half of the Rust fix was already wired in Java:
Worker.sourceUrl() flows through CatalogInfoSerializer, which emits the
source_url field rather than null.)

- Worker: add schemaTags(schema, Map) builder + schemaTags() accessor,
  paralleling schemaComment/schemaComments.
- VgiServiceImpl: add tags to the internal SchemaDesc, populate it in
  workerSchemas() from Worker.schemaTags(), and surface it via
  SchemaInfo.tags in catalog_schemas / catalog_schema_get (was Map.of()).
- Add SchemaInfoTagsTest covering tag flow-through, merge, and the
  empty-by-default case.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: rustyconover

vgi/src/main/java/farm/query/vgi/Worker.java

@@ -40,6 +40,7 @@ public final class Worker {
     private final Map<String, String> catalogTags = new LinkedHashMap<>();
     private String defaultSchema = "main";
     private final Map<String, String> schemaComments = new LinkedHashMap<>();
+    private final Map<String, Map<String, String>> schemaTags = new LinkedHashMap<>();
     private String implementationVersion;
     private String dataVersionSpec;
     private final List<CatalogDataVersionRelease> releases = new ArrayList<>();
@@ -316,6 +317,29 @@ public Worker schemaComment(String schema, String comment) {
      */
     public Map<String, String> schemaComments() { return schemaComments; }
 
+    /**
+     * Attach key/value metadata tags to a schema, surfaced via
+     * {@code catalog_schemas} / {@code catalog_schema_get} and reported through
+     * DuckDB's {@code duckdb_schemas().tags}. Typical keys are
+     * {@code vgi.description_llm} and {@code vgi.description_md}. Merged into any
+     * tags previously set for the schema (later calls overwrite duplicate keys).
+     *
+     * @param schema the schema name
+     * @param tags   schema tag key/value pairs to merge in
+     * @return this builder
+     */
+    public Worker schemaTags(String schema, Map<String, String> tags) {
+        schemaTags.computeIfAbsent(schema, k -> new LinkedHashMap<>()).putAll(tags);
+        return this;
+    }
+
+    /**
+     * Tags registered via {@link #schemaTags(String, Map)}.
+     *
+     * @return the per-schema tags keyed by schema name
+     */
+    public Map<String, Map<String, String>> schemaTags() { return schemaTags; }
+
     /**
      * An auxiliary catalog served by the same worker process next to the main
      * catalog, MetaWorker-style: it appears as its own row in

vgi/src/main/java/farm/query/vgi/internal/VgiServiceImpl.java

@@ -989,7 +989,7 @@ public ItemsResponse catalog_schemas(byte[] attach_opaque_data, byte[] transacti
         List<byte[]> items = new ArrayList<>();
         for (SchemaDesc s : workerSchemas()) {
             items.add(RecordCodec.serializeToBytes(
-                    new SchemaInfo(s.comment, Map.of(), attach_opaque_data, s.name, schemaCounts(s))));
+                    new SchemaInfo(s.comment, s.tags, attach_opaque_data, s.name, schemaCounts(s))));
         }
         return new ItemsResponse(items);
     }
@@ -1014,14 +1014,14 @@ public ItemsResponse catalog_schema_get(byte[] attach_opaque_data, String name,
         for (SchemaDesc s : workerSchemas()) {
             if (s.name.equals(name)) {
                 return new ItemsResponse(List.of(RecordCodec.serializeToBytes(
-                        new SchemaInfo(s.comment, Map.of(), attach_opaque_data, name, schemaCounts(s)))));
+                        new SchemaInfo(s.comment, s.tags, attach_opaque_data, name, schemaCounts(s)))));
             }
         }
         return ItemsResponse.empty();
     }
 
     /** Schema descriptors registered with the worker (default + auxiliary). */
-    private record SchemaDesc(String name, String comment) {}
+    private record SchemaDesc(String name, String comment, Map<String, String> tags) {}
 
     /**
      * The auxiliary catalog an attach belongs to, or {@code null} for the main
@@ -1107,9 +1107,11 @@ private Map<String, Long> extraSchemaCounts(Worker.ExtraCatalog extra) {
     private List<SchemaDesc> workerSchemas() {
         List<SchemaDesc> result = new ArrayList<>();
         Map<String, String> comments = worker.schemaComments();
+        Map<String, Map<String, String>> tags = worker.schemaTags();
         String defaultSchema = worker.defaultSchema();
         result.add(new SchemaDesc(defaultSchema,
-                comments.getOrDefault(defaultSchema, "Default schema")));
+                comments.getOrDefault(defaultSchema, "Default schema"),
+                tags.getOrDefault(defaultSchema, Map.of())));
         // The minimal `versioned` fixture exposes only its default schema
         // (vgi-python parity); the example worker's auxiliary `data` schema is
         // not part of it. attach/versioning_http.test asserts exactly one schema.
@@ -1124,7 +1126,10 @@ private List<SchemaDesc> workerSchemas() {
         for (var m : worker.macros()) {
             if (!defaultSchema.equals(m.schema())) extras.add(m.schema());
         }
-        for (String s : extras) result.add(new SchemaDesc(s, comments.getOrDefault(s, "")));
+        for (String s : extras) {
+            result.add(new SchemaDesc(s, comments.getOrDefault(s, ""),
+                    tags.getOrDefault(s, Map.of())));
+        }
         return result;
     }
 

vgi/src/test/java/farm/query/vgi/internal/SchemaInfoTagsTest.java

@@ -0,0 +1,73 @@
+// Copyright 2026 Query Farm LLC - https://query.farm
+
+package farm.query.vgi.internal;
+
+import farm.query.vgi.Worker;
+import farm.query.vgi.protocol.ItemsResponse;
+import farm.query.vgi.protocol.SchemaInfo;
+import farm.query.vgirpc.marshal.RecordCodec;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Verifies that per-schema tags declared on the {@link Worker} via
+ * {@link Worker#schemaTags(String, Map)} flow through the
+ * {@code catalog_schemas} enumeration onto the wire {@link SchemaInfo#tags()}
+ * (surfaced by DuckDB as {@code duckdb_schemas().tags}). The default — no tags
+ * configured — yields an empty tag map.
+ */
+class SchemaInfoTagsTest {
+
+    private static VgiServiceImpl service(Worker w) {
+        // sealOpaqueData=false ⇒ stdio/AF_UNIX contract: the attach token is the
+        // plaintext id, so we can drive catalog_schemas directly with any bytes.
+        return new VgiServiceImpl(w, List.of(), List.of(), List.of(), List.of(), false);
+    }
+
+    private static SchemaInfo onlySchema(ItemsResponse resp) {
+        List<byte[]> items = resp.items();
+        assertEquals(1, items.size(), "expected exactly one schema");
+        return RecordCodec.deserializeFromBytes(items.get(0), SchemaInfo.class);
+    }
+
+    @Test
+    void schemaTagsSurfaceOnSchemaInfo() {
+        Worker w = Worker.builder()
+                .schemaTags("main", Map.of(
+                        "vgi.description_llm", "Main schema for testing.",
+                        "vgi.description_md", "# Main\nMain schema."));
+
+        SchemaInfo info = onlySchema(service(w).catalog_schemas("attach".getBytes(), null));
+
+        assertEquals("main", info.name());
+        assertEquals("Main schema for testing.", info.tags().get("vgi.description_llm"));
+        assertEquals("# Main\nMain schema.", info.tags().get("vgi.description_md"));
+        assertEquals(2, info.tags().size());
+    }
+
+    @Test
+    void schemaTagsMergeAcrossCalls() {
+        Worker w = Worker.builder()
+                .schemaTags("main", Map.of("a", "1"))
+                .schemaTags("main", Map.of("b", "2"));
+
+        SchemaInfo info = onlySchema(service(w).catalog_schemas("attach".getBytes(), null));
+
+        assertEquals("1", info.tags().get("a"));
+        assertEquals("2", info.tags().get("b"));
+    }
+
+    @Test
+    void noTagsConfiguredYieldsEmptyMap() {
+        SchemaInfo info = onlySchema(
+                service(Worker.builder()).catalog_schemas("attach".getBytes(), null));
+
+        assertEquals("main", info.name());
+        assertTrue(info.tags().isEmpty(), "tags should default to empty");
+    }
+}