@@ -303,3 +303,44 @@ def test_cache_is_bounded(self) -> None:
         # Least-recently-used entries evicted; the newest survive.
         assert cache.get((bytes([19]), b"id")) == bytes([19])
         assert cache.get((bytes([0]), b"id")) is None
+
+
+class TestMetaWorkerCanOpenTheAttach:
+    """A MetaWorker must satisfy the same rehydrate contract a Worker does.
+
+    Since the stream state stopped carrying the attach decrypted, the rehydrate
+    path re-opens the sealed envelope on whatever the transport hands it as the
+    implementation. Under a MetaWorker that object is *not* a ``Worker`` —
+    MetaWorker does not subclass it, it proxies — so the method has to exist
+    there too. It did not, and every HTTP continuation of a catalog-backed
+    stream died on ``'MetaWorker' object has no attribute
+    '_unwrap_attach_full'``. Unit coverage missed it because nothing here drove
+    a MetaWorker over HTTP; the integration suite caught it.
+    """
+
+    @staticmethod
+    def _meta_and_worker() -> tuple[Any, Any]:
+        from vgi._test_fixtures.worker import ExampleWorker
+        from vgi.meta_worker import MetaWorker
+
+        worker = ExampleWorker(quiet=True)
+        worker._signing_key = _KEY
+        return MetaWorker([worker]), worker
+
+    def test_meta_worker_exposes_the_rehydrate_hook(self) -> None:
+        """The attribute the rehydrate path calls must exist on a MetaWorker."""
+        meta, _ = self._meta_and_worker()
+        assert hasattr(meta, "_unwrap_attach_full")
+
+    def test_meta_worker_unwraps_identically_to_its_sub_worker(self) -> None:
+        """Routing through the MetaWorker yields the sub-worker's plaintext."""
+        meta, worker = self._meta_and_worker()
+        with _as_principal(_DOMAIN, _PRINCIPAL):
+            envelope = bytes(worker._seal_attach(_UUID + _SECRET))
+            assert meta._unwrap_attach_full(envelope) == _UUID + _SECRET
+            assert meta._unwrap_attach_full(envelope) == worker._unwrap_attach_full(envelope)
+
+    def test_no_attach_passes_through(self) -> None:
+        """No catalog context means nothing to open."""
+        meta, _ = self._meta_and_worker()
+        assert meta._unwrap_attach_full(None) is None
@@ -1248,3 +1248,52 @@ def test_namespace_survives_a_round_trip(self) -> None:
         # The scan itself must accept what came back: _coerce_ns takes the
         # reserved b"_vgi/" prefix only from a genuine enum member.
         assert _coerce_ns(restored.ns) == b"_vgi/streaming_finalize"
+
+
+class TestPartitionFieldsCacheKeyIsMetadataAware:
+    """Partition-field lookup must not conflate schemas that differ only in metadata.
+
+    ``pa.Schema.__eq__``/``__hash__`` ignore field metadata, so an annotated
+    schema is equal to and hashes the same as an unannotated one of the same
+    names and types. A cache keyed straight on the schema therefore answers
+    whichever was scanned first for both — which made a function with no
+    partition annotations inherit an annotated neighbour's fields, so the
+    ``partition_values`` contract check passed where it had to raise.
+    """
+
+    @staticmethod
+    def _pair() -> tuple[pa.Schema, pa.Schema]:
+        from vgi.schema_utils import VGI_PARTITION_COLUMN_KEY
+
+        plain = pa.schema([("country", pa.string()), ("sales", pa.int64())])
+        annotated = pa.schema(
+            [
+                pa.field("country", pa.string(), metadata={VGI_PARTITION_COLUMN_KEY: b"true"}),
+                ("sales", pa.int64()),
+            ]
+        )
+        return plain, annotated
+
+    def test_arrow_equality_still_ignores_metadata(self) -> None:
+        """The premise: this is why keying on the schema alone is unsafe."""
+        plain, annotated = self._pair()
+        assert plain == annotated
+        assert hash(plain) == hash(annotated)
+        assert not plain.equals(annotated, check_metadata=True)
+
+    @pytest.mark.parametrize("annotated_first", [True, False])
+    def test_lookup_is_correct_in_either_cache_order(self, annotated_first: bool) -> None:
+        """Whichever schema populates the cache first, both answers stay right."""
+        from vgi.protocol import _partition_fields_cached, _partition_fields_from_schema
+
+        plain, annotated = self._pair()
+        _partition_fields_cached.cache_clear()
+        try:
+            first, second = (annotated, plain) if annotated_first else (plain, annotated)
+            _partition_fields_from_schema(first)
+            _partition_fields_from_schema(second)
+
+            assert [f.name for f in _partition_fields_from_schema(annotated)] == ["country"]
+            assert _partition_fields_from_schema(plain) == []
+        finally:
+            _partition_fields_cached.cache_clear()
@@ -247,6 +247,37 @@ def _worker_for_unattached(self, function_name: str, schema_name: str | None) ->
             raise ValueError(msg)
         return candidates[0]
 
