A1 Policy Enforcement

An A1 policy is a standing instruction from the Non-RT RIC (the SMO) — "keep this slice's downlink throughput above its guaranteed rate", for example. The Near-RT RIC accepts the policy and hands it to the xApps that can actually carry it out. An A1 enforcer is the piece of your xApp that receives such a policy, watches the RAN to see whether it is being met, acts to keep it met, and reports the enforcement status back.

You write one small class; the SDK owns everything around it — the Redis connection to the RIC's A1 bus, re-subscribing after a reconnect, routing each E2 indication back to you, and shipping status durably.

flowchart LR
  SMO[Non-RT RIC / SMO] -->|A1 policy| P[Near-RT RIC A1-P]
  P -->|distribute over Redis| E[Your A1Enforcer]
  E -->|E2: subscribe / control| RAN[(RAN — E2 nodes)]
  RAN -->|E2: indications| E
  E -->|status over Redis| P
  P -->|A1 status| SMO

Prerequisites

A1 enforcement rides on the same E2 machinery as any other xApp — see Building xApps first. It is Python-only today (Go and C bindings are planned).

The minimal enforcer

Subclass A1Enforcer, declare which policy types you handle, and implement the two policy methods. Pass the enforcer to BaseXApp and run:

from airpuls_ric_sdk.a1 import A1Enforcer, A1Policy, EnforceStatus, EnforceReason
from airpuls_ric_sdk.xapp import BaseXApp, setup_logging


class MyEnforcer(A1Enforcer):
    policy_types = ["QoSTarget_6.0.1"]

    def on_policy_put(self, policy: A1Policy):
        # A policy was created or updated — start observing.
        print(f"enforcing {policy.policy_id}: {policy.body}")

    def on_policy_deleted(self, policy_type_id, policy_id, ref):
        # A policy was withdrawn — stop observing.
        self.ctx.forget(ref)


class MyXApp(BaseXApp):
    XAPP_TYPE = "my-enforcer"
    XAPP_VERSION = "1.0.0"


if __name__ == "__main__":
    setup_logging()
    MyXApp(a1enforcers=[MyEnforcer()]).run_sync()

Passing a1enforcers=[...] is the whole opt-in. You can pass several enforcers (each owning different policy types), and an xApp that also does its own E2 work can still override on_indication — both run.

How a policy reaches you

sequenceDiagram
    participant SMO as Non-RT RIC
    participant RIC as Near-RT RIC (A1-P)
    participant You as A1Enforcer
    participant RAN as RAN (E2)
    SMO->>RIC: PUT policy
    RIC->>You: on_policy_put(policy)
    You->>RAN: ctx.observe → E2 subscribe
    RAN-->>You: on_indication(ref, …)
    You->>RAN: ctx.control_* (optional)
    You->>RIC: ctx.report(ref, status)
    RIC-->>SMO: policy status

The ref on a policy is an opaque handle the RIC assigns. You echo it back in status reports and pass it to ctx.observe / ctx.forget — you never parse it.

Anatomy of an A1Enforcer

Member	When it fires	What to do
policy_types	—	List the policy types you handle, e.g. ["QoSTarget_6.0.1"]. Exactly one enforcer per type.
plugins()	at startup	Return the SM decoders you need (e.g. kpm_plugin_get()).
on_policy_put(policy)	policy created/updated	Required. Set up observation.
on_policy_deleted(type_id, id, ref)	policy withdrawn	Required. Tear down (usually self.ctx.forget(ref)).
on_indication(ref, node, header, data)	E2 REPORT arrives	Evaluate the measurement, decide, report.
on_insert_indication(ref, node, hdr, msg, cpi)	E2 RC INSERT arrives	Answer with a CONTROL echoing cpi.
on_control_result(result)	async CONTROL completes	Consume the outcome.
on_start() / on_stop()	lifecycle	Initialise / clean up your own state.

A1Policy gives you policy.type_id, policy.policy_id, policy.ref, and policy.body (the policy JSON).

The self.ctx handle

self.ctx is bound before on_start and is your only handle to the RIC.

Observe. Declare what you want to watch. The SDK issues the subscription on every present node and re-issues it after any reconnect or newly appearing node — you never track that:

def on_policy_put(self, policy):
    self.ctx.observe(policy.ref, build=self._build_subscription)

def _build_subscription(self, node):
    # Return a subscription builder for this node, or None to skip it.
    builder = KpmSubscriptionBuilder(KpmStyle.STYLE_4)
    builder.set_report_period(1000)
    builder.add_measurement("DRB.UEThpDl")
    return builder

Indications from those subscriptions arrive at on_indication, keyed by the ref you observed. self.ctx.forget(ref) unsubscribes everything for a policy.

Act. Steer the RAN with an RC CONTROL:

self.ctx.control_rc(node, rc_control_builder)                 # synchronous
self.ctx.control_async(node, ran_func_id, header, message, cpi)  # non-blocking

Report. Tell the RIC whether the policy is enforced:

self.ctx.report(ref, EnforceStatus.ENFORCED)
self.ctx.report(ref, EnforceStatus.NOT_ENFORCED, EnforceReason.SCOPE_NOT_APPLICABLE)

A complete example: a QoS-target enforcer

This enforcer keeps a scoped flow's downlink throughput at or above its guaranteed bit rate. It observes with KPM Style 4 and reports the result. (A runnable copy ships as the qos-enforcer reference xApp.)

import logging

from airpuls_ric_sdk import (
    KPM_OID, KpmConditionExpr, KpmStyle, KpmSubscriptionBuilder,
    SmMeasurementType, decode_kpm_ran_function, kpm_plugin_get,
)
from airpuls_ric_sdk.a1 import A1Enforcer, A1Policy, EnforceStatus, EnforceReason
from airpuls_ric_sdk.xapp import BaseXApp, setup_logging

log = logging.getLogger("qos-enforcer")
DL_THROUGHPUT = "DRB.UEThpDl"


class QosTargetEnforcer(A1Enforcer):
    policy_types = ["QoSTarget_6.0.1"]

    def plugins(self):
        return [kpm_plugin_get()]

    def on_start(self):
        self._gfbr = {}     # ref -> guaranteed DL bit rate (bps)
        self._scope = {}    # ref -> policy scope

    def on_policy_put(self, policy: A1Policy):
        gfbr = policy.body.get("qosObjectives", {}).get("gfbr")
        if gfbr is None:
            log.warning("policy %s has no gfbr; ignoring", policy.ref)
            return
        self._gfbr[policy.ref] = int(gfbr)
        self._scope[policy.ref] = policy.body.get("scope", {})
        self.ctx.observe(policy.ref, build=lambda node: self._kpm_for(node, policy.ref))

    def on_indication(self, ref, node, header, data):
        target = self._gfbr.get(ref)
        measured = self._dl_throughput(data, self._scope.get(ref, {}))
        if target is None or measured is None:
            return
        if measured < target:
            # Below target — report NOT_ENFORCED (and optionally steer via RC here).
            self.ctx.report(ref, EnforceStatus.NOT_ENFORCED, EnforceReason.OTHER_REASON)
        else:
            self.ctx.report(ref, EnforceStatus.ENFORCED)

    def on_policy_deleted(self, policy_type_id, policy_id, ref):
        self.ctx.forget(ref)
        self._gfbr.pop(ref, None)
        self._scope.pop(ref, None)

    def _kpm_for(self, node, ref):
        kpm = node.find_ran_function_by_oid(KPM_OID)
        if kpm is None:
            return None                                   # node has no KPM
        desc = decode_kpm_ran_function(kpm.definition_bytes)
        if desc is None or DL_THROUGHPUT not in desc.all_measurements:
            return None
        builder = KpmSubscriptionBuilder(KpmStyle.STYLE_4)
        builder.set_report_period(1000)
        builder.add_measurement(DL_THROUGHPUT)
        slice_id = self._scope.get(ref, {}).get("sliceId")
        if slice_id and "sst" in slice_id:
            builder.add_condition_snssai_sst(KpmConditionExpr.EQUAL, int(slice_id["sst"]))
        return builder

    def _dl_throughput(self, data, scope):
        want_ue = scope.get("ueId", {}).get("value")
        for block in data.blocks:
            if want_ue and (block.ue_id is None or block.ue_id.display != want_ue):
                continue
            for m in block.measurements:
                if m.name != DL_THROUGHPUT:
                    continue
                if m.type == SmMeasurementType.INTEGER:
                    return int(m.integer)
                if m.type == SmMeasurementType.REAL:
                    return int(m.real)
        return None


