fix(scte35): Condition descriptor-based signals correctly

README.md

@@ -23,15 +23,15 @@ You configure a channel rule that caps long placement opportunities:
 ```text
 IF   segmentationTypeId = 52  (0x34)
 AND  duration > 30
-THEN set breakDuration to 30
+THEN set segmentationDuration to 30
 ```
 
 The POIS returns the conditioned signal in the same HTTP response:
 
 ```text
 Encoder → ESAM SignalProcessingEvent → POIS → rule engine
        → ESAM SignalProcessingNotification (action "replace")
-       → encoder emits SCTE-35 with a 30s break duration
+       → encoder emits SCTE-35 with a 30 second segmentation duration
 ```
 
 The same rule match can also select an alternate input for AWS Elemental Live Virtual Input Switching, or invoke an external action such as a webhook or an AWS Elemental MediaLive schedule update.
@@ -196,17 +196,27 @@ npx cdk deploy --all -c adminEmail=you@example.com
 The `adminEmail` context value provisions the initial admin user: Cognito sends an invitation email with a temporary password to that address (self sign-up is disabled). If you omit it, no user is created and you must create one later with two commands, because dashboard administration requires membership in the `admin` group:
 
 ```bash
+# 1. Create the account. Cognito emails a temporary password; EMAIL is stated
+#    explicitly to match the deployment path, since the pool signs in with
+#    email and stores no phone number.
 aws cognito-idp admin-create-user \
   --user-pool-id <USER_POOL_ID> \
   --username you@example.com \
-  --user-attributes Name=email,Value=you@example.com Name=email_verified,Value=true Name=name,Value=Administrator
+  --user-attributes Name=email,Value=you@example.com Name=email_verified,Value=true Name=name,Value=Administrator \
+  --desired-delivery-mediums EMAIL
 
+# 2. Grant administrator access. The handlers authorize writes from the
+#    "cognito:groups" claim, so an account created by step 1 alone belongs to
+#    no group: it can read channels and logs, but cannot change them, manage
+#    users, or view encoder credentials.
 aws cognito-idp admin-add-user-to-group \
   --user-pool-id <USER_POOL_ID> \
   --username you@example.com \
   --group-name admin
 ```
 
+`<USER_POOL_ID>` is the `UserPoolId` output of the deployment. The account signs in with the temporary password and is then prompted to choose a permanent one.
+
 CDK prompts for approval before creating IAM resources in each stack. To deploy non-interactively (CI or scripted deployments), add `--require-approval never`.
 
 The deployment region follows your AWS CLI configuration. To deploy to a specific region, set `AWS_REGION` (for example, `AWS_REGION=us-west-2 npx cdk deploy --all -c adminEmail=you@example.com`).