+    def _unwrap_attach_full(self, envelope: bytes | None) -> bytes | None:
+        """Open an attach envelope on the sub-worker that owns its catalog.
+
+        A stream's state no longer carries the attach decrypted, so the
+        rehydrate path re-opens the sealed envelope on whatever object the
+        transport hands it as the implementation — which under a MetaWorker is
+        *this* object, not a `[`Worker`][]`. Without this the HTTP continuation
+        of any catalog-backed stream dies on ``'MetaWorker' object has no
+        attribute '_unwrap_attach_full'``.
+
+        Routes the same way :meth:`_resolve_function` does, so the plaintext is
+        produced by the sub-worker that owns the catalog. Every sub-worker in a
+        process shares one signing key (``_maybe_worker_for_attach`` itself
+        opens the envelope through ``_workers[0]`` to read the catalog name),
+        so an attach naming no catalog this process knows can still be opened —
+        fall back to the first worker rather than failing, and let the sub-worker
+        raise if the seal genuinely does not verify.
+
+        Args:
+            envelope: The sealed ``attach_opaque_data``, or ``None``.
+
+        Returns:
+            The full framework plaintext (``uuid || catalog_bytes``), or
+            ``None`` when there is no attach.
+
+        """
+        if envelope is None:
+            return None
+        worker = self._maybe_worker_for_attach(envelope) or self._workers[0]
+        return worker._unwrap_attach_full(envelope)
+
     def _resolve_function(self, request: BindRequest) -> Any:
         """Dispatch function-class resolution to the worker that hosts it.
 
 
@@ -954,12 +954,44 @@ def _partition_fields_from_schema(bind_schema: pa.Schema) -> list[pa.Field[Any]]
     once at wrapper construction, so per-emit validation only does an
     O(P) walk where P is the partition column count.
     """
-    return list(_partition_fields_cached(bind_schema))
+    return list(_partition_fields_cached(_SchemaKey(bind_schema)))
+
+
+class _SchemaKey:
+    """Cache key for a schema whose *field metadata* is significant.
+
+    ``pa.Schema.__eq__`` and ``__hash__`` ignore field metadata, so a schema
+    annotated with partition columns is equal to — and hashes the same as — an
+    unannotated schema of the same names and types. Keying a cache directly on
+    a schema therefore conflates the two, and whichever was scanned first wins
+    for both. That is not hypothetical: it made a function with no annotations
+    inherit an annotated neighbour's partition fields, so the
+    ``partition_values`` contract check silently passed where it had to raise,
+    and it reproduced only on whichever transport happened to populate the
+    cache in the losing order.
+
+    The metadata-blind hash is kept deliberately — a hash may collide, it just
+    may not lie — and equality does the discriminating via
+    ``equals(check_metadata=True)``, measured at 0.042us.
+    """
+
+    __slots__ = ("schema",)
+
+    def __init__(self, schema: pa.Schema) -> None:
+        self.schema = schema
+
+    def __hash__(self) -> int:
+        return hash(self.schema)
+
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, _SchemaKey):
+            return NotImplemented
+        return bool(self.schema.equals(other.schema, check_metadata=True))
 
 
 @lru_cache(maxsize=512)
-def _partition_fields_cached(bind_schema: pa.Schema) -> tuple[pa.Field[Any], ...]:
-    """Memoized partition-field scan, keyed on the schema.
+def _partition_fields_cached(key: _SchemaKey) -> tuple[pa.Field[Any], ...]:
+    """Memoized partition-field scan, keyed on the schema *including* metadata.
 
     Which columns are partition columns is a pure function of the schema's
     field metadata, and the schema is fixed for a stream -- but this ran on
@@ -975,7 +1007,7 @@ def _partition_fields_cached(bind_schema: pa.Schema) -> tuple[pa.Field[Any], ...
     from vgi.schema_utils import VGI_PARTITION_COLUMN_KEY
 
     return tuple(
-        f for f in bind_schema if f.metadata is not None and f.metadata.get(VGI_PARTITION_COLUMN_KEY) == b"true"
+        f for f in key.schema if f.metadata is not None and f.metadata.get(VGI_PARTITION_COLUMN_KEY) == b"true"
     )