class QosEnforcerXApp(BaseXApp):
    XAPP_TYPE = "qos-enforcer"
    XAPP_VERSION = "1.0.0"


if __name__ == "__main__":
    setup_logging()
    QosEnforcerXApp(a1enforcers=[QosTargetEnforcer()]).run_sync()

Notice what is absent: no on_e2_node_available, no reconnect handling, no subscription-id bookkeeping, no Redis code, no status stream. That is all the SDK's job — you write only the loop.

Reporting status honestly

A policy is assumed ENFORCED the moment the RIC accepts it, so a successful enforcer need not report anything. Report NOT_ENFORCED when you cannot meet the target, and ENFORCED again when you recover.

When you report NOT_ENFORCED, give a reason:

Reason	Meaning
SCOPE_NOT_APPLICABLE	the policy's scope (UE / slice / cell) does not apply here
STATEMENT_NOT_APPLICABLE	the objective cannot be acted on
OTHER_REASON	anything else (e.g. the target is currently unmet)

A reason passed with ENFORCED is dropped — it would be non-conformant.

Acting on the RAN

To remediate rather than only observe, issue an RC CONTROL from on_indication (see the E2SM-RC guide for building the control):

def on_indication(self, ref, node, header, data):
    if self._below_target(ref, data):
        self.ctx.control_async(node, RC_FUNC_ID, header_bytes, message_bytes)
        self.ctx.report(ref, EnforceStatus.NOT_ENFORCED, EnforceReason.OTHER_REASON)

control_rc is synchronous and re-entrant

ctx.control_rc blocks until the node answers, and while it waits the SDK keeps dispatching — so a new indication can call back into your enforcer before control_rc returns. Keep your handlers re-entrancy-tolerant, or prefer ctx.control_async on a hot path (its result arrives at on_control_result).

Configuration

Add an a1 block to your xapp.yml:

xapp:
  deployment_name: qos-enforcer-1
  a1:
    enabled: true
    url: redis://nrtric-redis:6379/0
    password_env: REDIS_PASSWORD        # password read from env, never inline
    key_namespace: a1                   # must match the RIC's A1 namespace
    reconcile_interval_ms: 30000
    status_stream_maxlen: 100000
    enforcer_id: ""                     # blank → the xApp instance id
ric:
  endpoint: tcp://192.168.3.1:36422

Key	Default	Purpose
xapp.a1.enabled	true	Turn A1 enforcement on or off.
xapp.a1.url	—	Redis URL of the RIC's A1 bus.
xapp.a1.password_env	—	Env var holding the Redis password (secrets stay out of YAML).
xapp.a1.key_namespace	a1	Must match the RIC-side namespace.
xapp.a1.reconcile_interval_ms	30000	How often to poll for changes missed while briefly disconnected.
xapp.a1.status_stream_maxlen	100000	Cap on the status stream length.
xapp.a1.enforcer_id	instance id	Who this enforcer reports as; defaults to the xApp instance id.

Redis down at startup is fine

If the A1 Redis is unreachable when the xApp starts, the xApp still comes up and keeps serving E2; the enforcer reconnects in the background and heals any policy change it missed.

Rules to know

Use run_sync

Enforcers need a live RIC client on the dispatch thread. run_sync() is required; the async run() mode raises if you pass a1enforcers.

• One xApp instance per policy type. The RIC allows exactly one enforcer per policy; all enforcers in one xApp share the xApp's instance id, so run a single instance per type.
• Your callbacks run on one thread (the RIC dispatch thread), so you need no locking — mind only the re-entrancy of a blocking control_rc.
• One failing enforcer can't take down the others; an exception in a callback is logged and contained.

See also

• Building xApps — the xApp lifecycle these enforcers ride on.
• Configuration — the full xapp.yml schema.
• E2SM-KPM and E2SM-RC — building the subscriptions and controls your enforcer uses.