Exclude PM room types from shared daily report processing

tests/javascript/arr_data_review.cjs

@@ -19,6 +19,17 @@ test('cancelled source rows need no fields and the exclusion stays visible befor
   assert.equal(h.calls.every(call=>call.method==='GET'),true);
 });
 
+test('PM exclusions stay visible without requiring field completion',async()=>{
+  const h=harness(()=>response(200,{...review([]),excluded_cancelled_count:6,excluded_pm_count:19}));
+  await h.acceptARRDownloadTask(task,{sync:false});
+  const progress=h.element('#arr-data-review-progress').textContent;
+  assert.match(progress,/data_review\.excluded_cancelled/);
+  assert.match(progress,/data_review\.excluded_pm/);
+  assert.equal(h.element('#arr-data-review-body').innerHTML,'');
+  assert.equal(h.element('#arr-data-review-finalize').disabled,false);
+  assert.equal(h.calls.every(call=>call.method==='GET'),true);
+});
+
 test('removed cancelled items clear obsolete drafts but retain edits still in the review',async()=>{
   let current=review([item(),item({item_id:'2:DISP_ROOM_NO',source_sequence:2,field:'DISP_ROOM_NO',can_be_empty:false})]);
   const h=harness(()=>response(200,current));

tests/local_postgres.py

@@ -76,8 +76,8 @@ class TemporaryPostgres:
             connection.close()
             raise
 
-    def reset_database(self, *, schema_version=20):
-        if schema_version not in (19, 20):
+    def reset_database(self, *, schema_version=21):
+        if schema_version not in (19, 20, 21):
             raise ValueError("unsupported fixture schema version")
         # Only this owned cluster has passed data_directory/socket checks above.
         with self.connect(database="postgres", autocommit=True) as connection:

tests/test_arr_cancelled_ingestion.py

@@ -59,6 +59,7 @@ class CancelledIngestionContractTests(unittest.TestCase):
             old_envelope = replace(envelope, processor_version="4.2.0")
             old_payload = copy.deepcopy(payload)
             old_payload["processor_version"] = "4.2.0"
+            old_payload["outcome_counts"].pop("excluded_pm", None)
             with self.assertRaisesRegex(IngestionError, "outcome counts contract"):
                 _validate_structured_payload_v4(old_payload, old_envelope)
             del old_payload["outcome_counts"]["excluded_cancelled"]
@@ -69,6 +70,7 @@ class CancelledIngestionContractTests(unittest.TestCase):
             raw, store, envelope, payload = self.generated(Path(temporary))
             old = copy.deepcopy(payload)
             old["processor_version"] = "4.2.0"
+            old["outcome_counts"].pop("excluded_pm", None)
             del old["outcome_counts"]["excluded_cancelled"]
             with self.assertRaisesRegex(IngestionError, "structured outcome is invalid"):
                 _validate_structured_payload_v4(old, replace(envelope, processor_version="4.2.0"))
@@ -81,9 +83,9 @@ class CancelledIngestionContractTests(unittest.TestCase):
     def test_finance_counts_preserve_old_payload_and_add_new_count(self):
         counts = dict(retained=2, excluded_rate_code=1, duplicate=3,
                       validation_failed=4, price_unmatched=5)
-        self.assertEqual(_outcome_counts({"outcome_counts": counts}), (2, 1, 3, 4, 5, 0))
+        self.assertEqual(_outcome_counts({"outcome_counts": counts}), (2, 1, 3, 4, 5, 0, 0))
         counts["excluded_cancelled"] = 6
-        self.assertEqual(_outcome_counts({"outcome_counts": counts}), (2, 1, 3, 4, 5, 6))
+        self.assertEqual(_outcome_counts({"outcome_counts": counts}), (2, 1, 3, 4, 5, 6, 0))
 
     def test_cancelled_count_balances_pending_price_review(self):
         with tempfile.TemporaryDirectory() as temporary:
@@ -152,17 +154,17 @@ class CancelledFinanceMigrationTests(unittest.TestCase):
             connection.execute("RESET ROLE")
 
     def test_empty_database_allows_guarded_rollback_and_reapply(self):
-        self.database.reset_database()
+        self.database.reset_database(schema_version=20)
         repository = PostgresIngestionRepository(DatabaseConfig("owned-fixture"), connect=self.database.connect)
-        repository.assert_data_source_schema()
+        repository.assert_cancelled_scope_schema()
         with self.database.connect(autocommit=True) as connection:
             connection.execute((PROJECT / "database/020_daily_cancelled_exclusion.down.sql").read_text(), prepare=False)
         with self.assertRaises(IngestionError) as missing:
-            repository.assert_data_source_schema()
+            repository.assert_cancelled_scope_schema()
         self.assertEqual(missing.exception.code, "DATABASE_MIGRATION_MISSING")
         with self.database.connect(autocommit=True) as connection:
             connection.execute((PROJECT / "database/020_daily_cancelled_exclusion.sql").read_text(), prepare=False)
-        repository.assert_data_source_schema()
+        repository.assert_cancelled_scope_schema()
 
 
 if __name__ == "__main__":

tests/test_arr_cancelled_scope.py

@@ -157,7 +157,7 @@ class CancelledScopeTests(unittest.TestCase):
         self.assertEqual((payload["source_rows"], payload["removed_by_rate_code"], payload["output_rows"]), (2, 0, 1))
         self.assertEqual(payload["outcome_counts"]["excluded_cancelled"], 1)
         self.assertEqual(set(payload["outcome_counts"]), core.FINAL_OUTCOMES)
-        self.assertEqual(payload["processor_version"], "4.3.0")
+        self.assertEqual(payload["processor_version"], "4.4.0")
         self.assertEqual([row["outcome"] for row in payload["records"]], ["excluded_cancelled", "retained"])
         self.assertEqual(payload["records"][0]["decision_codes"], ["RESERVATION_CANCELLED_EXCLUDED"])
         self.assertEqual(payload["records"][0]["confirmation_no"], "SYNTHETIC-CONF-1")

tests/test_arr_download_runtime.py

@@ -41,6 +41,7 @@ class RuntimeTests(unittest.TestCase):
         self.store = ManagedObjectStore(FilesystemObjectBackend(self.root / 'objects', create=True))
         self.repository = InMemoryIngestionRepository()
         self.repository.assert_cancelled_scope_schema = Mock()
+        self.repository.assert_pm_scope_schema = Mock()
         self.processor = CountingProcessor(self.policy)
         self.client = Mock()
         with patch.object(processing_runtime, 'compose_object_store', return_value=
@@ -59,6 +60,21 @@ class RuntimeTests(unittest.TestCase):
 
     def test_processing_composition_checks_finance_cancellation_schema(self):
         self.repository.assert_cancelled_scope_schema.assert_called_once_with()
+        self.repository.assert_pm_scope_schema.assert_called_once_with()
+
+    def test_missing_pm_migration_closes_storage_and_disables_composition(self):
+        from arr_ingestion.contracts import IngestionError
+        self.repository.assert_pm_scope_schema.side_effect = IngestionError(
+            "DATABASE_MIGRATION_MISSING", "ARR processing requires migration 021")
+        client = Mock()
+        with patch.object(processing_runtime, 'compose_object_store', return_value=
+                          processing_runtime.ObjectStoreRuntime(client, self.store)), \
+             patch.object(processing_runtime, 'PostgresIngestionRepository', return_value=self.repository), \
+             self.assertRaises(IngestionError) as error:
+            processing_runtime.compose_programmatic_processing(
+                project_root=Path(__file__).resolve().parents[1], connect=Mock())
+        self.assertEqual(error.exception.code, "DATABASE_MIGRATION_MISSING")
+        client.close.assert_called_once_with()
 
     def test_missing_cancellation_migration_closes_storage_and_disables_composition(self):
         from arr_ingestion.contracts import IngestionError

tests/test_arr_ingestion_postgres.py

@@ -533,7 +533,7 @@ class PostgresIngestionTests(unittest.TestCase):
             and "INSERT INTO finance.daily_versions" in value
         )
 
-        self.assertEqual(version_insert.count("%s"), 21)
+        self.assertEqual(version_insert.count("%s"), 22)
 
 
 class Migration008ContractTests(unittest.TestCase):

tests/test_arr_opera_daily_ingest.py

@@ -236,7 +236,7 @@ class ArrOperaDailyIngestTests(unittest.TestCase):
             self.assertNotIn(forbidden, scripts)
         self.assertEqual(core.RESULT_VERSION, "4.0")
         self.assertEqual(core.STRUCTURED_RESULT_SCHEMA_VERSION, "4.0")
-        self.assertEqual(core.PROCESSOR_VERSION, "4.3.0")
+        self.assertEqual(core.PROCESSOR_VERSION, "4.4.0")
         self.assertEqual(len(core.DAILY_HEADERS), 19)
         self.assertEqual(len(core.RATE_WHITELIST), 20)
 
@@ -302,7 +302,7 @@ class ArrOperaDailyIngestTests(unittest.TestCase):
             )
 
             self.assertEqual(payload["result_schema_version"], "4.0")
-            self.assertEqual(payload["processor_version"], "4.3.0")
+            self.assertEqual(payload["processor_version"], "4.4.0")
             self.assertEqual(payload["source_rows"], 5)
             self.assertEqual(
                 payload["outcome_counts"],
@@ -310,6 +310,7 @@ class ArrOperaDailyIngestTests(unittest.TestCase):
                     "candidate": 0,
                     "duplicate": 1,
                     "excluded_cancelled": 0,
+                    "excluded_pm": 0,
                     "excluded_rate_code": 1,
                     "price_unmatched": 0,
                     "retained": 3,
@@ -802,7 +803,7 @@ class ArrOperaDailyIngestTests(unittest.TestCase):
                     },
                 }
             )
