@@ -8,6 +8,7 @@ Each document declares its own status; “proposed” means the contract is stil
 | [VGI Filter Encoding v2](vgi-filter-encoding-v2-spec.md) | **Normative, proposed** | Engine-neutral wire format, semantics, validation, and conformance requirements for `vgi.filters.v2`. |
 | [VGI Runtime-Filter Artifacts](vgi-runtime-filter-artifacts.md) | **Normative optional extension, proposed** | Additional requirements for capability-gated advisory runtime-pruning artifacts. Base filter-v2 conformance does not require this extension. |
 | [VGI DuckDB Filter Adapter](vgi-duckdb-filter-adapter.md) | **Informative implementation guide, proposed** | DuckDB 1.5 and 2.0 mappings into the normative filter-v2 wire contract. |
+| [VGI Secrets](vgi-secrets.md) | **Normative** | How a worker asks for client-side secrets, the two-phase bind, and the layout of the secrets batch. |
 | [VGI catalog query pushdown](vgi-catalog-query-pushdown-design.md) | **Informative implementation design, proposed** | Updated 2026-10-06: proposed VGI 2.2 query preparation, DuckDB 1.5 explicit execution, and DuckDB 2.0 bound-plan SQL export for automatic pushdown. |
 | [Proposed VGI protocol changes for DuckDB 2.0](vgi-protocol-proposed-changes.md) | **Informative design/audit report** | Rationale, migration inventory, implementation status, and deferred protocol decisions. It is not an implementation specification. |
 
 
