[uss_qualifier/reports] Specify participants associated with resources

AGENTS.md

@@ -18,6 +18,8 @@ This document contains key context, nuances, and troubleshooting tips specifical
 
 
 ## 2. Navigating Data Schemas
+- **Resource base-class imports**: `configurations.configuration` imports resource definitions through the resources package, whose initializer loads the `Resource` base class. Import `ParticipantID` from configuration under `TYPE_CHECKING` with postponed annotations in `resources/resource.py` to avoid a circular import when scenarios are imported first.
+- **Generic method annotations**: With postponed annotations, `typing.get_type_hints(SomeGenericClass.__init__)` cannot resolve class-scoped PEP 695 type parameters automatically (verified on Python 3.13 and 3.14). Use `monitoring.monitorlib.typing.get_method_type_hints` to supply the declaring class's namespace, including for inherited methods and generic return types. Regression tests should exercise resource construction and selector resolution with postponed annotations enabled.
 - **Implicit Types**: Many schema objects inherit from `ImplicitDict`. This means that reading their raw Python class definitions may not reveal all their expected structure. Rely on their `__annotations__` or their OpenAPI documentation.
 - **Optional `ImplicitDict` Fields**: `ImplicitDict` fields annotated with `| None` (or `Optional[...]`) indicate optional dictionary keys. In implicitdict 5.0.0, accessing a missing optional attribute returns `None` without inserting a key; use `"field" in obj` to distinguish absence from an explicitly stored null. Constructor keyword arguments set to `None` normally omit optional values, but class defaults may still populate the key; assigning `obj.field = None` or `obj["field"] = None` stores an explicit null. Use dictionary membership rather than `hasattr` when key presence matters.
 - **implicitdict 5.0.0 Typing**: Its unannotated `__getattribute__` introduces `None` into the inferred type of undeclared attributes (`Unknown | Any | None` in basedpyright 1.40.1), while declared fields and properties retain their types. New optional-access diagnostics may expose a nonexistent member or the wrong schema class rather than a missing optional value. Check the receiver's declaration before adding a None guard or suppressing the diagnostic. An explicit `-> Any` return annotation on the accessor restores the previous permissive fallback without changing runtime behavior, but also hides these pre-existing member-access mistakes.
@@ -26,9 +28,10 @@ This document contains key context, nuances, and troubleshooting tips specifical
 - **Datetime Types**: Fields annotated as `StringBasedDateTime` (or generic `Time` structures wrapping it) inherently expose a `.datetime` property mapping directly to Python's built-in `datetime.datetime` type. Always rely on this `.datetime` attribute for date-math instead of manually parsing string strings.
 - **DSS Base URL Suffixes**: When configuring DSS instances, the base URL for ASTM F3411-22a (NetRID v2) requires the `/rid/v2` suffix path (e.g. `http://dss1.uss1.localutm/rid/v2`), whereas F3411-19 and SCD (F3548) typically use the host base URL directly without that suffix.
 
-## 3. `uss_qualifier` Test Development Rules
+## 3. `uss_qualifier` Development Rules
 - **Markdown Documentation is Mandatory**: Every check (e.g., `self._scenario.check(...)`) added to a Python test scenario must be meticulously documented in the corresponding Markdown documentation file for that specific test scenario or fragment.
 - **Documentation Traceability**: All documented test checks must trace back to exactly one or more requirements using a specific bold format, and feature a severity emoji prefix (e.g., `## 🛑 Correct operational intent details check`). You must refer to `monitoring/uss_qualifier/scenarios/README.md` for specific markup details before modifying test steps.
+- **Resource name shortcuts**: The shortcuts for resource names added to __int__.py (e.g., `from .planning_area import PlanningAreaResource as PlanningAreaResource`) should not be continued. All future development should use only the canonical resource location (e.g., `resources.planning_area.PlanningAreaResource`).
 
 ## 4. Local Testing constraints
 - **Unit-test working directory**: Run `uss_qualifier` unit tests from `monitoring/uss_qualifier/` with the repository root on `PYTHONPATH` (for example, `PYTHONPATH=../.. ../../.venv/bin/python -m pytest scenarios/scenario_test resources/flight_planning`). Some resource fixtures resolve test-data paths relative to the working directory and fail when run from the repository root.