@@ -269,7 +279,7 @@ This rule caps placement opportunities longer than 30 seconds:
   ],
   "action": "replace",
   "modifications": [
-    { "target": "breakDuration", "operation": "set", "value": 30 }
+    { "target": "segmentationDuration", "operation": "set", "value": 30 }
   ]
 }
 ```
@@ -284,7 +294,7 @@ Segmentation type `52` is `0x34`, Provider Placement Opportunity Start. Configur
 
 **Modification targets:** `breakDuration` and `segmentationDuration` (in seconds), `ptsAdjustment`, `segmentationTypeId`, `commandType`, `upidType`, `upidValue`, `webDeliveryAllowed`, `noRegionalBlackout`, `archiveAllowed`, `deviceRestrictions`, `addDescriptor`, and `removeDescriptor`.
 
-Choose the duration target that matches the signal you are conditioning: `breakDuration` applies to a splice insert command, and `segmentationDuration` applies to a segmentation descriptor.
+Choose the duration target that matches the signal you are conditioning: `breakDuration` applies to a splice insert command, and `segmentationDuration` applies to a segmentation descriptor. Targeting the wrong one leaves the payload unchanged even though the response still reports `replace`, so verify the conditioned signal for your source format.
 
 ### Descriptor Priority
 

backend/domain/services/rule_evaluator.py

@@ -12,6 +12,7 @@
     SpliceCommandType,
     SpliceInsert,
 )
+
 from domain.models.channel import (
     Rule,
     Condition,
@@ -424,19 +425,23 @@ def _get_event_id(signal: SpliceInfoSection) -> Optional[int]:
 
 
 def _get_duration(signal: SpliceInfoSection) -> Optional[int]:
-    """Get duration from command or descriptor (in seconds)."""
-    # Try break duration from Splice Insert
+    """
+    Get duration from command or descriptor, in seconds.
+
+    The two carriers use different units in this model: BreakDuration.duration
+    holds 90kHz ticks, while SegmentationDescriptor.segmentation_duration holds
+    seconds, as parsed from threefive.
+    """
+    # Break duration from Splice Insert, stored as 90kHz ticks
     if signal.splice_command_type == SpliceCommandType.SPLICE_INSERT:
         if isinstance(signal.splice_command, SpliceInsert):
             if signal.splice_command.break_duration:
-                # Convert from 90kHz ticks to seconds
                 return signal.splice_command.break_duration.duration // 90000
 
-    # Try segmentation duration from descriptor
+    # Segmentation duration from descriptor, already in seconds
     for desc in signal.splice_descriptors:
         if desc.descriptor_tag == 0x02 and desc.segmentation_duration:
-            # Convert from 90kHz ticks to seconds
-            return desc.segmentation_duration // 90000
+            return int(desc.segmentation_duration)
 
     return None
 

backend/domain/services/scte35_encoder.py

@@ -181,3 +181,17 @@ def _apply_modifications_to_cue(cue: threefive.Cue, signal: SpliceInfoSection) -
                     cue_desc.segmentation_type_id = signal_desc.segmentation_type_id
                     cue_desc.segment_num = signal_desc.segment_num
                     cue_desc.segments_expected = signal_desc.segments_expected
+
+                    # SCTE-35 carries sub_segment_num and sub_segments_expected
+                    # for the placement opportunity Start types, and threefive
+                    # refuses to encode a descriptor of those types while they
+                    # are unset. Sources are not required to send them, and
+                    # decoding leaves them as None, so a rule that modified
+                    # such a descriptor used to fail encoding and silently
+                    # return the unmodified signal. Default them to 0, the
+                    # spec's value for "not used".
+                    if cue_desc.segmentation_type_id in (0x34, 0x36, 0x38, 0x3A):
+                        if getattr(cue_desc, "sub_segment_num", None) is None:
+                            cue_desc.sub_segment_num = 0
+                        if getattr(cue_desc, "sub_segments_expected", None) is None:
+                            cue_desc.sub_segments_expected = 0

backend/domain/services/signal_processor.py

@@ -732,21 +732,19 @@ def calculate_break_expiry_time(
     """
     duration_seconds = 0
 
-    # Get duration from Splice Insert command
+    # Get duration from Splice Insert command, stored as 90kHz ticks
     if signal.splice_command_type == SpliceCommandType.SPLICE_INSERT:
         if isinstance(signal.splice_command, SpliceInsert):
             if signal.splice_command.break_duration:
-                # Convert from 90kHz ticks to seconds
                 duration_seconds = (
                     signal.splice_command.break_duration.duration // 90000
                 )
 
-    # Get duration from segmentation descriptor
+    # Get duration from segmentation descriptor, already in seconds
     if duration_seconds == 0 and signal.splice_descriptors:
         for desc in signal.splice_descriptors:
             if desc.descriptor_tag == 0x02 and desc.segmentation_duration:
-                # Convert from 90kHz ticks to seconds
-                duration_seconds = desc.segmentation_duration // 90000
+                duration_seconds = int(desc.segmentation_duration)
                 break
 
     if duration_seconds == 0:

backend/tests/unit/test_descriptor_signal_conditioning.py

@@ -0,0 +1,158 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: MIT-0
+
+"""
+Regression tests for conditioning descriptor-based SCTE-35 signals.
+
+Modern SCTE-35 usually arrives as a time_signal carrying a segmentation
+descriptor rather than a splice_insert, and two defects made that path behave
+incorrectly:
+
+1. ``segmentation_duration`` is stored in seconds, but the rule engine and the
+   stateful break expiry divided it by 90000 as if it were 90kHz ticks, so a
+   60 second break was read as 0 seconds and duration conditions never matched.
+
+2. threefive refuses to encode a segmentation descriptor whose type is a
+   placement opportunity Start (0x34, 0x36, 0x38, 0x3A) while
+   ``sub_segment_num`` is unset. Sources are not required to send that field,
+   so encoding raised and the encoder returned the *unmodified* signal while
+   the response still reported the modification as applied.
+"""
+
+import pytest
+
+from domain.models.channel import Channel
+from domain.services.rule_evaluator import _get_duration
+from domain.services.scte35_encoder import encode_scte35
+from domain.services.scte35_parser import parse_scte35
+from domain.services.signal_processor import calculate_break_expiry_time, process_signal
+
+# time_signal carrying a segmentation descriptor, no sub_segment fields present.
+TIME_SIGNAL_WITH_DESCRIPTOR = (
+    "/DAvAAAAAAAA///wBQb+dGKQoAAZAhdDVUVJSAAAjn+fCAgAAAAALKChijUCAKnMZ1g="
+)
+
+PROVIDER_PLACEMENT_OPPORTUNITY_START = 0x34
+
+
+def _signal_with_duration(seconds: float, type_id: int) -> str:
+    """Build a descriptor-based signal with the given type and duration."""
+    signal = parse_scte35(TIME_SIGNAL_WITH_DESCRIPTOR)
+    for descriptor in signal.splice_descriptors:
+        if descriptor.descriptor_tag == 0x02:
+            descriptor.segmentation_type_id = type_id
+            descriptor.segmentation_duration = seconds
+            descriptor.segmentation_duration_flag = True
+    return encode_scte35(signal, original_base64=TIME_SIGNAL_WITH_DESCRIPTOR)
+
+
+def _segmentation_duration(base64_signal: str) -> float | None:
+    for descriptor in parse_scte35(base64_signal).splice_descriptors:
+        if descriptor.descriptor_tag == 0x02:
+            return descriptor.segmentation_duration
+    return None
+
+
+def _channel(modification_target: str) -> Channel:
+    now = "2026-01-01T00:00:00Z"
+    return Channel(
+        channelId="sports-live-east",
+        name="sports-live-east",
+        enabled=True,
+        statefulMode=False,
+        defaultAction="noop",
+        createdAt=now,
+        updatedAt=now,
+        rules=[
+            {
+                "ruleId": "cap-long-breaks",
+                "name": "Cap breaks longer than 30s",
+                "priority": 1,
+                "enabled": True,
+                "conditions": [
+                    {
+                        "field": "segmentationTypeId",
+                        "operator": "eq",
+                        "value": PROVIDER_PLACEMENT_OPPORTUNITY_START,
+                    },
+                    {"field": "duration", "operator": "gt", "value": 30},
+                ],
+                "action": "replace",
+                "modifications": [
+                    {"target": modification_target, "operation": "set", "value": 30}
+                ],
+            }
+        ],
+    )
+
+
+class TestDescriptorDurationUnits:
+    """segmentation_duration is seconds, not 90kHz ticks."""
+
+    def test_duration_is_read_in_seconds(self):
+        signal = parse_scte35(
+            _signal_with_duration(60.0, PROVIDER_PLACEMENT_OPPORTUNITY_START)
+        )
+
+        assert _get_duration(signal) == 60
+
+    @pytest.mark.parametrize(
+        "seconds,threshold,expected", [(60.0, 30, True), (10.0, 30, False)]
+    )
+    def test_duration_conditions_compare_against_seconds(
+        self, seconds, threshold, expected
+    ):
+        signal = parse_scte35(
+            _signal_with_duration(seconds, PROVIDER_PLACEMENT_OPPORTUNITY_START)
+        )
+
+        assert ((_get_duration(signal) or 0) > threshold) is expected
+
+    def test_break_expiry_is_calculated_for_descriptor_breaks(self):
+        signal = parse_scte35(
+            _signal_with_duration(60.0, PROVIDER_PLACEMENT_OPPORTUNITY_START)
+        )
+
+        assert calculate_break_expiry_time(signal, 0) is not None
+
+
+class TestPlacementOpportunityStartEncoding:
+    """Descriptors of the *Start types encode without sub_segment fields."""
+
+    @pytest.mark.parametrize("type_id", [0x34, 0x36, 0x38, 0x3A])
+    def test_start_types_do_not_silently_return_the_original(self, type_id):
+        encoded = _signal_with_duration(45.0, type_id)
+
+        assert encoded != TIME_SIGNAL_WITH_DESCRIPTOR
+        descriptor = next(
+            d
+            for d in parse_scte35(encoded).splice_descriptors
+            if d.descriptor_tag == 0x02
+        )
+        assert descriptor.segmentation_type_id == type_id
+        assert descriptor.segmentation_duration == pytest.approx(45.0)
+
+
+class TestConditioningDescriptorSignals:
+    """The documented rule caps a long placement opportunity."""
+
+    def test_segmentation_duration_is_capped(self):
+        original = _signal_with_duration(60.0, PROVIDER_PLACEMENT_OPPORTUNITY_START)
+
+        result, _ = process_signal(
+            scte35_binary=original, channel=_channel("segmentationDuration")
+        )
+
+        assert result.action == "replace"
+        assert result.matched_rule_id == "cap-long-breaks"
+        assert _segmentation_duration(result.modified_signal) == pytest.approx(30.0)
+
+    def test_break_duration_target_leaves_a_time_signal_untouched(self):
+        """breakDuration only applies to splice_insert, so the payload is unchanged."""
+        original = _signal_with_duration(60.0, PROVIDER_PLACEMENT_OPPORTUNITY_START)
+
+        result, _ = process_signal(
+            scte35_binary=original, channel=_channel("breakDuration")
+        )
+
+        assert _segmentation_duration(result.modified_signal) == pytest.approx(60.0)