@@ -0,0 +1,131 @@
+# VGI Secrets
+
+**Status:** Normative. It records the contract the DuckDB extension and the SDKs already implement.
+**Protocol:** `vgi.v2`. No wire change.
+**Reference implementations:** client, `BuildSecretsBatch` and `ExtractVgiSecrets` in the DuckDB
+extension; worker, `SecretsAccessor` and `ResolvedSecrets` in vgi-python `vgi/table_function.py`.
+
+The words MUST, MUST NOT and SHOULD are used as in RFC 2119.
+
+## 1. Purpose
+
+A function reads credentials that the user keeps in the client's secret store (in DuckDB,
+`CREATE SECRET`). The worker never reads that store. It says which secrets it needs, the client
+resolves them, and the resolved values travel to the worker inside the bind request.
+
+Two things are specified here: how a worker asks for a secret, and the layout of the batch that
+carries the answers.
+
+## 2. Asking for a secret
+
+A lookup is a triple:
+
+| Field | Meaning |
+|---|---|
+| `secret_type` | Required. The secret type, such as `s3`. |
+| `secret_name` | Optional. The name of one specific secret. |
+| `scope` | Optional. A path the secret must apply to, such as `s3://bucket/key`. |
+
+A worker asks in one of two ways.
+
+**Statically.** Function metadata lists lookups in `required_secrets`. The client resolves them
+before the first bind, so they are in the first bind request.
+
+**Dynamically.** A bind may need a lookup it can only compute from its arguments, such as the bucket
+in a path argument. The worker answers the bind with a *secret-scope request*: a `BindResponse`
+whose `lookup_secret_types` list is non-empty. `lookup_scopes` and `lookup_names` are parallel
+lists of the same length, with the empty string where a lookup has no scope or no name. The other
+fields of that response are ignored.
+
+The client resolves the requested lookups, adds them to the secrets it already resolved, and sends
+the bind again with `resolved_secrets_provided = true`. A worker MUST NOT answer that second bind
+with another secret-scope request; the client rejects it. A lookup that is still unanswered on the
+second bind does not exist.
+
+## 3. How the client resolves a lookup
+
+- **With a name:** the secret of that name, if its type equals `secret_type`. The scope is ignored.
+- **Without a name:** the secret of that type whose scope best matches `scope` under the client's
+  own rule. DuckDB uses the longest scope that is a prefix of the path. An empty `scope` matches
+  only secrets that have no scope.
+
+A lookup that matches nothing is not an error for the client. The secret is left out of the batch
+and the worker decides what that means.
+
+A secret type that sets a default scope is not matched by a lookup with an empty scope. DuckDB's
+`s3` secrets default to the scope `s3://`, so a worker looking for "the user's S3 secret" asks for
+scope `s3://`, not for the type alone.
+
+## 4. The secrets batch
+
+`BindRequest.secrets` is a binary field. It holds an Arrow IPC stream with one record batch, or is
+null when no secret was resolved.
+
+The batch has **one row** and **one column per resolved secret**:
+
+- The column is **named by the secret's name**, the name the user gave it or the client's default
+  name. It is not named by the secret's type. Several secrets of one type therefore coexist, one
+  per bucket for example.
+- The column is a **struct**. Its fields are the secret's own key/value pairs, each with the Arrow
+  type of its value, plus the four fields below.
+- A secret that was resolved by more than one lookup appears once.
+
+| Field | Type | Content |
+|---|---|---|
+| `type` | utf8 | The secret type. |
+| `provider` | utf8 | The provider that created the secret, such as `config` or `credential_chain`. |
+| `name` | utf8 | The secret name. Equal to the column name. |
+| `scope` | utf8 | The secret's scope prefixes joined by a newline (`\n`). Empty when it has none. |
+
+Example, for two `s3` secrets named `prod` and `dev`:
+
+```text
+prod: struct<key_id: utf8, name: utf8, provider: utf8, region: utf8, scope: utf8, secret: utf8, type: utf8>
+dev:  struct<key_id: utf8, name: utf8, provider: utf8, region: utf8, scope: utf8, secret: utf8, type: utf8>
+```
+
+Rules:
+
+1. A client MUST name each column by the secret name and MUST include `type`, `name` and `scope`.
+2. A worker MUST find a secret by column name or by the `type` and `scope` fields. It MUST NOT
+   depend on column order, on column position, or on Arrow field metadata.
+3. A worker MUST accept struct fields it does not know, and MUST treat a null column as an
+   unresolved secret.
+4. The batch does not say which lookup produced which secret. A worker matches them itself (§5).
+5. The batch contains credentials in plain text. Neither side may log it or copy it into
+   `opaque_data`.
+
+The same batch is sent wherever a bind request is sent or embedded, which includes the bind request
+carried inside each init request. A function therefore sees the same secrets at execution time as
+at bind time.
+
+## 5. How the worker finds its secret
+
+A worker selects from the batch with the rule the client used, so that it picks the secret its own
+lookup resolved:
+
+- **By name:** the column of that name, if its `type` field equals the requested type.
+- **By scope:** among secrets of the requested type, the one with the longest scope prefix of the
+  requested path; failing that, one with an empty scope.
+- **By type alone:** a secret of that type with an empty scope; failing that, the first secret of
+  that type.
+
+SDKs expose these as selectors on the resolved-secrets object.
+
+| SDK | By type | By scope and type |
+|---|---|---|
+| Python | `ResolvedSecrets.of_type`, `.lookup` | `.for_scope_of_type`, `.lookup` |
+| Rust | `Secrets::of_type` | `Secrets::for_scope_of_type` |
+| Go | `Secrets.OfType` | `Secrets.ForScopeOfType` |
+| C++ | `Secrets::of_type` | `Secrets::for_scope_of_type` |
+| Java | `Secrets.ofType` | `Secrets.forScopeOfType` |
+| C# | `SecretArgCodec.AllOfType` | `SecretArgCodec.ForScopeOfType` |
+| TypeScript | `secretsOfType` | `secretForScopeOfType` |
+
+## 6. History
+
+vgi-python's bind-time `SecretsAccessor` used to expect dynamically requested secrets in columns
+named `secret_0`, `secret_1`, … with the lookup's type, name and scope in Arrow field metadata. No
+client sends that layout, so `params.secrets.get(type, name=…)` and `get(type, scope=…)` returned
+`None` on the second bind although the secret was in the batch. The accessor now reads the layout
+in §4, and `secret_0` is an ordinary secret name.
@@ -150,6 +150,7 @@ nav:
     - "Catalog query pushdown (proposed design)": protocol/vgi-catalog-query-pushdown-design.md
     - "DuckDB 2.0 design audit (informative)": protocol/vgi-protocol-proposed-changes.md
     - "Attach tickets (normative)": protocol/vgi-attach-tickets.md
