diff --git a/docs/data-stream-lifecycle.md b/docs/data-stream-lifecycle.md
index 5c95e6e..6a571f8 100644
--- a/docs/data-stream-lifecycle.md
+++ b/docs/data-stream-lifecycle.md
@@ -68,7 +68,7 @@ ref_uri  →  (point_uri, source_id, ref_name, value_kind)
 
 This table is populated three ways:
 
-- **On stream registration** (`register_stream` / `register_streams`): the client writes the graph triple `point_uri → ref:hasExternalReference → ref_uri` when `point_uri` is known, and the server's `_sync_stream_refs_from_graph` method scans for these triples and upserts them into the streams table.
+- **On stream registration** (`register_stream` / `register_streams`): the client always writes the managed reference node carrying `acq:sourceId` and `acq:refName`. When `point_uri` is known it also writes `point_uri → ref:hasExternalReference → ref_uri`. The server's `_sync_stream_refs_from_graph` method scans for managed reference nodes and upserts them into the streams table, recording `point_uri` when that link exists and `NULL` when it does not.
 - **On data insert** (`insert_timeseries`, `insert_timeseries_batch`, `insert_timeseries_arrow`): the server computes `ref_uri` from `(source_id, ref_name)` and upserts a streams row even when no `point_uri` is known yet. In that case `streams.point_uri` is `NULL`.
 - **On graph insert**: any time RDF is inserted, the server re-scans the graph for managed reference patterns.
 
@@ -113,7 +113,16 @@ The RDF graph for a managed stream looks like this:
     rdfs:label "mybox-system-metrics" .
 ```
 
-The `ref_uri` node (the UUID5 URI) is intentionally thin. It is an indirection node, not a description of the measurement itself. Semantic metadata goes on `point_uri`; provenance and routing metadata goes on `ref_uri`.
+The `ref_uri` node (the UUID5 URI) is intentionally thin when a semantic point
+exists. It is an indirection node first, and a semantic fallback second:
+
+- when `point_uri` is present, semantic metadata such as unit and quantity kind
+  is written on the point node, while provenance/routing metadata lives on
+  `ref_uri`
+- when `point_uri` is absent, Acquirium writes stream-level semantic metadata
+  such as unit, quantity kind, medium, substance, and data source on `ref_uri`
+  so the stream is still queryable and self-describing before a point node is
+  minted later
 
 ---
 
@@ -169,9 +178,23 @@ aq.register_streams([
 Registration:
 - writes the external reference node with `acq:sourceId`, `acq:refName`, `acq:valueKind`, and `ref:storedAt`
 - if `point_uri` is provided, also creates the point node and links it to the ref node
-- resolves plain-text unit/quantity_kind strings to QUDT URIs via the server's embedding matcher
+- writes semantic stream metadata (`unit`, `quantity_kind`, `medium`, `substance`, `data_source`) on the point node when `point_uri` exists, otherwise on the managed ref node
+- resolves plain-text semantic fields jointly per stream before writing them:
+  `unit` resolves as a unit, `quantity_kind` as a quantity kind, and
+  `medium`/`substance` as substances
+- passes through values already supplied as URI strings, `rdflib.URIRef`, or `None`
+- uses any already-known sibling URIs as disambiguation context for the remaining text fields during joint resolution
 - triggers `_sync_stream_refs_from_graph`, which upserts the mapping into the streams table
 
+The joint-resolution helper is intentionally conservative:
+
+- text fields are resolved together so a known quantity kind can disambiguate an
+  ambiguous unit, and vice versa
+- already-known URIs do not get re-resolved; they are forwarded unchanged and
+  only used as context
+- context is a rerank hint, not an override. Today it is mainly effective for
+  QUDT unit `<->` quantity-kind disambiguation
+
 Stream registration is purely a metadata operation. No timeseries rows are written.
 
 ### Step 3 — insert data
diff --git a/docs/drivers.md b/docs/drivers.md
index 0a1662a..326f0c9 100644
--- a/docs/drivers.md
+++ b/docs/drivers.md
@@ -400,6 +400,14 @@ CSV and XLSX drivers are polling tabular drivers. They watch a directory,
 normalize file rows to canonical observations, register streams as columns or
 IDs are discovered, and return rows for insertion by the common ingest base.
 
+When drivers call `aq.register_streams(...)`, the semantic metadata fields
+`unit`, `quantity_kind`, `medium`, and `substance` can be supplied either as
+plain text or as already-known URIs. Plain text is jointly resolved per stream
+before graph insertion; URI strings, `rdflib.URIRef`, and `None` values pass
+through unchanged. If `point_uri` is omitted, Acquirium still writes the
+managed reference node and stores that semantic metadata on the `ref_uri` so
+the stream remains queryable before a point node is added later.
+
 ```toml
 [[drivers]]
 spec = "acquirium.BuiltinDrivers.csv_ingest:CSVIngestDriver"