monitoring/monitorlib/typing.py

@@ -1,5 +1,5 @@
 import types
-from typing import TypeVar, Union, get_args, get_origin
+from typing import Any, TypeVar, Union, get_args, get_origin, get_type_hints
 
 TypeName = str
 """Plain string representing a Python class/type name without generic type arguments (e.g., `resources.flight_planning.FlightPlannerResource`)."""
@@ -14,6 +14,20 @@
 """A type annotation as may appear on a constructor/function parameter or generic base class."""
 
 
+def get_method_type_hints(cls: type, method_name: str) -> dict[str, Any]:
+    """Resolve method annotations using the namespace of its declaring class.
+
+    Python 3.13's get_type_hints does not automatically include class-scoped
+    PEP 695 type parameters when evaluating postponed method annotations.
+    Inherited methods need their declaring class's parameters, even when the
+    subclass defines type parameters with the same names.
+    """
+    owner = next(base for base in cls.__mro__ if method_name in vars(base))
+    localns = dict(vars(owner))
+    localns.update((p.__name__, p) for p in getattr(owner, "__type_params__", ()))
+    return get_type_hints(getattr(cls, method_name), localns=localns)
+
+
 def substitute_typevars(
     tp: TypeAnnotation, mapping: dict[TypeVar, ResolvedType]
 ) -> TypeAnnotation:

monitoring/uss_qualifier/reports/report.py

@@ -293,6 +293,9 @@ class TestScenarioReport(ImplicitDict):
     resource_origins: Optional[dict[ResourceID, str]]
     """For each resource used by this test scenario, the place that resource originated."""
 
+    resource_participants: Optional[dict[ResourceID, list[ParticipantID]]]
+    """For each resource used by this test scenario that is associated with one or more participants, the participants associated with that resource."""
+
     notes: Optional[dict[str, Note]]
     """Additional information about this scenario that may be useful"""
 

monitoring/uss_qualifier/reports/sequence_view/events.py

@@ -274,6 +274,7 @@ def compute_tested_scenario(
             if "resource_origins" in report and report.resource_origins is not None
             else {}
         ),
+        resource_participants=report.get("resource_participants") or {},
     )
     action_indexer.index += 1
     return scenario

monitoring/uss_qualifier/reports/sequence_view/summary_types.py

@@ -117,6 +117,7 @@ class TestedScenario:
     participants: dict[ParticipantID, TestedParticipant]
     execution_error: ErrorReport | None
     resource_origins: dict[ResourceID, str]
+    resource_participants: dict[ResourceID, list[ParticipantID]]
 
     @property
     def rows(self) -> int:

monitoring/uss_qualifier/reports/templates/sequence_view/scenario.html

@@ -1,6 +1,15 @@
 <!DOCTYPE html>
 {% from "explorer.html" import explorer_header, explorer_content, explorer_footer %}
 {% from "sequence_view/severity.html" import severity_symbol, severity_class with context %}
+{% macro participant_list(participants) -%}
+  {%- if participants|length == 1 -%}
+    {{ participants[0]|e }}
+  {%- elif participants|length == 2 -%}
+    {{ participants[0]|e }} and {{ participants[1]|e }}
+  {%- else -%}
+    {{ participants[:-1]|join(', ')|e }}, and {{ participants[-1]|e }}
+  {%- endif -%}
+{%- endmacro %}
 <html lang="en">
 <head>
   <meta charset="UTF-8">
@@ -22,7 +31,8 @@ <h3>{{ test_scenario.type }}</h3>
     <div class="node_value">
       <ul id="resource_origin_list">
         {% for local_resource_id, resource_origin in test_scenario.resource_origins.items() %}
-          <li><a href="{{ test_scenario.url }}#{{ local_resource_id }}"><code>{{ local_resource_id }}</code></a> provided by <a href="./index.html#resources-configuration">{{ resource_origin }}</a></li>
+          {% set participants = test_scenario.resource_participants.get(local_resource_id, []) %}
+          <li><a href="{{ test_scenario.url }}#{{ local_resource_id }}"><code>{{ local_resource_id }}</code></a>{% if participants %} for {{ participant_list(participants) }}{% endif %} provided by <a href="./index.html#resources-configuration">{{ resource_origin }}</a></li>
         {% endfor %}
       </ul>
       {% set collapsible.queries = collapsible.queries + ["resource_origin_list"] %}

