feat: update tiled node insertion structure

src/blueapi/service/interface.py

@@ -1,3 +1,4 @@
+import json
 import logging
 from collections.abc import Mapping
 from dataclasses import dataclass
@@ -8,7 +9,9 @@
 from bluesky.callbacks.tiled_writer import TiledWriter
 from bluesky_stomp.messaging import StompClient
 from bluesky_stomp.models import Broker, DestinationBase, MessageTopic
+from fastapi import status
 from tiled.client import from_uri
+from tiled.client.utils import ClientError
 
 from blueapi.cli.scratch import get_python_environment
 from blueapi.config import ApplicationConfig, OIDCConfig, ServiceAccount, StompConfig
@@ -25,6 +28,7 @@
     TaskRequest,
     WorkerTask,
 )
+from blueapi.utils import TILED_PROPOSAL_RE
 from blueapi.utils.serialization import access_blob
 from blueapi.worker.event import ProgressEvent, TaskStatusEnum, WorkerEvent, WorkerState
 from blueapi.worker.task import Task
@@ -205,7 +209,43 @@ def begin_task(
                 api_key=tiled_config.authentication,
                 headers=pass_through_headers,
             )
-
+        if task.task_id is not None:
+            task_ = get_task_by_id(task_id=task.task_id)
+            if task_ is not None:
+                task_metadata = task_.task.metadata
+                instrument = active_context.run_engine.md["instrument"]
+                instrument_session = task_metadata["instrument_session"]
+                if not (match := TILED_PROPOSAL_RE.match(instrument_session)):
+                    raise ValueError("Invalid instrument session")
+                proposal = match["proposal"]
+                # Each level's access blob is the prefix of the full
+                # (beamline, proposal, visit) one that access_blob() builds,
+                # matching the beamline/proposal/session tiers the tiled
+                # access policy expects a container to be tagged with.
+                session_blob = json.loads(access_blob(instrument_session, instrument))
+                level_access_tags = [
+                    [json.dumps({"beamline": instrument})],
+                    [json.dumps({"beamline": instrument, "proposal": proposal})],
+                    [json.dumps(session_blob)],
+                ]
+                for key, access_tags in zip(
+                    (instrument, proposal, instrument_session),
+                    level_access_tags,
+                    strict=True,
+                ):
+                    if key not in tiled_client:
+                        try:
+                            tiled_client.create_container(
+                                key=key, access_tags=access_tags
+                            )
+                        except ClientError as e:
+                            if (
+                                e.response.status_code == status.HTTP_409_CONFLICT
+                            ):  # already exists
+                                ...
+                            else:
+                                raise
+                    tiled_client = tiled_client[key]
         tiled_writer_token = active_context.run_engine.subscribe(
             TiledWriter(tiled_client, batch_size=1)
         )