+    - "Secrets (normative)": protocol/vgi-secrets.md
   - API Reference:
     - api/index.md
     - Functions: api/functions.md
 
@@ -9,6 +9,7 @@
 from typing import Annotated, Any
 
 import pyarrow as pa
+import pytest
 
 from tests.conftest import FIXTURE_WORKER
 from vgi.arguments import (
@@ -418,6 +419,16 @@ def test_bind_params_secrets_type(self) -> None:
         assert len(outputs) > 0
 
 
+def _wire_secrets(*secrets: dict[str, str]) -> pa.RecordBatch:
+    """Build a secrets batch the way the DuckDB extension does.
+
+    One row, one struct column per secret, the column named by the secret's
+    ``name``; ``type``, ``provider``, ``name`` and ``scope`` are fields.
+    """
+    columns = {s["name"]: [{"provider": "config", **s}] for s in secrets}
+    return pa.RecordBatch.from_pydict(columns)
+
+
 # ---------------------------------------------------------------------------
 # Tests for SecretsAccessor
 # ---------------------------------------------------------------------------
@@ -461,57 +472,157 @@ def test_scoped_get_registers_pending(self) -> None:
         assert entry.secret_type == "s3"
         assert entry.scope == "s3://bucket/"
 
-    def test_to_dict_combines_unscoped_and_scoped(self) -> None:
-        """to_dict() includes both unscoped and scoped entries."""
-        # Build a batch with an unscoped column and a scoped column
-        struct_type = pa.struct([("k", pa.string())])
-        unscoped_arr = pa.array([{"k": "unscoped_val"}], type=struct_type)
-
-        scoped_field = pa.field(
-            "secret_0",
-            struct_type,
-            metadata={"secret_type": "scoped_type", "scope": "s3://x/"},
+    def test_to_dict_is_keyed_by_secret_name(self) -> None:
+        """to_dict() keeps every resolved secret under its own name, nulls dropped."""
+        batch = _wire_secrets(
+            {"type": "s3", "name": "prod", "scope": "s3://prod/", "key_id": "P"},
+            {"type": "s3", "name": "dev", "scope": "s3://dev/", "key_id": "D"},
         )
-        scoped_arr = pa.array([{"k": "scoped_val"}], type=struct_type)
+        struct_type = pa.struct([("type", pa.string())])
+        batch = batch.append_column(pa.field("gone", struct_type), pa.array([None], type=struct_type))
+        result = SecretsAccessor(batch).to_dict()
+        assert sorted(result) == ["dev", "prod"]
+        assert result["prod"]["key_id"].as_py() == "P"
 
-        batch = pa.RecordBatch.from_arrays(
-            [unscoped_arr, scoped_arr],
-            schema=pa.schema([pa.field("my_type", struct_type), scoped_field]),
+    def test_a_secret_named_like_the_old_wire_columns_is_an_ordinary_secret(self) -> None:
+        """A user may name a secret ``secret_0``; it is not a special column."""
+        batch = _wire_secrets({"type": "s3", "name": "secret_0", "scope": "s3://", "key_id": "K"})
+        accessor = SecretsAccessor(batch, is_retry=True)
+        assert accessor.to_dict()["secret_0"]["key_id"].as_py() == "K"
+        found = accessor.get("s3", name="secret_0")
+        assert found is not None
+        assert found["key_id"].as_py() == "K"
+
+    def test_get_by_name_on_retry_returns_the_secret(self) -> None:
+        """The two-phase bind: a by-name lookup registers, then the retry returns it.
+
+        Regression: the retry looked for ``secret_N`` columns that no client
+        sends, so the lookup came back ``None`` although the secret was in the
+        batch, and ``required=True`` raised.
+        """
+        first = SecretsAccessor(None)
+        assert first.get("s3", name="prod", required=True) is None
+        assert first.pending_lookups == [SecretLookupEntry(secret_type="s3", secret_name="prod")]
+
+        batch = _wire_secrets({"type": "s3", "name": "prod", "scope": "s3://", "key_id": "P"})
+        retry = SecretsAccessor(batch, is_retry=True)
+        found = retry.get("s3", name="prod", required=True)
+        assert found is not None
+        assert found["key_id"].as_py() == "P"
+        assert retry.all_resolved
+
+    def test_get_by_name_checks_the_type(self) -> None:
+        """A secret with the right name but another type is not a match."""
+        batch = _wire_secrets({"type": "gcs", "name": "prod", "scope": "gcs://"})
+        accessor = SecretsAccessor(batch, is_retry=True)
+        assert accessor.get("s3", name="prod") is None
+        with pytest.raises(ValueError, match="Required secret 's3' not found"):
+            accessor.get("s3", name="prod", required=True)
+
+    def test_get_by_name_already_resolved_needs_no_retry(self) -> None:
+        """A by-name lookup the client pre-resolved is answered on the first bind."""
+        batch = _wire_secrets({"type": "s3", "name": "prod", "scope": "s3://", "key_id": "P"})
+        accessor = SecretsAccessor(batch)
+        assert accessor.get("s3", name="prod") is not None
+        assert not accessor.needs_resolution
+
+    def test_get_by_scope_on_retry_picks_the_longest_prefix(self) -> None:
+        """A scope lookup returns the secret whose scope is the longest prefix."""
+        batch = _wire_secrets(
+            {"type": "s3", "name": "any", "scope": "s3://", "key_id": "ANY"},
+            {"type": "s3", "name": "prod", "scope": "s3://prod/\ns3n://prod/", "key_id": "PROD"},
+            {"type": "gcs", "name": "other", "scope": "s3://prod/data/", "key_id": "WRONG-TYPE"},
         )
-        result = SecretsAccessor(batch).to_dict()
-        assert "my_type" in result
-        assert result["my_type"]["k"].as_py() == "unscoped_val"
-        assert "scoped_type" in result
-        assert result["scoped_type"]["k"].as_py() == "scoped_val"
-
-    def test_to_dict_skips_null_scoped(self) -> None:
-        """to_dict() skips scoped entries with null values."""
-        struct_type = pa.struct([("k", pa.string())])
-        scoped_field = pa.field(
-            "secret_0",
-            struct_type,
-            metadata={"secret_type": "null_type"},
+        accessor = SecretsAccessor(batch, is_retry=True)
+        prod = accessor.get("s3", scope="s3://prod/data/x.parquet", required=True)
+        assert prod is not None and prod["key_id"].as_py() == "PROD"
+        other = accessor.get("s3", scope="s3://elsewhere/x.parquet", required=True)
+        assert other is not None and other["key_id"].as_py() == "ANY"
+        assert accessor.get("s3", scope="az://nope/") is None
+
+    def test_get_by_scope_always_asks_the_client_first(self) -> None:
+        """Only the client knows whether a longer-scoped secret exists."""
+        batch = _wire_secrets({"type": "s3", "name": "any", "scope": "s3://", "key_id": "ANY"})
+        accessor = SecretsAccessor(batch)
+        assert accessor.get("s3", scope="s3://prod/x") is None
+        assert accessor.pending_lookups == [SecretLookupEntry(secret_type="s3", scope="s3://prod/x")]
+
+    def test_get_by_type_finds_a_secret_whatever_its_name(self) -> None:
+        """A by-type lookup matches the ``type`` field, not the column name."""
+        batch = _wire_secrets(
+            {"type": "vgi_example", "name": "scoped", "scope": "x://a/", "v": "scoped"},
+            {"type": "vgi_example", "name": "__default_vgi_example", "scope": "", "v": "default"},
+            {"type": "s3", "name": "aaa", "scope": "", "v": "s3"},
         )
-        scoped_arr = pa.array([None], type=struct_type)
-        batch = pa.RecordBatch.from_arrays(
-            [scoped_arr],
-            schema=pa.schema([scoped_field]),
+        accessor = SecretsAccessor(batch)
+        found = accessor.get("vgi_example")
+        assert found is not None
+        assert found["v"].as_py() == "default"
+        assert not accessor.needs_resolution
+
+    def test_get_by_type_falls_back_to_a_scoped_secret(self) -> None:
+        """With no unscoped secret of the type, the first one of that type is used."""
+        batch = _wire_secrets({"type": "s3", "name": "prod", "scope": "s3://", "v": "p"})
+        found = SecretsAccessor(batch).get("s3")
+        assert found is not None
+        assert found["v"].as_py() == "p"
+
+    def test_annotated_secret_is_selected_by_name_and_scope(self) -> None:
+        """A ``Secret()`` annotation on on_bind gets the secret it names, through a real bind.
+
+        Regression: the table-function path looked for a column named by the
+        secret *type*, so an annotated secret arrived as ``None`` unless the
+        user had named the secret after its type.
+        """
+        from vgi.arguments import Arguments
+        from vgi.invocation import FunctionType
+        from vgi.protocol import BindRequest
+
+        seen: dict[str, Any] = {}
+
+        @dataclass(slots=True, frozen=True)
+        class _Args:
+            count: Annotated[int, Arg(0, doc="Count")] = 10
+
+        class _TestFunc(TableFunctionGenerator[_Args]):
+            @classmethod
+            def on_bind(
+                cls,
+                params: BindParams[_Args],
+                *,
+                by_name: Annotated[dict[str, pa.Scalar[Any]] | None, Secret("s3", name="dev")] = None,
+                by_scope: Annotated[dict[str, pa.Scalar[Any]] | None, Secret("s3", scope="s3://prod/x")] = None,
+                missing: Annotated[dict[str, pa.Scalar[Any]] | None, Secret("gcs")] = None,
+            ) -> BindResponse:
+                seen.update(by_name=by_name, by_scope=by_scope, missing=missing)
+                return BindResponse(output_schema=pa.schema([pa.field("x", pa.int64())]))
+
+            @classmethod
+            def process(cls, params: Any, state: Any, out: Any) -> None: ...
+
+        request = BindRequest(
+            function_name="_test_func",
+            arguments=Arguments(positional=(pa.scalar(1),)),
+            function_type=FunctionType.TABLE,
+            secrets=_wire_secrets(
+                {"type": "s3", "name": "dev", "scope": "s3://dev/", "key_id": "DEV"},
+                {"type": "s3", "name": "prod", "scope": "s3://prod/", "key_id": "PROD"},
+            ),
         )
-        result = SecretsAccessor(batch).to_dict()
-        assert "null_type" not in result
+        _TestFunc.bind(request)
+
+        assert seen["by_name"]["key_id"].as_py() == "DEV"
+        assert seen["by_scope"]["key_id"].as_py() == "PROD"
+        assert seen["missing"] is None
 
     def test_required_raises_on_retry(self) -> None:
         """required=True raises ValueError when secret is genuinely missing on retry."""
-        import pytest
-
         accessor = SecretsAccessor(None, is_retry=True)
         with pytest.raises(ValueError, match="Required secret"):
             accessor.get("missing_type", required=True)
 
     def test_scoped_required_raises_on_retry(self) -> None:
         """required=True with scope raises ValueError on retry when not found."""
-        import pytest
-
         accessor = SecretsAccessor(None, is_retry=True)
         with pytest.raises(ValueError, match="Required secret"):
             accessor.get("s3", scope="s3://bucket/", required=True)
@@ -557,7 +668,6 @@ def test_non_serializable_state_raises(self) -> None:
         still rejected, and rejected at class-definition time rather than
         silently falling back to initial_state() on every HTTP exchange.
         """
-        import pytest
 
         @dataclass
         class BadState: