Refine stream registration resolution flow

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

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"

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:

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

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}
 

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):

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()}
 

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()

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.