@@ -230,7 +270,7 @@ def remove_callback_when_task_finished(
     if task.task_id is not None:
         try:
             active_worker.begin_task(task.task_id)
-        except:
+        except Exception:
             for channel, token in subscribers:
                 channel.unsubscribe(token)
             raise

src/blueapi/utils/__init__.py

@@ -32,6 +32,8 @@
 Return = TypeVar("Return")
 
 INSTRUMENT_SESSION_RE = re.compile(r"^[a-z]{2}(?P<proposal>\d+)-(?P<visit>\d+)$")
+# Full proposal code (e.g. "cm12345" from "cm12345-1") for building tiled node paths.
+TILED_PROPOSAL_RE = re.compile(r"^(?P<proposal>[a-z]{2}\d+)-\d+$")
 
 
 def report_successful_devices(

src/blueapi/utils/serialization.py

@@ -30,15 +30,18 @@ def serialize(obj: Any) -> Any:
 
 
 def access_blob(instrument_session: str, beamline: str) -> str:
-    m = utils.INSTRUMENT_SESSION_RE.match(instrument_session)
-    if m is None:
+    session_match = utils.INSTRUMENT_SESSION_RE.match(instrument_session)
+    proposal_match = utils.TILED_PROPOSAL_RE.match(instrument_session)
+    if session_match is None or proposal_match is None:
         raise ValueError(
             "Unable to extract proposal and visit from "
             f"instrument session {instrument_session}"
         )
     blob = {
-        "proposal": int(m["proposal"]),
-        "visit": int(m["visit"]),
+        # The full proposal code (e.g. "cm12345"), not just its number - the
+        # tiled access policy strips the letters itself where it needs them.
+        "proposal": proposal_match["proposal"],
+        "visit": int(session_match["visit"]),
         "beamline": beamline,
     }
     return json.dumps(blob)

tests/system_tests/services/opa_config/config.yaml

@@ -5,7 +5,7 @@ services:
 bundles:
   diamond-policies:
     service: ghcr
-    resource: ghcr.io/diamondlightsource/authz-policy:0.0.24
+    resource: ghcr.io/zohebshaikh/authz-policy:0.0.25-alpha
     polling:
       min_delay_seconds: 30
       max_delay_seconds: 120

tests/system_tests/services/tiled_config/dls.py

@@ -2,7 +2,7 @@
 import logging
 
 from fastapi import HTTPException
-from pydantic import BaseModel, HttpUrl, TypeAdapter
+from pydantic import BaseModel, HttpUrl, TypeAdapter, ValidationError
 from starlette.status import (
     HTTP_401_UNAUTHORIZED,
 )
@@ -21,9 +21,35 @@
 
 
 class DiamondAccessBlob(BaseModel):
-    proposal: int
-    visit: int
     beamline: str
+    # The full proposal code, e.g. "cm12345" - tiled.rego strips the leading
+    # letters itself where it needs the bare number.
+    proposal: str | None = None
+    visit: int | None = None
+
+
+# Maps a composite access tag's "key" (as produced by tiled.rego's
+# beamline_tag/proposal_tag/session_tag, e.g.
+# "beamline:i22,proposal:cm111,session:cm111-1") to the corresponding OPA
+# input field. All of these are strings on an existing node's tag - "session"
+# here is the full "cm111-1" instrument session, not the internal numeric
+# session id used by modify_session, so it's kept under its own field name
+# rather than aliased to "visit".
+_TAG_KEY_TO_INPUT_FIELD = {
+    "beamline": "beamline",
+    "proposal": "proposal",
+    "session": "session",
+}
+
+
+def _parse_composite_tag(tag: str) -> dict[str, str]:
+    fields: dict[str, str] = {}
+    for part in tag.split(","):
+        key, _, value = part.partition(":")
+        field = _TAG_KEY_TO_INPUT_FIELD.get(key)
+        if field is not None:
+            fields[field] = value
+    return fields
 
 
 def _check_principal(principal: Principal | None):
@@ -131,9 +157,17 @@ def build_input(
             and "tags" in access_blob
             and len(access_blob["tags"]) > 0
         ):
-            blob = self._type_adapter.validate_json(access_blob["tags"][0])
+            tag = access_blob["tags"][0]
+            try:
+                blob = self._type_adapter.validate_json(tag)
+            except ValidationError:
+                # Not a create-request JSON blob - it's a composite tag
+                # already assigned to an existing node (e.g. tiled checking
+                # scopes on a parent before creating a child).
+                blob = None
+                _input.update(_parse_composite_tag(tag))
             if isinstance(blob, DiamondAccessBlob):
-                _input.update(blob.model_dump())
+                _input.update(blob.model_dump(exclude_none=True))
             elif isinstance(blob, int):
                 _input["session"] = str(blob)
 

tests/system_tests/test_blueapi_system.py

@@ -36,6 +36,7 @@
     TaskResponse,
     WorkerTask,
 )
+from blueapi.utils import TILED_PROPOSAL_RE
 from blueapi.worker.event import (
     TaskResult,
     TaskStatus,
@@ -354,7 +355,7 @@ def test_task_metadata_propagated(
         "user": User.alice,
         "instrument_session": VALID_INSTRUMENT_SESSION[User.alice],
         "tiled_access_tags": [
-            '{"proposal": 12345, "visit": 1, "beamline": "adsim"}',
+            '{"proposal": "cm12345", "visit": 1, "beamline": "adsim"}',
         ],
         "blueapi_task_id": response.task_id,
     }
@@ -612,7 +613,12 @@ def on_event(event: AnyEvent) -> None:
     assert stream_resource["run_start"] == start_doc["uid"]
     assert stream_resource["uri"] == f"file://localhost/tmp/adsim-{scan_id}-det.h5"
 
-    tiled_url = f"http://localhost:8407/api/v1/metadata/{start_doc['uid']}"
+    proposal = TILED_PROPOSAL_RE.match(start_doc["instrument_session"])["proposal"]  # type: ignore
+    tiled_url = (
+        "http://localhost:8407/api/v1/metadata/"
+        f"{start_doc['instrument']}/{proposal}/{start_doc['instrument_session']}/"
+        f"{start_doc['uid']}"
+    )
     response = requests.get(
         tiled_url, headers={"authorization": "Bearer " + get_access_token(user)}
     )