monitoring/uss_qualifier/resources/astm/f3411/dss.py

@@ -70,6 +70,9 @@ def is_same_as(self, other: DSSInstance) -> bool:
 class DSSInstanceResource(Resource[DSSInstanceSpecification]):
     dss_instance: DSSInstance
 
+    def get_associated_participants(self) -> list[ParticipantID]:
+        return [self.dss_instance.participant_id]
+
     def __init__(
         self,
         specification: DSSInstanceSpecification,
@@ -119,6 +122,9 @@ class DSSInstancesSpecification(ImplicitDict):
 class DSSInstancesResource(Resource[DSSInstancesSpecification]):
     dss_instances: list[DSSInstance]
 
+    def get_associated_participants(self) -> list[ParticipantID]:
+        return list(dict.fromkeys(d.participant_id for d in self.dss_instances))
+
     def __init__(
         self,
         specification: DSSInstancesSpecification,

monitoring/uss_qualifier/resources/astm/f3548/v21/dss.py

@@ -51,6 +51,7 @@
 from monitoring.monitorlib.inspection import calling_function_name, fullname
 from monitoring.monitorlib.mutate import scd as mutate
 from monitoring.monitorlib.mutate.scd import MutatedSubscription
+from monitoring.uss_qualifier.configurations.configuration import ParticipantID
 from monitoring.uss_qualifier.resources.astm.dss import NotificationIndexImplementation
 from monitoring.uss_qualifier.resources.communications import AuthAdapterResource
 from monitoring.uss_qualifier.resources.plural import PluralResource
@@ -75,7 +76,7 @@ class DSSInstanceSpecification(ImplicitDict):
 
     notification_index_implementation: Optional[NotificationIndexImplementation]
     """Style of implementation this instance uses for notification index.
-    
+
     If not specified, TimeBased is assumed."""
 
     def __init__(self, *args, **kwargs):
@@ -713,6 +714,9 @@ class DSSInstanceResource(Resource[DSSInstanceSpecification]):
     _auth_adapter: AuthAdapterResource
     _client: UTMClientSession
 
+    def get_associated_participants(self) -> list[ParticipantID]:
+        return [self.participant_id]
+
     def __init__(
         self,
         specification: DSSInstanceSpecification,

monitoring/uss_qualifier/resources/communications/auth_adapter.py

@@ -36,6 +36,9 @@ class AuthAdapterResource(Resource[AuthAdapterSpecification]):
     scopes: set[str]
     participant_id: ParticipantID | None = None
 
+    def get_associated_participants(self) -> list[ParticipantID]:
+        return [self.participant_id] if self.participant_id else []
+
     def __init__(self, specification: AuthAdapterSpecification, resource_origin: str):
         super().__init__(specification, resource_origin)
         if (

monitoring/uss_qualifier/resources/flight_planning/flight_planners.py

@@ -29,6 +29,9 @@ class FlightPlannerResource(Resource[FlightPlannerSpecification]):
     client: FlightPlannerClient
     participant_id: ParticipantID
 
+    def get_associated_participants(self) -> list[ParticipantID]:
+        return [self.participant_id]
+
     def __init__(
         self,
         specification: FlightPlannerSpecification,

monitoring/uss_qualifier/resources/geospatial_info/geospatial_info_providers.py

@@ -54,6 +54,9 @@ class GeospatialInfoProviderResource(Resource[GeospatialInfoProviderSpecificatio
     client: GeospatialInfoClient
     participant_id: ParticipantID
 
+    def get_associated_participants(self) -> list[ParticipantID]:
+        return [self.participant_id]
+
     def __init__(
         self,
         specification: GeospatialInfoProviderSpecification,

monitoring/uss_qualifier/resources/interuss/datastore/datastore.py

@@ -7,6 +7,7 @@
 import psycopg
 from implicitdict import ImplicitDict
 
+from monitoring.uss_qualifier.configurations.configuration import ParticipantID
 from monitoring.uss_qualifier.resources.resource import Resource
 
 
@@ -30,6 +31,9 @@ def __init__(self, *args, **kwargs):
 class DatastoreDBNodeResource(Resource[DatastoreDBNodeSpecification]):
     _specification: DatastoreDBNodeSpecification
 
+    def get_associated_participants(self) -> list[ParticipantID]:
+        return [self.participant_id]
+
     def __init__(
         self,
         specification: DatastoreDBNodeSpecification,
@@ -66,6 +70,9 @@ class DatastoreDBClusterSpecification(ImplicitDict):
 class DatastoreDBClusterResource(Resource[DatastoreDBClusterSpecification]):
     nodes: list[DatastoreDBNodeResource]
 
+    def get_associated_participants(self) -> list[ParticipantID]:
+        return list(dict.fromkeys(n.participant_id for n in self.nodes))
+
     def __init__(
         self,
         specification: DatastoreDBClusterSpecification,

monitoring/uss_qualifier/resources/interuss/mock_uss/client.py

@@ -170,6 +170,9 @@ class MockUSSSpecification(ImplicitDict):
 class MockUSSResource(Resource[MockUSSSpecification]):
     mock_uss: MockUSSClient
 
+    def get_associated_participants(self) -> list[ParticipantID]:
+        return [self.mock_uss.participant_id]
+
     def __init__(
         self,
         specification: MockUSSSpecification,
@@ -192,6 +195,9 @@ class MockUSSsSpecification(ImplicitDict):
 class MockUSSsResource(Resource[MockUSSsSpecification]):
     mock_uss_instances: list[MockUSSClient]
 
+    def get_associated_participants(self) -> list[ParticipantID]:
+        return list(dict.fromkeys(m.participant_id for m in self.mock_uss_instances))
+
     def __init__(
         self,
         specification: MockUSSsSpecification,

monitoring/uss_qualifier/resources/interuss/uss_identification.py

@@ -48,6 +48,9 @@ def __init__(
         super().__init__(specification, resource_origin)
         self.identifiers = specification.uss_identifiers or {}
 
+    def get_associated_participants(self) -> list[ParticipantID]:
+        return list(self.identifiers)
+
     def attribute_query_server(self, query: Query) -> None:
         """Identify the participant ID of the server responding to this query and mutate `query` accordingly, if possible"""
         claims = query.request.token

monitoring/uss_qualifier/resources/netrid/observers.py

@@ -8,6 +8,7 @@
 
 from monitoring.monitorlib import fetch, infrastructure
 from monitoring.monitorlib.fetch import QueryType
+from monitoring.uss_qualifier.configurations.configuration import ParticipantID
 from monitoring.uss_qualifier.resources.communications import AuthAdapterResource
 from monitoring.uss_qualifier.resources.resource import Resource
 
@@ -96,6 +97,9 @@ class NetRIDObserversSpecification(ImplicitDict):
 class NetRIDObserversResource(Resource[NetRIDObserversSpecification]):
     observers: list[RIDSystemObserver]
 
+    def get_associated_participants(self) -> list[ParticipantID]:
+        return list(dict.fromkeys(o.participant_id for o in self.observers))
+
     def __init__(
         self,
         specification: NetRIDObserversSpecification,

monitoring/uss_qualifier/resources/netrid/service_providers.py

@@ -12,6 +12,7 @@
     SCOPE_RID_QUALIFIER_INJECT,
     CreateTestParameters,
 )
+from monitoring.uss_qualifier.configurations.configuration import ParticipantID
 from monitoring.uss_qualifier.resources.communications import AuthAdapterResource
 from monitoring.uss_qualifier.resources.resource import Resource
 
@@ -109,6 +110,9 @@ def get_user_notifications(
 class NetRIDServiceProviders(Resource[NetRIDServiceProvidersSpecification]):
     service_providers: list[NetRIDServiceProvider]
 
+    def get_associated_participants(self) -> list[ParticipantID]:
+        return list(dict.fromkeys(s.participant_id for s in self.service_providers))
+
     def __init__(
         self,
         specification: NetRIDServiceProvidersSpecification,

monitoring/uss_qualifier/resources/plural.py

@@ -1,17 +1,25 @@
-from abc import ABC, abstractmethod
-
-from monitoring.uss_qualifier.resources.resource import Resource
-
-
-class PluralResource[TSingularResource: Resource](Resource, ABC):
-    """Resource that provides multiple instances of a singular resources."""
-
-    @abstractmethod
-    def get_resource_instances_count(self) -> int:
-        """Get the number of singular resources this plural resource contains."""
-        raise NotImplementedError()
-
-    @abstractmethod
-    def get_resource_instance(self, index: int) -> TSingularResource:
-        """Get the specific singular resource at the specified index contained by this plural resource."""
-        raise NotImplementedError()
+from abc import ABC, abstractmethod
+
+from monitoring.uss_qualifier.configurations.configuration import ParticipantID
+from monitoring.uss_qualifier.resources.resource import Resource
+
+
+class PluralResource[TSingularResource: Resource](Resource, ABC):
+    """Resource that provides multiple instances of a singular resources."""
+
+    def get_associated_participants(self) -> list[ParticipantID]:
+        participants = []
+        for i in range(self.get_resource_instances_count()):
+            resource = self.get_resource_instance(i)
+            participants.extend(resource.get_associated_participants())
+        return list(dict.fromkeys(participants))
+
+    @abstractmethod
+    def get_resource_instances_count(self) -> int:
+        """Get the number of singular resources this plural resource contains."""
+        raise NotImplementedError()
+
+    @abstractmethod
+    def get_resource_instance(self, index: int) -> TSingularResource:
+        """Get the specific singular resource at the specified index contained by this plural resource."""
+        raise NotImplementedError()

monitoring/uss_qualifier/resources/resource.py

@@ -1,7 +1,9 @@
+from __future__ import annotations
+
 import inspect
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
-from typing import TypeVar, get_origin, get_type_hints
+from typing import TYPE_CHECKING, TypeVar, get_origin
 
 from implicitdict import ImplicitDict
 from loguru import logger
@@ -14,6 +16,7 @@
     TypeAnnotation,
     collect_typevars,
     format_type_name,
+    get_method_type_hints,
     infer_typevars_from_arg,
     is_type_covariant_match,
     substitute_typevars,
@@ -25,6 +28,9 @@
     ResourceTypeName,
 )
 
+if TYPE_CHECKING:
+    from monitoring.uss_qualifier.configurations.configuration import ParticipantID
+
 SpecificationType = TypeVar("SpecificationType", bound=ImplicitDict)
 
 
@@ -65,7 +71,7 @@ def _get_typevar_map(self) -> dict[TypeVar, ResolvedType]:
         params: tuple[TypeVar, ...] = getattr(self.__class__, "__parameters__", ())
         if params:
             try:
-                hints = get_type_hints(self.__class__.__init__)
+                hints = get_method_type_hints(self.__class__, "__init__")
                 for arg_name, hint in hints.items():
                     if (
                         isinstance(hint, TypeVar)
@@ -108,7 +114,11 @@ def is_or_inherits(self, resource_type: ResourceTypeName) -> bool:
             self._get_type_annotation(), specified_type, self._get_typevar_map()
         )
 
-    def resolve_resource(self) -> "Resource":
+    def get_associated_participants(self) -> list[ParticipantID]:
+        """Return distinct participant IDs associated with this resource, if any."""
+        return []
+
+    def resolve_resource(self) -> Resource:
         """Return the resource to expose in the resource pool after construction.
 
         Selectors may return another resource.  Its type metadata and origin are
@@ -176,7 +186,7 @@ def __init__(
     def resolve_resource(self) -> SingularResourceType:
         resource = self._provider.provide_resource_for(index=self._specification.index)
         expected_type = substitute_typevars(
-            get_type_hints(SingularResource.resolve_resource)["return"],
+            get_method_type_hints(SingularResource, "resolve_resource")["return"],
             self._get_typevar_map(),
         )
         if not isinstance(resource, Resource) or not is_type_covariant_match(
@@ -329,7 +339,7 @@ def get_resource_types(
 
     collect_typevars(resource_type, typevar_map)
 
-    raw_signature = get_type_hints(resource_type.__init__)
+    raw_signature = get_method_type_hints(resource_type, "__init__")
     constructor_signature: dict[str, TypeAnnotation] = {
         name: substitute_typevars(t, typevar_map) for name, t in raw_signature.items()
     }