diff --git a/src/acquirium/Client/acquirium.py b/src/acquirium/Client/acquirium.py
index 64a9fcd..d8d917f 100644
--- a/src/acquirium/Client/acquirium.py
+++ b/src/acquirium/Client/acquirium.py
@@ -83,18 +83,24 @@ def _build_stream_triples(
         g.add((ref_uri, STORED_AT,           ACQUIRIUM_DB_URI))
         g.add((ACQUIRIUM_DB_URI, RDFS.label, Literal("Acquirium TimescaleDB")))
 
+    semantic_target = None
     if point_uri_raw is not None:
         subj = URIRef(str(point_uri_raw))
+        semantic_target = subj
         g.add((subj, RDF.type, VIRTUAL_POINT))
         if label is not None:
             g.add((subj, RDFS.label, Literal(label)))
-        _add_triple(g, subj, HAS_UNIT,          _coerce_resolved(resolved, "unit", stream.get("unit")))
-        _add_triple(g, subj, HAS_QUANTITY_KIND, _coerce_resolved(resolved, "quantity_kind", stream.get("quantity_kind")))
-        _add_triple(g, subj, HAS_MEDIUM,        _coerce_resolved(resolved, "medium", stream.get("medium")))
-        _add_triple(g, subj, OF_SUBSTANCE,      _coerce_resolved(resolved, "substance", stream.get("substance")))
-        _add_triple(g, subj, DATA_SOURCE, stream.get("data_source"))
         if ref_uri is not None:
             g.add((subj, HAS_EXTERNAL_REFERENCE, ref_uri))
+    elif ref_uri is not None:
+        semantic_target = ref_uri
+
+    if semantic_target is not None:
+        _add_triple(g, semantic_target, HAS_UNIT,          _coerce_resolved(resolved, "unit", stream.get("unit")))
+        _add_triple(g, semantic_target, HAS_QUANTITY_KIND, _coerce_resolved(resolved, "quantity_kind", stream.get("quantity_kind")))
+        _add_triple(g, semantic_target, HAS_MEDIUM,        _coerce_resolved(resolved, "medium", stream.get("medium")))
+        _add_triple(g, semantic_target, OF_SUBSTANCE,      _coerce_resolved(resolved, "substance", stream.get("substance")))
+        _add_triple(g, semantic_target, DATA_SOURCE, stream.get("data_source"))
 
     target = ref_uri if ref_uri is not None else (URIRef(str(point_uri_raw)) if point_uri_raw is not None else None)
     if target is not None:
diff --git a/src/acquirium/Client/client.py b/src/acquirium/Client/client.py
index 215dc20..cf12c80 100644
--- a/src/acquirium/Client/client.py
+++ b/src/acquirium/Client/client.py
@@ -3,6 +3,7 @@
 import requests
 from requests import HTTPError
 from pathlib import Path
+from rdflib import URIRef
 import polars as pl
 import pyarrow.ipc as ipc
 from acquirium.internals.models import (
@@ -14,6 +15,7 @@
     StreamInsert,
     RegisterDatasourceRequest,
     looks_like_uri,
+    split_record_uri_inputs,
 )
 from acquirium.internals.internals_namespaces import *
 from acquirium.Grafana.grafana_dashboard_creator import GrafanaDashboardCreator
@@ -281,15 +283,19 @@ def resolve_record(
         fields: dict[str, tuple[str, Optional[str]]],
         top_k: int = 5,
         min_score: float = 0.5,
+        context: Optional[list[str]] = None,
     ) -> dict[str, list[dict]]:
         """Jointly resolve a record's fields (server ``/resolve_record``).
 
         ``fields`` maps a caller-chosen label to ``(text, kind)``. The
         label is echoed back unchanged as the result key and is never read
-        by the resolver; resolution is driven by ``(text, kind)``. Returns
-        the ranked matches per label; related fields (e.g. a unit and its
-        quantity kind) reinforce each other server-side. The example labels
-        below mimic a historian export's column headers (real-source feel).
+        by the resolver; resolution is driven by ``(text, kind)``.
+        ``context`` is an optional list of already-chosen sibling URIs
+        that should participate in disambiguation even when those siblings
+        themselves do not need resolving. Returns the ranked matches per
+        label; related fields (e.g. a unit and its quantity kind) reinforce
+        each other server-side. The example labels below mimic a historian
+        export's column headers (real-source feel).
 
         Example::
 
@@ -306,6 +312,8 @@ def resolve_record(
             "top_k": top_k,
             "min_score": min_score,
         }
+        if context:
+            body["context"] = context
         response = requests.post(f"{self.base_url}/resolve_record", json=body)
         _raise_for_status(response)
         return response.json().get("matches", {})
@@ -314,15 +322,17 @@ def resolve_record_uris(
         self,
         fields: dict[str, tuple[Any, Optional[str]]],
         min_score: float = 0.5,
-    ) -> dict[str, Optional[str]]:
+    ) -> dict[str, str | URIRef | None]:
         """Jointly resolve a record to one best URI per field, or ``None``.
 
         Keys are caller-chosen labels echoed back unchanged (never read by
         the resolver); ``(text, kind)`` drives resolution. Per-field URI
         passthrough (like :meth:`resolve_concept`); the rest are resolved
         together so a confident field disambiguates an ambiguous sibling.
-        ``None`` inputs and unresolved fields map to ``None``. Example
-        labels below mimic a historian export's column headers.
+        ``None`` inputs and unresolved fields map to ``None``. URI/URIRef
+        inputs pass through unchanged and also become disambiguating context
+        for the remaining text fields. Example labels below mimic a historian
+        export's column headers.
 
         Example::
 
@@ -331,17 +341,14 @@ def resolve_record_uris(
             # -> {"FIT-101.EU":  "http://qudt.org/vocab/unit/GAL_US-PER-MIN",
             #     "FIT-101.QTY": ".../quantitykind/VolumeFlowRate"}
         """
-        out: dict[str, Optional[str]] = {}
-        to_resolve: dict[str, tuple[str, Optional[str]]] = {}
-        for name, (text, kind) in fields.items():
-            if text is None:
-                out[name] = None
-            elif looks_like_uri(text):
-                out[name] = text
-            else:
-                to_resolve[name] = (text, kind)
+        out, to_resolve, context = split_record_uri_inputs(fields)
         if to_resolve:
-            matches = self.resolve_record(to_resolve, top_k=1, min_score=min_score)
+            matches = self.resolve_record(
+                to_resolve,
+                top_k=1,
+                min_score=min_score,
+                context=context or None,
+            )
             for name in to_resolve:
                 m = matches.get(name) or []
                 out[name] = m[0]["uri"] if m else None
diff --git a/src/acquirium/Server/app.py b/src/acquirium/Server/app.py
index 17f0893..2a6c83e 100644
--- a/src/acquirium/Server/app.py
+++ b/src/acquirium/Server/app.py
@@ -146,6 +146,7 @@ class ResolveRecordRequest(BaseModel):
     fields: list[RecordFieldSpec]
     top_k: int = 5
     min_score: float = 0.5
+    context: list[str] | None = None
 
 
 @asynccontextmanager
@@ -490,7 +491,7 @@ def resolve_text(
 def resolve_record(req: ResolveRecordRequest) -> dict[str, Any]:
     fields = {f.name: (f.text, f.kind) for f in req.fields}
     matches = app.state.manager.resolve_record(
-        fields, top_k=req.top_k, min_score=req.min_score
+        fields, top_k=req.top_k, min_score=req.min_score, context=req.context
     )
     return {"matches": matches}
 
diff --git a/src/acquirium/Server/direct_client.py b/src/acquirium/Server/direct_client.py
index f6d511c..7b3a9be 100644
--- a/src/acquirium/Server/direct_client.py
+++ b/src/acquirium/Server/direct_client.py
@@ -19,9 +19,11 @@
 
 from datetime import datetime
 from typing import TYPE_CHECKING, Any, Optional
+from rdflib import URIRef
 
 from acquirium.Client.acquirium import Acquirium
 from acquirium.Server.insert_stats import insert_stats
+from acquirium.internals.models import split_record_uri_inputs
 
 import warnings
 
@@ -105,8 +107,47 @@ def resolve_text(
         kind: Optional[str] = None,
         top_k: int = 5,
         min_score: float = 0.5,
+        context: Optional[list[str]] = None,
     ) -> list[dict]:
-        return self._manager.resolve_text(text, kind=kind, top_k=top_k, min_score=min_score)
+        return self._manager.resolve_text(
+            text,
+            kind=kind,
+            top_k=top_k,
+            min_score=min_score,
+            context=context,
+        )
+
+    def resolve_record(
+        self,
+        fields: dict[str, tuple[str, Optional[str]]],
+        top_k: int = 5,
+        min_score: float = 0.5,
+        context: Optional[list[str]] = None,
+    ) -> dict[str, list[dict]]:
+        return self._manager.resolve_record(
+            fields,
+            top_k=top_k,
+            min_score=min_score,
+            context=context,
+        )
+
+    def resolve_record_uris(
+        self,
+        fields: dict[str, tuple[Any, Optional[str]]],
+        min_score: float = 0.5,
+    ) -> dict[str, str | URIRef | None]:
+        out, to_resolve, context = split_record_uri_inputs(fields)
+        if to_resolve:
+            matches = self.resolve_record(
+                to_resolve,
+                top_k=1,
+                min_score=min_score,
+                context=context or None,
+            )
+            for name in to_resolve:
+                m = matches.get(name) or []
+                out[name] = m[0]["uri"] if m else None
+        return out
 
 
 class DirectAcquirium(Acquirium):
diff --git a/src/acquirium/Server/manager.py b/src/acquirium/Server/manager.py
index f30fd61..3e983b4 100644
--- a/src/acquirium/Server/manager.py
+++ b/src/acquirium/Server/manager.py
@@ -545,6 +545,7 @@ def resolve_record(
         fields: dict[str, tuple[str, str | None]],
         top_k: int = 5,
         min_score: float = 0.5,
+        context: list[str] | None = None,
     ) -> dict[str, list[dict[str, Any]]]:
         """Jointly resolve a record's fields.
 
@@ -566,7 +567,7 @@ def resolve_record(
             #                      ...}, ...]}
         """
         resolved = self._concept_resolver.resolve_record(
-            fields, top_k=top_k, min_score=min_score
+            fields, top_k=top_k, min_score=min_score, context=context
         )
         return {name: [asdict(r) for r in rs] for name, rs in resolved.items()}
 
diff --git a/src/acquirium/TextMatch/resolver.py b/src/acquirium/TextMatch/resolver.py
index 0642a52..1f9e77a 100644
--- a/src/acquirium/TextMatch/resolver.py
+++ b/src/acquirium/TextMatch/resolver.py
@@ -194,6 +194,7 @@ def resolve_record(
         fields: dict[str, tuple[str, str | None]],
         top_k: int = 5,
         min_score: float = 0.5,
+        context: list[str] | None = None,
     ) -> dict[str, list[ResolveResult]]:
         """Resolve a record's fields *jointly*.
 
@@ -214,8 +215,9 @@ def resolve_record(
              "AIT-330.MED": ("treated water", "class")}
 
         Each field's candidates are gathered independently (cascade
-        early-exit off, so the joint decode sees depth); then for every
-        :data:`RELATIONS` entry
+        early-exit off, so the joint decode sees depth), with optional
+        sibling-URI ``context`` applied as a per-field rerank hint; then
+        for every :data:`RELATIONS` entry
         whose two kinds are both present, the pair is chosen by
         ``argmax a.score + b.score + weight·compat(a, b)``. A side whose top
         candidate is an exact hit is pinned (the compat bonus may not demote
@@ -231,13 +233,14 @@ def resolve_record(
         # early-exit; the rest resolve with the normal cheap settings.
         def _gather(text: str, kind: str | None) -> list[ResolveResult]:
             joint = kind in _RELATION_KINDS
-            return self._ranked_candidates(
+            ranked = self._ranked_candidates(
                 text,
                 kind,
-                max(top_k, self._CONTEXT_FETCH_K) if joint else top_k,
+                max(top_k, self._CONTEXT_FETCH_K) if (joint or context) else top_k,
                 min_score,
-                early_exit=not joint,
+                early_exit=not (joint or context),
             )
+            return self._rerank_by_context(ranked, context) if context else ranked
 
         cands = {
             name: _gather(text, kind) for name, (text, kind) in fields.items()
diff --git a/src/acquirium/internals/models.py b/src/acquirium/internals/models.py
index c3dc941..e7c3bd3 100644
--- a/src/acquirium/internals/models.py
+++ b/src/acquirium/internals/models.py
@@ -4,7 +4,7 @@
 
 import uuid
 from datetime import datetime, timedelta
-from typing import Literal, Any, TYPE_CHECKING
+from typing import Literal, Any, TYPE_CHECKING, Optional
 from dataclasses import dataclass
 from pydantic import BaseModel, ConfigDict, Field, RootModel
 from acquirium.internals.internals_namespaces import ACQUIRIUM_NS
@@ -27,6 +27,50 @@ def looks_like_uri(value: object) -> bool:
     )
 
 
+def split_record_uri_inputs(
+    fields: dict[str, tuple[Any, Optional[str]]],
+) -> tuple[
+    dict[str, str | URIRef | None],
+    dict[str, tuple[str, Optional[str]]],
+    list[str],
+]:
+    """Partition record fields into passthrough outputs, text to resolve, and URI context.
+
+    Input shape matches ``resolve_record_uris``: each key is a caller-chosen
+    label and each value is ``(text, kind)`` where ``kind`` is the resolver
+    role for that field.
+
+    The returned tuple is:
+
+    1. ``out``: fields that already have their final answer and therefore
+       should bypass text resolution. This includes ``None``, plain URI
+       strings, and ``rdflib.URIRef`` objects.
+    2. ``to_resolve``: fields whose first tuple element is plain text and
+       still needs joint resolution.
+    3. ``context``: string URIs derived from the passthrough URI inputs.
+
+    ``context`` is intentionally stringified even for ``URIRef`` inputs because
+    the resolver transport and matching logic operate on raw URI strings.
+    Callers can feed ``context`` into joint record resolution so already-known
+    sibling URIs can help disambiguate the remaining text fields.
+    """
+    out: dict[str, str | URIRef | None] = {}
+    to_resolve: dict[str, tuple[str, Optional[str]]] = {}
+    context: list[str] = []
+    for name, (text, kind) in fields.items():
+        if text is None:
+            out[name] = None
+        elif isinstance(text, URIRef):
+            out[name] = text
+            context.append(str(text))
+        elif looks_like_uri(text):
+            out[name] = text
+            context.append(text)
+        else:
+            to_resolve[name] = (text, kind)
+    return out, to_resolve, context
+
+
 def compute_ref_uri(source_id: str, ref_name: str) -> URIRef:
     """Return a deterministic UUID5 ref URI for a (source_id, ref_name) pair.
 
diff --git a/tests/unit/test_acquirium_stream_registration.py b/tests/unit/test_acquirium_stream_registration.py
index 3195fc9..ba2c55c 100644
--- a/tests/unit/test_acquirium_stream_registration.py
+++ b/tests/unit/test_acquirium_stream_registration.py
@@ -15,6 +15,10 @@
     ACQUIRIUM_VALUE_KIND,
     DATA_SOURCE,
     HAS_EXTERNAL_REFERENCE,
+    HAS_QUANTITY_KIND,
+    HAS_UNIT,
+    HAS_MEDIUM,
+    OF_SUBSTANCE,
     STORED_AT,
     VIRTUAL_POINT,
 )
@@ -67,8 +71,23 @@ def test_register_streams_inserts_one_graph_for_multiple_streams():
 def test_register_stream_without_point_uri_writes_only_ref_node():
     aq = Acquirium.__new__(Acquirium)
     aq.client = MagicMock()
+    aq.client.resolve_record_uris.return_value = {
+        "unit": "http://qudt.org/vocab/unit/PERCENT",
+        "quantity_kind": "http://qudt.org/vocab/quantitykind/DimensionlessRatio",
+        "medium": "urn:nawi-water-ontology#Water",
+        "substance": "urn:nawi-water-ontology#Water",
+    }
 
-    aq.register_streams([{"source_id": "demo-source", "ref_name": "cpu_percent", "value_kind": "numeric"}])
+    aq.register_streams([{
+        "source_id": "demo-source",
+        "ref_name": "cpu_percent",
+        "value_kind": "numeric",
+        "unit": "http://qudt.org/vocab/unit/PERCENT",
+        "quantity_kind": "http://qudt.org/vocab/quantitykind/DimensionlessRatio",
+        "medium": "urn:nawi-water-ontology#Water",
+        "substance": "urn:nawi-water-ontology#Water",
+        "data_source": "driver",
+    }])
 
     aq.client.insert_graph.assert_called_once()
     graph_text = aq.client.insert_graph.call_args[0][0]
@@ -79,6 +98,11 @@ def test_register_stream_without_point_uri_writes_only_ref_node():
     assert (ref_uri, ACQUIRIUM_REF_NAME, Literal("cpu_percent")) in g
     assert (ref_uri, ACQUIRIUM_VALUE_KIND, Literal("numeric")) in g
     assert (ref_uri, STORED_AT, ACQUIRIUM_DB_URI) in g
+    assert (ref_uri, HAS_UNIT, URIRef("http://qudt.org/vocab/unit/PERCENT")) in g
+    assert (ref_uri, HAS_QUANTITY_KIND, URIRef("http://qudt.org/vocab/quantitykind/DimensionlessRatio")) in g
+    assert (ref_uri, HAS_MEDIUM, URIRef("urn:nawi-water-ontology#Water")) in g
+    assert (ref_uri, OF_SUBSTANCE, URIRef("urn:nawi-water-ontology#Water")) in g
+    assert (ref_uri, DATA_SOURCE, Literal("driver")) in g
     assert list(g.subjects(RDF.type, VIRTUAL_POINT)) == []
     assert list(g.subjects(HAS_EXTERNAL_REFERENCE, ref_uri)) == []
 
diff --git a/tests/unit/test_client_http.py b/tests/unit/test_client_http.py
index a2fcb6d..0753345 100644
--- a/tests/unit/test_client_http.py
+++ b/tests/unit/test_client_http.py
@@ -3,8 +3,10 @@
 import pytest
 from unittest.mock import patch, MagicMock
 from datetime import datetime, timezone
+from rdflib import URIRef
 
 from acquirium.Client.client import AcquiriumClient
+from acquirium.Server.direct_client import _DirectClient
 
 
 @pytest.fixture
@@ -108,6 +110,126 @@ def test_with_kind_filter(self, mock_requests, client):
         assert "class" in str(call_kwargs)
 
 
+class TestResolveRecord:
+    @patch("acquirium.Client.client.requests")
+    def test_resolve_record_forwards_context(self, mock_requests, client):
+        mock_resp = MagicMock()
+        mock_resp.json.return_value = {"matches": {"unit": [{"uri": "urn:u"}]}}
+        mock_resp.raise_for_status = MagicMock()
+        mock_requests.post.return_value = mock_resp
+
+        out = client.resolve_record(
+            {"unit": ("kg", "unit")},
+            context=["http://qudt.org/vocab/quantitykind/Mass"],
+        )
+
+        assert out["unit"][0]["uri"] == "urn:u"
+        body = mock_requests.post.call_args.kwargs["json"]
+        assert body["context"] == ["http://qudt.org/vocab/quantitykind/Mass"]
+
+    def test_resolve_record_uris_passes_through_uri_and_uses_it_as_context(self, client):
+        captured = {}
+
+        def fake_resolve_record(fields, top_k=5, min_score=0.5, context=None):
+            captured["fields"] = fields
+            captured["top_k"] = top_k
+            captured["min_score"] = min_score
+            captured["context"] = context
+            return {"unit": [{"uri": "http://qudt.org/vocab/unit/KiloGM"}]}
+
+        client.resolve_record = fake_resolve_record
+
+        out = client.resolve_record_uris({
+            "quantity_kind": ("http://qudt.org/vocab/quantitykind/Mass", "quantity_kind"),
+            "unit": ("kg", "unit"),
+        })
+
+        assert out == {
+            "quantity_kind": "http://qudt.org/vocab/quantitykind/Mass",
+            "unit": "http://qudt.org/vocab/unit/KiloGM",
+        }
+        assert captured["fields"] == {"unit": ("kg", "unit")}
+        assert captured["top_k"] == 1
+        assert captured["min_score"] == 0.5
+        assert captured["context"] == ["http://qudt.org/vocab/quantitykind/Mass"]
+
+    def test_resolve_record_uris_passes_through_uriref_and_uses_it_as_context(self, client):
+        captured = {}
+
+        def fake_resolve_record(fields, top_k=5, min_score=0.5, context=None):
+            captured["fields"] = fields
+            captured["top_k"] = top_k
+            captured["min_score"] = min_score
+            captured["context"] = context
+            return {"unit": [{"uri": "http://qudt.org/vocab/unit/KiloGM"}]}
+
+        client.resolve_record = fake_resolve_record
+        qk = URIRef("http://qudt.org/vocab/quantitykind/Mass")
+
+        out = client.resolve_record_uris({
+            "quantity_kind": (qk, "quantity_kind"),
+            "unit": ("kg", "unit"),
+        })
+
+        assert out == {
+            "quantity_kind": qk,
+            "unit": "http://qudt.org/vocab/unit/KiloGM",
+        }
+        assert captured["fields"] == {"unit": ("kg", "unit")}
+        assert captured["top_k"] == 1
+        assert captured["min_score"] == 0.5
+        assert captured["context"] == ["http://qudt.org/vocab/quantitykind/Mass"]
+
+
+class TestDirectClientResolveRecord:
+    def test_resolve_record_uris_passes_through_uri_and_uses_it_as_context(self):
+        manager = MagicMock()
+        manager.resolve_record.return_value = {
+            "unit": [{"uri": "http://qudt.org/vocab/unit/KiloGM"}]
+        }
+        client = _DirectClient(manager, origin="test")
+
+        out = client.resolve_record_uris({
+            "quantity_kind": ("http://qudt.org/vocab/quantitykind/Mass", "quantity_kind"),
+            "unit": ("kg", "unit"),
+        })
+
+        assert out == {
+            "quantity_kind": "http://qudt.org/vocab/quantitykind/Mass",
+            "unit": "http://qudt.org/vocab/unit/KiloGM",
+        }
+        manager.resolve_record.assert_called_once_with(
+            {"unit": ("kg", "unit")},
+            top_k=1,
+            min_score=0.5,
+            context=["http://qudt.org/vocab/quantitykind/Mass"],
+        )
+
+    def test_resolve_record_uris_passes_through_uriref_and_uses_it_as_context(self):
+        manager = MagicMock()
+        manager.resolve_record.return_value = {
+            "unit": [{"uri": "http://qudt.org/vocab/unit/KiloGM"}]
+        }
+        client = _DirectClient(manager, origin="test")
+        qk = URIRef("http://qudt.org/vocab/quantitykind/Mass")
+
+        out = client.resolve_record_uris({
+            "quantity_kind": (qk, "quantity_kind"),
+            "unit": ("kg", "unit"),
+        })
+
+        assert out == {
+            "quantity_kind": qk,
+            "unit": "http://qudt.org/vocab/unit/KiloGM",
+        }
+        manager.resolve_record.assert_called_once_with(
+            {"unit": ("kg", "unit")},
+            top_k=1,
+            min_score=0.5,
+            context=["http://qudt.org/vocab/quantitykind/Mass"],
+        )
+
+
 # ── register_app / run_app / stop_app / list_app_runs ──────
 
 
diff --git a/tests/unit/test_resolve_point_metadata.py b/tests/unit/test_resolve_point_metadata.py
index d4cd15b..669e598 100644
--- a/tests/unit/test_resolve_point_metadata.py
+++ b/tests/unit/test_resolve_point_metadata.py
@@ -7,6 +7,7 @@
 from __future__ import annotations
 
 from unittest.mock import MagicMock
+from rdflib import URIRef
 
 from acquirium.Client.acquirium import Acquirium
 
@@ -59,6 +60,18 @@ def test_passthrough_and_none_delegated():
                    "quantity_kind": None}
 
 
+def test_uriref_passthrough_and_none_delegated():
+    unit = URIRef("http://qudt.org/vocab/unit/W")
+    aq = _aq(lambda fields, min_score=0.5: {
+        "unit": unit,
+        "quantity_kind": None,
+    })
+    out = aq.resolve_point_metadata(
+        {"unit": unit, "quantity_kind": "??"}
+    )
+    assert out == {"unit": unit, "quantity_kind": None}
+
+
 def test_client_failure_degrades_to_none():
     def boom(fields, min_score=0.5):
         raise RuntimeError("server down")
diff --git a/tests/unit/test_resolve_record.py b/tests/unit/test_resolve_record.py
index c68cb16..562995a 100644
--- a/tests/unit/test_resolve_record.py
+++ b/tests/unit/test_resolve_record.py
@@ -143,3 +143,16 @@ def test_exact_match_is_pinned_against_weak_sibling(self):
         assert out["unit"][0].match_stage == "exact"
         # The uncertain sibling still resolves (against the pinned unit).
         assert out["quantity_kind"][0].uri == "q:DataRate"
+
+
+class TestExternalContext:
+    def test_context_reranks_record_candidates_before_joint_decode(self):
+        r = _resolver(qudt=[
+            _rr("u:KiloGAUSS", "unit", 0.92),
+            _rr("u:KiloGM", "unit", 0.90, related=["q:Mass"]),
+        ])
+        out = r.resolve_record(
+            {"unit": ("kg", "unit")},
+            context=["q:Mass"],
+        )
+        assert out["unit"][0].uri == "u:KiloGM"