-            self.assertEqual(delivery.processor_version, "4.3.0")
+            self.assertEqual(delivery.processor_version, "4.4.0")
             self.assertEqual(delivery.result_schema_version, "4.0")
             self.assertEqual(delivery.business_date, date(2026, 7, 27))
 

tests/test_arr_pm_ingestion.py

@@ -0,0 +1,315 @@
+"""Approved PM exclusion contracts and additive Finance upgrade; all inputs synthetic."""
+from __future__ import annotations
+
+import copy
+from dataclasses import replace
+import json
+import os
+from pathlib import Path
+import tempfile
+import unittest
+
+from openpyxl import load_workbook
+
+from arr_ingestion.contracts import DeliveryEnvelope, IngestionError
+from arr_ingestion.postgres import DatabaseConfig, PostgresIngestionRepository, _outcome_counts
+from arr_ingestion.validation import DeliveryValidator, _validate_structured_payload_v4
+from arr_processing.policy import load_processor_policy
+from tests.local_postgres import TemporaryPostgres
+from tests import test_arr_direct_data as direct_helpers
+from tests.test_arr_ingestion_validation import build_delivery, policy
+from tests.test_arr_opera_daily_ingest import reservation, xml_document
+
+
+PROJECT = Path(__file__).resolve().parents[1]
+
+
+def pm_row(*, status=None):
+    value = reservation(1, rate_code="", rate_amount="0", company="").replace(
+        "<ROOM_CATEGORY_LABEL>SYNTHETIC ROOM TYPE</ROOM_CATEGORY_LABEL>",
+        "<ROOM_CATEGORY_LABEL> pm </ROOM_CATEGORY_LABEL>",
+    ).replace("<DISP_ROOM_NO>SYNTHETIC-ROOM-1</DISP_ROOM_NO>", "<DISP_ROOM_NO/>")
+    if status:
+        value = value.replace("</G_RESERVATION>", f"<RESV_STATUS>{status}</RESV_STATUS></G_RESERVATION>")
+    return value
+
+
+class PMIngestionContractTests(unittest.TestCase):
+    def generated(self, root, *rows):
+        raw, store, _ = build_delivery(xml_document(*rows), root)
+        envelope = DeliveryEnvelope.from_dict(json.loads(raw))
+        payload = json.loads(store.objects[envelope.artifacts["structured_result_json"].object_key])
+        return raw, store, envelope, payload
+
+    def test_pm_missing_required_fields_is_audited_before_validation_and_pricing(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            raw, store, _, payload = self.generated(Path(temporary), pm_row(), reservation(2))
+            verified = DeliveryValidator(store, policy()).validate(raw)
+            self.assertEqual(verified.envelope.status, "success")
+            self.assertEqual(payload["outcome_counts"]["excluded_pm"], 1)
+            self.assertEqual(payload["output_rows"], 1)
+            row = payload["records"][0]
+            self.assertEqual(row["outcome"], "excluded_pm")
+            self.assertEqual(row["decision_codes"], ["ROOM_TYPE_PM_EXCLUDED"])
+            for field in ("real_price", "total_price", "kb_amount", "channel_key", "pricing_method"):
+                self.assertIsNone(row[field])
+
+    def test_pm_count_is_required_only_for_44_and_older_versions_stay_readable(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            _, _, envelope, payload = self.generated(Path(temporary), reservation(1))
+            self.assertEqual(payload["processor_version"], "4.4.0")
+            self.assertEqual(payload["outcome_counts"]["excluded_pm"], 0)
+            for field in ("excluded_pm", "excluded_cancelled"):
+                missing = copy.deepcopy(payload)
+                del missing["outcome_counts"][field]
+                with self.assertRaisesRegex(IngestionError, "outcome counts contract"):
+                    _validate_structured_payload_v4(missing, envelope)
+            for version in ("4.3.0", "4.2.0"):
+                older = copy.deepcopy(payload)
+                older["processor_version"] = version
+                with self.assertRaisesRegex(IngestionError, "outcome counts contract"):
+                    _validate_structured_payload_v4(older, replace(envelope, processor_version=version))
+                del older["outcome_counts"]["excluded_pm"]
+                if version == "4.2.0":
+                    del older["outcome_counts"]["excluded_cancelled"]
+                _validate_structured_payload_v4(older, replace(envelope, processor_version=version))
+
+    def test_pm_record_cannot_be_relabelled_or_given_price_facts(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            _, _, envelope, payload = self.generated(Path(temporary), pm_row(), reservation(2))
+            changes = {
+                "room_category_label": "PM-SUITE",
+                "decision_codes": ["RATE_CODE_EXCLUDED"],
+                "real_price": 0,
+                "total_price": 0,
+                "kb_amount": 0,
+                "channel_key": "Group",
+                "pricing_method": "zero_price_exception",
+            }
+            for field, value in changes.items():
+                changed = copy.deepcopy(payload)
+                changed["records"][0][field] = value
+                with self.subTest(field=field), self.assertRaisesRegex(IngestionError, "PM exclusion"):
+                    _validate_structured_payload_v4(changed, envelope)
+            older = copy.deepcopy(payload)
+            older["processor_version"] = "4.3.0"
+            del older["outcome_counts"]["excluded_pm"]
+            with self.assertRaisesRegex(IngestionError, "structured outcome is invalid"):
+                _validate_structured_payload_v4(older, replace(envelope, processor_version="4.3.0"))
+            changed = copy.deepcopy(payload)
+            changed["outcome_counts"]["excluded_pm"] = 0
+            with self.assertRaisesRegex(IngestionError, "outcomes do not reconcile"):
+                _validate_structured_payload_v4(changed, envelope)
+
+    def test_pm_balances_price_review_and_failure_and_cancelled_takes_precedence(self):
+        cases = (
+            (reservation(2, rate_amount="1800"), "review_required"),
+            (reservation(2, departure="2026-07-26"), "failed"),
+        )
+        for row, status in cases:
+            with self.subTest(status=status), tempfile.TemporaryDirectory() as temporary:
+                raw, store, _, payload = self.generated(Path(temporary), pm_row(), row)
+                self.assertEqual(DeliveryValidator(store, policy()).validate(raw).envelope.status, status)
+                self.assertEqual(payload["outcome_counts"]["excluded_pm"], 1)
+        with tempfile.TemporaryDirectory() as temporary:
+            raw, store, _, payload = self.generated(Path(temporary), pm_row(status="CXL"), reservation(2))
+            DeliveryValidator(store, policy()).validate(raw)
+            self.assertEqual(payload["outcome_counts"]["excluded_cancelled"], 1)
+            self.assertEqual(payload["outcome_counts"]["excluded_pm"], 0)
+
+    def test_finance_defaults_only_old_pm_counts_and_rejects_44_missing_keys(self):
+        counts = dict(retained=2, excluded_rate_code=1, duplicate=3,
+                      validation_failed=4, price_unmatched=5)
+        self.assertEqual(_outcome_counts({"outcome_counts": counts}), (2, 1, 3, 4, 5, 0, 0))
+        counts["excluded_cancelled"] = 6
+        self.assertEqual(_outcome_counts({"processor_version": "4.3.0", "outcome_counts": counts}),
+                         (2, 1, 3, 4, 5, 6, 0))
+        with self.assertRaises(IngestionError):
+            _outcome_counts({"processor_version": "4.4.0", "outcome_counts": counts})
+        counts["excluded_pm"] = 7
+        self.assertEqual(_outcome_counts({"processor_version": "4.4.0", "outcome_counts": counts}),
+                         (2, 1, 3, 4, 5, 6, 7))
+        del counts["excluded_cancelled"]
+        with self.assertRaises(IngestionError):
+            _outcome_counts({"processor_version": "4.4.0", "outcome_counts": counts})
+
+
+@unittest.skipUnless(os.environ.get("ARR_TEST_LOCAL_POSTGRES") == "1", "owned PostgreSQL opt-in required")
+class PMFinanceMigrationTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.database = TemporaryPostgres().__enter__()
+        cls.addClassCleanup(cls.database.__exit__, None, None, None)
+
+    def test_additive_upgrade_preserves_cancelled_history_and_role_privileges(self):
+        import psycopg
+        self.database.reset_database(schema_version=20)
+        repository = PostgresIngestionRepository(DatabaseConfig("owned-fixture"), connect=self.database.connect)
+        repository.assert_cancelled_scope_schema()
+        with self.assertRaises(IngestionError) as missing:
+            repository.assert_pm_scope_schema()
+        self.assertEqual(missing.exception.code, "DATABASE_MIGRATION_MISSING")
+        with self.database.connect(autocommit=True) as connection:
+            artifact = connection.execute("""INSERT INTO ingestion.artifacts
+                (artifact_kind,storage_provider,bucket_alias,object_key,original_filename,sha256,byte_size)
+                VALUES ('opera_xml','local_fixture','fixture','old/source.xml','source.xml',%s,0) RETURNING id""",
+                ("a" * 64,)).fetchone()[0]
+            output_ids = {}
+            for kind in ("daily_xlsx", "result_json", "structured_result_json"):
+                output_ids[kind] = connection.execute("""INSERT INTO ingestion.artifacts
+                    (artifact_kind,storage_provider,bucket_alias,object_key,original_filename,sha256,byte_size)
+                    VALUES (%s,'local_fixture','fixture',%s,%s,%s,0) RETURNING id""",
+                    (kind, "old/" + kind, kind, "d" * 64)).fetchone()[0]
+            run = connection.execute("""INSERT INTO ingestion.processing_runs
+                (run_key,pipeline_type,source_artifact_id,result_artifact_id,run_status,result_delivery_mode,business_date,
+                 delivered_processor_version,delivered_rule_set_sha256,result_schema_version,
+                 delivery_sha256,validated_at,finished_at)
+                VALUES ('old-cancelled','opera_daily',%s,%s,'accepted','artifact_callback','2026-07-27',
+                        '4.3.0',%s,'4.0',%s,now(),now()) RETURNING id""",
+                (artifact, output_ids["result_json"], "b" * 64, "c" * 64)).fetchone()[0]
+            version = connection.execute("""INSERT INTO finance.daily_versions
+                (business_date,version_no,processing_run_id,source_artifact_id,daily_report_artifact_id,
+                 result_json_artifact_id,structured_result_artifact_id,version_status,
+                 processor_version,rule_set_sha256,result_schema_version,result_sha256,source_rows,
+                 retained_rows,excluded_rate_code_rows,duplicate_rows,validation_failed_rows,
+                 price_unmatched_rows,excluded_cancelled_rows,validated_at,result_delivery_mode)
+                VALUES ('2026-07-27',1,%s,%s,%s,%s,%s,'validated','4.3.0',%s,'4.0',%s,
+                        2,1,0,0,0,0,1,now(),'artifact_callback') RETURNING id""",
+                (run, artifact, output_ids["daily_xlsx"], output_ids["result_json"],
+                 output_ids["structured_result_json"], "b" * 64, "c" * 64)).fetchone()[0]
+            connection.execute("""INSERT INTO finance.daily_records
+                (daily_version_id,source_sequence,source_location,outcome,decision_codes,
+                 booking_source_match_status)
+                VALUES (%s,1,'reservation[1]','excluded_cancelled',
+                        ARRAY['RESERVATION_CANCELLED_EXCLUDED'],'missing_group_code')""", (version,))
+            # Processor 4.3 included this PM row. A schema upgrade must preserve it.
+            connection.execute("""INSERT INTO finance.daily_records
+                (daily_version_id,source_sequence,source_location,outcome,decision_codes,block_code,
+                 adults,children,company_name,company_key,confirmation_no,disp_room_no,
+                 effective_rate_amount,full_name,no_of_rooms,products,rate_code,room_category_label,
+                 arrival,departure,nights,real_price,total_price,channel_key,pricing_method,
+                 booking_source_match_status)
+                VALUES (%s,2,'reservation[2]','retained',ARRAY['PRICE_REFERENCE_EXACT'],'',2,0,
+                        'SYNTHETIC COMPANY','SYNTHETIC COMPANY','SYNTHETIC CONF','SYNTHETIC ROOM',
+                        900,'SYNTHETIC GUEST',1,'SYNTHETIC','GRPA1','PM','2026-07-27','2026-07-28',
+                        1,900,900,'Group','price_reference_exact','missing_group_code')""", (version,))
+            before = connection.execute("SELECT to_jsonb(v) FROM finance.daily_versions v WHERE id=%s", (version,)).fetchone()[0]
+            records_before = connection.execute("SELECT to_jsonb(r) FROM finance.daily_records r ORDER BY source_sequence").fetchall()
+            guards = connection.execute("SELECT tgname,pg_get_triggerdef(oid) FROM pg_trigger WHERE NOT tgisinternal ORDER BY tgname").fetchall()
+            connection.execute("CREATE ROLE arr_pm_fixture")
+            connection.execute("CREATE ROLE arr_pm_column_fixture")
+            connection.execute("GRANT USAGE ON SCHEMA finance TO arr_pm_fixture,arr_pm_column_fixture")
+            connection.execute("GRANT SELECT,INSERT ON finance.daily_versions TO arr_pm_fixture")
+            connection.execute("GRANT INSERT (excluded_cancelled_rows) ON finance.daily_versions TO arr_pm_column_fixture")
+            connection.execute((PROJECT / "database/021_daily_pm_exclusion.sql").read_text(), prepare=False)
+            repository.assert_data_source_schema()
+            after = connection.execute("SELECT to_jsonb(v) FROM finance.daily_versions v WHERE id=%s", (version,)).fetchone()[0]
+            self.assertEqual(after.pop("excluded_pm_rows"), 0)
+            self.assertEqual(after, before)
+            self.assertEqual(connection.execute("SELECT to_jsonb(r) FROM finance.daily_records r ORDER BY source_sequence").fetchall(), records_before)
+            self.assertEqual(connection.execute("SELECT tgname,pg_get_triggerdef(oid) FROM pg_trigger WHERE NOT tgisinternal ORDER BY tgname").fetchall(), guards)
+            self.assertEqual(connection.execute("SELECT has_column_privilege('arr_pm_fixture','finance.daily_versions','excluded_pm_rows','INSERT'),has_table_privilege('arr_pm_fixture','finance.daily_versions','UPDATE'),has_column_privilege('arr_pm_column_fixture','finance.daily_versions','excluded_pm_rows','INSERT')").fetchone(), (True, False, False))
+            for statement in (
+                "UPDATE finance.daily_versions SET excluded_pm_rows=-1 WHERE id=%s",
+                "UPDATE finance.daily_versions SET excluded_pm_rows=1 WHERE id=%s",
+            ):
+                with self.assertRaises(psycopg.errors.CheckViolation):
+                    connection.execute(statement, (version,))
+            connection.execute("UPDATE finance.daily_versions SET excluded_cancelled_rows=0,excluded_pm_rows=1 WHERE id=%s", (version,))
+            with self.assertRaisesRegex(psycopg.errors.RaiseException, "PM-exclusion facts exist"):
+                connection.execute((PROJECT / "database/021_daily_pm_exclusion.down.sql").read_text(), prepare=False)
+            connection.execute("ROLLBACK")
+            connection.execute("SET ROLE arr_pm_fixture")
+            with self.assertRaises(psycopg.errors.InsufficientPrivilege):
+                connection.execute("UPDATE finance.daily_versions SET excluded_pm_rows=0 WHERE id=%s", (version,))
+            connection.execute("RESET ROLE")
+
+    def test_empty_rollback_preserves_020_and_allows_reapply(self):
+        self.database.reset_database()
+        repository = PostgresIngestionRepository(DatabaseConfig("owned-fixture"), connect=self.database.connect)
+        repository.assert_data_source_schema()
+        with self.database.connect(autocommit=True) as connection:
+            connection.execute((PROJECT / "database/021_daily_pm_exclusion.down.sql").read_text(), prepare=False)
+        repository.assert_cancelled_scope_schema()
+        with self.assertRaises(IngestionError) as missing:
+            repository.assert_pm_scope_schema()
+        self.assertEqual(missing.exception.code, "DATABASE_MIGRATION_MISSING")
+        with self.database.connect(autocommit=True) as connection:
+            connection.execute((PROJECT / "database/021_daily_pm_exclusion.sql").read_text(), prepare=False)
+        repository.assert_data_source_schema()
+
+
+@unittest.skipUnless(os.environ.get("ARR_TEST_LOCAL_POSTGRES") == "1", "owned PostgreSQL opt-in required")
+class PMFinanceProcessingTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.database = TemporaryPostgres().__enter__()
+        cls.addClassCleanup(cls.database.__exit__, None, None, None)
+        cls.policy = load_processor_policy(PROJECT)
+
+    setUp = direct_helpers.DirectDataPostgresTests.setUp
+    configure = direct_helpers.DirectDataTests.configure
+    execute = direct_helpers.DirectDataTests.execute
+    sql = direct_helpers.DirectDataPostgresTests.sql
+    count = direct_helpers.DirectDataPostgresTests.count
+    monthly_worker = direct_helpers.DirectDataPostgresTests.monthly_worker
+
+    def make_first_pm(self):
+        row = self.transport.rows[0]
+        row["roomStay"]["roomType"] = "PM"
+        row["roomStay"]["currentRoomInfo"]["roomType"] = "PM"
+        for rate in row["roomStay"]["roomRates"]:
+            rate["roomType"] = "PM"
+
+    def test_direct_pm_is_audited_and_daily_monthly_exclude_it_without_refetch(self):
+        self.make_first_pm()
+        self.assertEqual(self.execute().status, "succeeded")
+        self.assertEqual(self.sql("SELECT result_schema_version,source_rows,retained_rows,excluded_pm_rows FROM finance.daily_versions"), [("5.0", 6, 5, 1)])
+        self.assertEqual(self.sql("SELECT outcome,decision_codes,real_price,channel_key FROM finance.daily_records WHERE source_sequence=1"), [("excluded_pm", ["ROOM_TYPE_PM_EXCLUDED"], None, None)])
+        self.assertEqual(self.count("finance.v_active_daily_facts"), 5)
+        self.assertEqual(self.monthly_worker().process_next().status, "published")
+        key, size = self.sql("SELECT a.object_key,a.byte_size FROM reporting.monthly_runs r JOIN ingestion.artifacts a ON a.id=r.workbook_artifact_id")[0]
+        output = self.files / "monthly.xlsx"
+        self.store.materialize(key, output, size)
+        workbook = load_workbook(output, data_only=False)
+        try:
+            self.assertEqual(sum(cell.data_type == "f" for sheet in workbook for row in sheet for cell in row), 5)
+        finally:
+            workbook.close()
+        calls = len(self.transport.calls)
+        self.assertEqual(self.execute().status, "succeeded")
+        self.assertEqual(len(self.transport.calls), calls)
+        self.assertEqual(self.count("finance.daily_versions"), 1)
+
+    def test_xml_pm_is_audited_without_required_field_or_price_review(self):
+        outcome = self.coordinator.submit("source.xml", xml_document(pm_row(), reservation(2)).encode())
+        self.assertEqual(outcome["status"], "succeeded")
+        self.assertEqual(self.sql("SELECT result_schema_version,source_rows,retained_rows,excluded_pm_rows FROM finance.daily_versions"), [("4.0", 2, 1, 1)])
+        self.assertEqual(self.sql("SELECT outcome FROM finance.daily_records ORDER BY source_sequence"), [("excluded_pm",), ("retained",)])
+        self.assertEqual(self.count("finance.v_active_daily_facts"), 1)
+        self.assertEqual(self.monthly_worker().process_next().status, "published")
+
+    def test_xml_failed_batch_keeps_pm_audit_without_monthly_commit(self):
+        outcome = self.coordinator.submit("source.xml", xml_document(pm_row(), reservation(2, departure="2026-07-26")).encode())
+        self.assertEqual(outcome["status"], "failed")
+        self.assertEqual(self.sql("SELECT version_status,source_rows,excluded_pm_rows,validation_failed_rows FROM finance.daily_versions"), [("rejected", 2, 1, 1)])
+        self.assertEqual(self.count("finance.current_daily_versions"), 0)
+        self.assertEqual(self.monthly_worker().process_next().status, "idle")
+
+    def test_database_pm_record_classification_guard_rejects_wrong_room_or_price(self):
+        import psycopg
+        self.make_first_pm()
+        self.assertEqual(self.execute().status, "succeeded")
+        with self.database.connect(autocommit=True) as connection:
+            for statement in (
+                "UPDATE finance.daily_records SET room_category_label='PM-SUITE' WHERE source_sequence=1",
+                "UPDATE finance.daily_records SET real_price=0 WHERE source_sequence=1",
+                "UPDATE finance.daily_records SET channel_key='Group' WHERE source_sequence=1",
+                "UPDATE finance.daily_records SET decision_codes=ARRAY['RATE_CODE_EXCLUDED'] WHERE source_sequence=1",
+            ):
+                with self.subTest(statement=statement), self.assertRaises(psycopg.errors.CheckViolation):
+                    connection.execute(statement)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_arr_pm_review.py

@@ -0,0 +1,98 @@
+"""PM eligibility is shared with processing, before source-field completion."""
+import json
+from pathlib import Path
+import tempfile
+import unittest
+
+from arr_processing.policy import load_processor_policy
+from arr_web.arr_data_review import DataFieldReviews
+from arr_web.contracts import PortalError
+from tests.test_arr_data_review import source_document, raw, gap, CONTEXT, DAY, REQUEST, ACTOR, SOURCE_MANIFEST
+
+
+class PMReviewTests(unittest.TestCase):
+    def setUp(self):
+        temporary = tempfile.TemporaryDirectory(prefix="arr-pm-review-")
+        self.addCleanup(temporary.cleanup)
+        self.root = Path(temporary.name) / "reviews"
+        self.policy = load_processor_policy(Path(__file__).resolve().parents[1])
+        self.service = DataFieldReviews(root=self.root, policy=self.policy, context=CONTEXT)
+        self.document = source_document(2)
+
+    def prepare(self):
+        return self.service.prepare(REQUEST, raw(self.document), SOURCE_MANIFEST, DAY)
+
+    def test_explicit_pm_needs_no_fields_and_does_not_hide_other_rows(self):
+        self.document["records"][0]["fields"]["ROOM_CATEGORY_LABEL"] = {"state": "available", "value": " pm "}
+        for field in ("DISP_ROOM_NO", "RATE_CODE", "BLOCK_CODE", "ADULTS"):
+            gap(self.document, field)
+        gap(self.document, "DISP_ROOM_NO", sequence=2)
+        review = self.prepare()
+        self.assertEqual([item["item_id"] for item in review["items"]], ["2:DISP_ROOM_NO"])
+        self.assertEqual(review["excluded_pm_count"], 1)
+        self.assertEqual(review["excluded_cancelled_count"], 0)
+        with self.assertRaisesRegex(PortalError, "PM"):
+            self.service.update(REQUEST, "1:BLOCK_CODE", review["revision"], "IGNORED", ACTOR)
+
+    def test_zero_price_or_nine_thousand_room_number_does_not_identify_pm(self):
+        for row in self.document["records"]:
+            row["fields"]["EFFECTIVE_RATE_AMOUNT"] = {"state": "available", "value": "0"}
+            row["fields"]["DISP_ROOM_NO"] = {"state": "available", "value": "9002"}
+            row["fields"]["ROOM_CATEGORY_LABEL"] = {"state": "available", "value": "PMX"}
+        gap(self.document, "ADULTS")
+        review = self.prepare()
+        self.assertEqual(review["excluded_pm_count"], 0)
+        self.assertIn("1:ADULTS", [item["item_id"] for item in review["items"]])
+
+    def test_cancelled_pm_is_counted_once_as_cancelled(self):
+        row = self.document["records"][0]
+        row["reservation_status"] = "Cancelled"
+        row["fields"]["ROOM_CATEGORY_LABEL"] = {"state": "available", "value": "PM"}
+        gap(self.document, "BLOCK_CODE", sequence=2)
+        review = self.prepare()
+        self.assertEqual((review["excluded_cancelled_count"], review["excluded_pm_count"]), (1, 0))
+
+    def test_prior_decisions_remain_audited_when_completed_room_type_is_pm(self):
+        gap(self.document, "BLOCK_CODE")
+        gap(self.document, "ROOM_CATEGORY_LABEL")
+        review = self.prepare()
+        original = self.service.original(REQUEST)
+        review = self.service.update(REQUEST, "1:BLOCK_CODE", review["revision"], "VERIFIED", ACTOR)
+        review = self.service.update(REQUEST, "1:ROOM_CATEGORY_LABEL", review["revision"], "PM", ACTOR)
+        self.assertEqual([item["item_id"] for item in review["items"]], ["1:ROOM_CATEGORY_LABEL"])
+        self.assertTrue(review["items"][0]["confirmed"])
+        self.assertEqual(review["excluded_pm_count"], 1)
+        self.assertTrue(review["can_finalize"])
+        with self.assertRaisesRegex(PortalError, "PM"):
+            self.service.update(REQUEST, "1:BLOCK_CODE", review["revision"], "CHANGED", ACTOR)
+        self.service.finalize(REQUEST, review["revision"], ACTOR)
+        derived = json.loads(self.service.payload(REQUEST)[0])
+        self.assertEqual(self.service.original(REQUEST), original)
+        changes = derived["manual_data_review"]["manifest"]["changes"]
+        self.assertEqual({item["field"]: item["value"] for item in changes},
+                         {"BLOCK_CODE": "VERIFIED", "ROOM_CATEGORY_LABEL": "PM"})
+
+    def test_existing_room_type_normalization_is_shared_by_exclusion_count_and_edit_guard(self):
+        self.document["records"][0]["fields"]["ROOM_CATEGORY_LABEL"] = {"state": "available", "value": "PM😀"}
+        gap(self.document, "BLOCK_CODE")
+        gap(self.document, "DISP_ROOM_NO", sequence=2)
+        review = self.prepare()
+        self.assertEqual(review["excluded_pm_count"], 1)
+        self.assertEqual([item["item_id"] for item in review["items"]], ["2:DISP_ROOM_NO"])
+        with self.assertRaisesRegex(PortalError, "PM"):
+            self.service.update(REQUEST, "1:BLOCK_CODE", review["revision"], "IGNORED", ACTOR)
+
+    def test_manually_completed_room_type_can_be_corrected_before_finalization(self):
+        gap(self.document, "ROOM_CATEGORY_LABEL")
+        gap(self.document, "BLOCK_CODE")
+        review = self.prepare()
+        review = self.service.update(REQUEST, "1:ROOM_CATEGORY_LABEL", review["revision"], "PM", ACTOR)
+        self.assertEqual(review["pending_count"], 0)
+        review = self.service.update(REQUEST, "1:ROOM_CATEGORY_LABEL", review["revision"], "SU6", ACTOR)
+        self.assertEqual(review["excluded_pm_count"], 0)
+        self.assertEqual(review["pending_count"], 1)
+        self.assertIn("1:BLOCK_CODE", [item["item_id"] for item in review["items"]])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_arr_pm_scope.py

@@ -0,0 +1,228 @@
+"""Offline PM scope checks for both daily inputs and the frozen legacy replay."""
+from __future__ import annotations
+
+import argparse
+import contextlib
+import copy
+import io
+import json
+from pathlib import Path
+import tempfile
+import unittest
+import xml.etree.ElementTree as ET
+
+from tests.test_arr_opera_daily_ingest import (
+    core, reservation, run_processor, validator, xml_document,
+)
+from tests.test_arr_data_review import source_document
+
+
+def pm_xml_row(sequence, *, label="PM", room=None, status=None, incomplete=False,
+               rate_amount="900"):
+    node = ET.fromstring(reservation(sequence, room=room, rate_amount=rate_amount))
+    node.find("ROOM_CATEGORY_LABEL").text = label
+    if status is not None:
+        ET.SubElement(node, "SHORT_RESV_STATUS").text = status
+    if incomplete:
+        for field, value in (("RATE_CODE", ""), ("DISP_ROOM_NO", ""),
+                             ("COMPANY_NAME", ""), ("FULL_NAME", ""),
+                             ("ADULTS", "-1"), ("NO_OF_ROOMS", "0"),
+                             ("EFFECTIVE_RATE_AMOUNT", "INVALID"),
+                             ("TRUNC_END", "2026-07-26")):
+            node.find(field).text = value
+    return ET.tostring(node, encoding="unicode")
+
+
+class PMScopeTests(unittest.TestCase):
+    def setUp(self):
+        temporary = tempfile.TemporaryDirectory(prefix="arr-pm-scope-")
+        self.addCleanup(temporary.cleanup)
+        self.root = Path(temporary.name)
+
+    def classify_xml(self, *rows, apply_scope=True):
+        path = self.root / "source.xml"
+        path.write_text(xml_document(*rows))
+        business_date, nodes = core.read_xml(path)
+        return core.classify_input_records(nodes, business_date, apply_scope=apply_scope)
+
+    def run_data(self, document, name="output"):
+        path = self.root / (name + ".json")
+        path.write_text(json.dumps(document))
+        output = self.root / name
+        args = argparse.Namespace(xml=None, data_json=str(path), output_dir=str(output),
+            result_json=str(output / "result.json"),
+            structured_result_json=str(output / "structured-result.json"))
+        before = path.read_bytes()
+        with contextlib.redirect_stdout(io.StringIO()):
+            exit_code = core.process(args)
+        self.assertEqual(path.read_bytes(), before)
+        return exit_code, json.loads((output / "structured-result.json").read_text())
+
+    def test_xml_pm_precedes_rate_and_required_fields_but_cancelled_takes_precedence(self):
+        rows, eligible, removed_rate, duplicates, errors = self.classify_xml(
+            pm_xml_row(1, label=" pm ", incomplete=True),
+            pm_xml_row(2, status="CXL", incomplete=True), reservation(3))
+        self.assertEqual(errors, [])
+        self.assertEqual((removed_rate, duplicates), (0, 0))
+        self.assertEqual([row["_SOURCE_INDEX"] for row in eligible], [3])
+        self.assertEqual([row["_OUTCOME"] for row in rows],
+                         ["excluded_pm", "excluded_cancelled", "pending"])
+        self.assertEqual(rows[0]["_DECISION_CODES"], ["ROOM_TYPE_PM_EXCLUDED"])
+        self.assertEqual(rows[1]["_DECISION_CODES"], ["RESERVATION_CANCELLED_EXCLUDED"])
+        self.assertEqual(rows[0]["ROOM_CATEGORY_LABEL"], "pm")
+        self.assertEqual(rows[0]["CONFIRMATION_NO"], "SYNTHETIC-CONF-1")
+        self.assertEqual(rows[0]["_SOURCE_LOCATION"], "reservation[1]")
+
+    def test_only_exact_explicit_pm_room_type_is_excluded(self):
+        for label in ("PM", "pm", " pM ", "\tPm\n"):
+            with self.subTest(label=label):
+                rows, _, _, _, errors = self.classify_xml(
+                    pm_xml_row(1, label=label, incomplete=True), reservation(2))
+                self.assertEqual(errors, [])
+                self.assertTrue(core.is_pm_record(rows[0]))
+        for label in (None, "", "PM1", "PM ROOM", "P M", "SU6"):
+            with self.subTest(label=label):
+                rows, _, _, _, errors = self.classify_xml(
+                    pm_xml_row(1, label=label, room="PM", rate_amount="0"), reservation(2))
+                self.assertEqual(errors, [])
+                self.assertFalse(core.is_pm_record(rows[0]))
+                self.assertEqual(rows[0]["_OUTCOME"], "pending")
+
+    def test_pm_does_not_consume_a_real_room_duplicate_key_or_create_price_issue(self):
+        rows, eligible, _, duplicates, errors = self.classify_xml(
+            pm_xml_row(1, room="SAME-ROOM", rate_amount="0"),
+            reservation(2, room="SAME-ROOM"), reservation(3, room="SAME-ROOM"))
+        self.assertEqual(errors, [])
+        self.assertEqual([row["_SOURCE_INDEX"] for row in eligible], [2])
+        self.assertEqual(duplicates, 1)
+        self.assertEqual([row["_OUTCOME"] for row in rows],
+                         ["excluded_pm", "pending", "duplicate"])
+        self.assertEqual(rows[2]["_DUPLICATE_OF_SOURCE_SEQUENCE"], 2)
+        prices = core.load_price_map(core.PRICE_REFERENCE)
+        self.assertEqual(core.apply_prices_classified(eligible, prices), [])
+        self.assertEqual(core.review_issues(eligible, prices), [])
+
+    def test_xml_success_and_independent_replay_count_pm_and_cancelled_separately(self):
+        code, source, output, _, payload = run_processor(xml_document(
+            pm_xml_row(1, label=" pm ", incomplete=True),
+            pm_xml_row(2, status="CXL", incomplete=True), reservation(3)), self.root)
+        self.assertEqual(code, 0, payload["errors"])
+        self.assertEqual(payload["processor_version"], "4.4.0")
+        self.assertEqual(payload["outcome_counts"]["excluded_pm"], 1)
+        self.assertEqual(payload["outcome_counts"]["excluded_cancelled"], 1)
+        self.assertEqual(sum(payload["outcome_counts"].values()), 3)
+        self.assertEqual((payload["source_rows"], payload["output_rows"]), (3, 1))
+        self.assertEqual(payload["artifacts"]["source_xml"]["sha256"], core.sha256_file(source))
+        self.assertEqual(payload["records"][0]["room_category_label"], "pm")
+        for field in ("real_price", "total_price", "kb_amount", "channel_key", "pricing_method"):
+            self.assertIsNone(payload["records"][0][field])
+
+        # A structurally valid cancellation reclassification cannot replace PM evidence.
+        forged = copy.deepcopy(payload)
+        forged["records"][0]["outcome"] = "excluded_cancelled"
+        forged["records"][0]["decision_codes"] = ["RESERVATION_CANCELLED_EXCLUDED"]
+        forged["outcome_counts"]["excluded_cancelled"] += 1
+        forged["outcome_counts"]["excluded_pm"] -= 1
+        core.validate_structured_completeness(forged)
+        structured = output / "structured-result.json"
+        structured.write_text(json.dumps(forged))
+        errors = validator.validate(argparse.Namespace(xml=str(source),
+            daily=str(next(output.glob("*.xlsx"))), result_json=str(output / "result.json"),
+            structured_result_json=str(structured), price_reference=str(core.PRICE_REFERENCE)))
+        self.assertIn("OUTPUT_STRUCTURED_OUTCOME_MISMATCH", {error.code for error in errors})
+        self.assertIn("OUTPUT_STRUCTURED_RECORD_MISMATCH", {error.code for error in errors})
+
+    def test_pm_self_validation_and_schemas_require_explicit_type_reason_and_null_facts(self):
+        code, _, _, _, payload = run_processor(xml_document(
+            pm_xml_row(1, label="pM"), reservation(2)), self.root)
+        self.assertEqual(code, 0, payload["errors"])
+        for field, value in (("room_category_label", "SU6"),
+                             ("decision_codes", ["RATE_CODE_NOT_WHITELISTED"]),
+                             ("real_price", 0), ("total_price", 0), ("kb_amount", 0),
+                             ("channel_key", "QBD"), ("pricing_method", "manual_review")):
+            with self.subTest(field=field):
+                forged = copy.deepcopy(payload)
+                forged["records"][0][field] = value
+                with self.assertRaises(core.ProcessingFailure):
+                    core.validate_structured_completeness(forged)
+        forged = copy.deepcopy(payload)
+        del forged["outcome_counts"]["excluded_pm"]
+        with self.assertRaises(core.ProcessingFailure):
+            core.validate_structured_completeness(forged)
+        for name in ("structured-result.schema.json", "data-structured-result.schema.json"):
+            schema = json.loads((core.SKILL_ROOT / "references" / name).read_text())
+            self.assertEqual(schema["properties"]["processor_version"]["const"], "4.4.0")
+            self.assertIn("excluded_pm", schema["properties"]["outcome_counts"]["required"])
+            rule = next(item for item in schema["$defs"]["record"]["allOf"]
+                        if item["if"]["properties"]["outcome"].get("const") == "excluded_pm")
+            self.assertEqual(rule["then"]["properties"]["decision_codes"]["const"],
+                             ["ROOM_TYPE_PM_EXCLUDED"])
+            self.assertEqual(rule["then"]["properties"]["real_price"]["type"], "null")
+
+    def test_xml_review_and_failure_keep_pm_outside_pricing_and_other_validation(self):
+        code, _, _, _, payload = run_processor(xml_document(
+            pm_xml_row(1, incomplete=True), reservation(2, rate_amount="8765")),
+            self.root / "review")
+        self.assertEqual(code, 0, payload["errors"])
+        self.assertEqual(payload["status"], "review_required")
+        self.assertEqual(payload["outcome_counts"]["excluded_pm"], 1)
+        self.assertEqual(payload["review_required_rows"], 1)
+        self.assertEqual(payload["review_issues"][0]["affected_records"], 1)
+        invalid = ET.fromstring(reservation(2))
+        invalid.find("DISP_ROOM_NO").text = ""
+        code, _, _, _, payload = run_processor(xml_document(
+            pm_xml_row(1, incomplete=True), ET.tostring(invalid, encoding="unicode")),
+            self.root / "failure")
+        self.assertNotEqual(code, 0)
+        self.assertEqual(payload["status"], "failed")
+        self.assertEqual(payload["outcome_counts"]["excluded_pm"], 1)
+        self.assertEqual(payload["outcome_counts"]["validation_failed"], 1)
+
+    def test_data_success_review_and_failure_keep_pm_outside_business_rules(self):
+        baseline = source_document(2)
+        for field in baseline["records"][0]["fields"]:
+            baseline["records"][0]["fields"][field] = {"state": "missing", "value": None}
+        baseline["records"][0]["fields"]["ROOM_CATEGORY_LABEL"] = {"state": "available", "value": " pm "}
+        baseline["input_complete"] = False
+        baseline["status"] = "collected_with_gaps"
+        code, payload = self.run_data(baseline)
+        self.assertEqual(code, 0, payload["errors"])
+        self.assertEqual(payload["outcome_counts"]["excluded_pm"], 1)
+        self.assertEqual(payload["output_rows"], 1)
+        self.assertEqual(payload["removed_by_rate_code"], 0)
+        self.assertEqual(payload["records"][0]["room_category_label"], "pm")
+        review = copy.deepcopy(baseline)
+        review["records"][1]["fields"]["EFFECTIVE_RATE_AMOUNT"]["value"] = "8765"
+        code, payload = self.run_data(review, "review")
+        self.assertEqual(code, 0, payload["errors"])
+        self.assertEqual(payload["status"], "review_required")
+        self.assertEqual(payload["outcome_counts"]["excluded_pm"], 1)
+        self.assertEqual(payload["review_required_rows"], 1)
+        failed = copy.deepcopy(baseline)
+        failed["records"][1]["fields"]["DISP_ROOM_NO"] = {"state": "missing", "value": None}
+        code, payload = self.run_data(failed, "failure")
+        self.assertNotEqual(code, 0)
+        self.assertEqual(payload["status"], "failed")
+        self.assertEqual(payload["outcome_counts"]["excluded_pm"], 1)
+        self.assertEqual(payload["outcome_counts"]["validation_failed"], 1)
+
+    def test_frozen_legacy_v3_replay_keeps_pm_and_never_projects_new_outcomes(self):
+        code, _, _, _, payload = run_processor(xml_document(
+            pm_xml_row(1), reservation(2)), self.root, legacy_v3_output=True)
+        self.assertEqual(code, 0, payload["errors"])
+        self.assertEqual(payload["result_schema_version"], "3.0")
+        self.assertEqual(payload["output_rows"], 2)
+        self.assertEqual(set(payload["outcome_counts"]), core.LEGACY_DIRECT_FINAL_OUTCOMES)
+        self.assertEqual([row["outcome"] for row in payload["records"]], ["retained", "retained"])
+        rows, _, _, _, _ = self.classify_xml(pm_xml_row(1), reservation(2))
+        for function in (core.build_legacy_direct_structured_result,
+                         core.build_legacy_direct_failed_structured_result):
+            with self.subTest(function=function.__name__), self.assertRaises(core.ProcessingFailure):
+                if function is core.build_legacy_direct_structured_result:
+                    function(None, rows[:1], [], self.root / "source.xml", None, None)
+                else:
+                    function(None, rows[:1], self.root / "source.xml", None, None, [])
+
+
+if __name__ == "__main__":
+    unittest.main()