diff --git a/.project-docs/30-worklog/tasks/20261008-production-review-9e7b.md b/.project-docs/30-worklog/tasks/20261008-production-review-9e7b.md
index e947816..884f00b 100644
--- a/.project-docs/30-worklog/tasks/20261008-production-review-9e7b.md
+++ b/.project-docs/30-worklog/tasks/20261008-production-review-9e7b.md
@@ -12,7 +12,7 @@
## Scope
-- 修复生产 OHIP 查询、环境内任务恢复及接口字段人工完善；保留 XML 和原筛选、定价、日报/月报规则。
+- 修复生产 OHIP 查询、环境内任务恢复及接口字段人工完善；保留 XML、原费率筛选、定价、日报/月报规则，并落实用户确认的取消预订排除规则。
## Intent And Constraints
@@ -24,7 +24,7 @@
- Added source-field completion before deterministic processing: only candidate records' unresolved/invalid fields may be edited. Normal optional block/package omission is now recognized under the narrow Oracle association policy described below; unresolved, failed or contradictory evidence still needs resolution. Original bytes are retained, revisioned staff decisions record the authenticated actor, and confirmation freezes a derived source. The same download request resumes from its saved acquisition without querying OHIP again. Existing price review, Finance validation, daily generation and monthly outbox behavior remain in use; XML processing is retained.
- Browser recovery is scoped to this service/source context and user, preventing an old sandbox request from restoring into the production instance. Lost save/finalize responses, concurrent revisions and interrupted confirmation recover without a second processing intent.
- Updated the existing local service at `http://127.0.0.1:8875/` to run this isolated checkout, retaining hotel57106, credentials, private database and output storage. Previous private launcher/config/plist were backed up before restart. No production deployment, remote push or integration into the unowned dirty primary checkout occurred.
-- Reused the complete, immutable real 2026-10-07 capture; verified all531 capture files and both original hashes before importing into request `45e0e20ee75a46e098a0e32d650133a9`. Its initial26 source decisions were1 room,18 block codes and7 package lists. The optional-association follow-up below reinterpreted only corroborated normal omissions, preserving all original bytes and staff decisions. Current live review is revision1 with only1 unresolved required room; no processing job, staff confirmations or real daily/monthly exist. Import and reanalysis made zero new Oracle calls.
+- Reused the complete, immutable real 2026-10-07 capture; verified all531 capture files and both original hashes before importing into request `45e0e20ee75a46e098a0e32d650133a9`. Its initial26 source decisions were1 room,18 block codes and7 package lists. The optional-association follow-up below reinterpreted only corroborated normal omissions, preserving all original bytes and staff decisions. The later approved cancellation follow-up supersedes this pending-room state: the source review is finalized revision4 with zero field decisions, and the same10/7 request has entered the existing two-key price review. No real daily/monthly exists yet. Import, reanalysis and cancellation-scope continuation made zero new Oracle calls.
## Verification
@@ -38,7 +38,7 @@
## Follow-ups
-- The user explicitly deferred API/XML population alignment. That issue remains separate and must not block fixing normal optional omission or be represented as resolved by this change. The remaining10/7 room belongs to confirmation300007418, a Cancelled/unassigned source reservation; do not invent a room or silently exclude the reservation. Real report acceptance still requires resolving the actual remaining input and, when resumed by the user, the desired source selection/text mappings. Existing price review and real daily/monthly acceptance remain pending.
+- The user explicitly deferred API/XML population alignment. That issue remains separate and must not block fixing normal optional omission or be represented as resolved by this change. The user has now explicitly approved cancellation exclusion, so300007418 no longer requires a room.10/7 awaits only two existing manual-price keys (LIAN TAI/WHO1/Opera0 and QBD/GRPA3/Opera0); each affects one record/three nights. Do not choose a price, equate an Oracle0 with the processed price, or claim final daily/monthly acceptance. Source population/text alignment remains separately deferred.9/17 has received saved-status evidence but its old optional-source gaps were not reinterpreted this turn;30 active source decisions remain.
- Integration owner should promote the accepted behavior and production compatibility evidence, then integrate this branch into the primary project. Keep this checkout while the local service runs from it. It is manually created, not eligible for automatic skill-managed retirement.
- Production deployment/configuration and remote publishing are separate follow-ups. This turn only activated the local instance against the existing production read service.
@@ -155,3 +155,37 @@ Read: memory-index, project-positioning, current-state latest September sections
- Independent saved-evidence recheck: confirmation300007418/internal5131529 has no roomId in first/final search, detail roomStay/currentRoomInfo or arrival-day rate segments. Room calendar returned HTTP200 with empty roomCalendar and no associated room/history. `PM` is a room type, not a usable room number. Both search/detail show Cancelled and rateGL2200KR in the original whitelist. Existing capture cannot legitimately supply a room; cancellation is not asserted as the cause of missing data. No new Oracle call occurred.
- Verification:46 JavaScript tests passed, including failure/conflict precedence for room issues, pending dates/drafts, explicit finalization and uncertain request recovery; node syntax and Git whitespace checks passed. CUA refreshed the live page and verified the new explanation for300007418,0/1 confirmed and disabled report generation. Selecting10/6 enabled download without submitting; returning to10/7 restored the same item. No new task or actual download was started. Authenticated local read confirms revision1/pending1/job=null; original source SHA remains805799ba2f3edd7f1cf18c14e5304908061d6194a241c5fc255192728b7f388a. Static assets were served immediately; no service restart was needed.
- Private screenshot: `/Users/chillishark/Library/Application Support/ARR2.0/production-validation-20261007/room-missing-in-scope-20261008.png`. Result tab retained as deliverable. No real source/manual decision was changed and no real daily/monthly generated. Follow-up remains a verified room value or a separately requested business-rule/source-scope decision; do not invent a room, silently drop a candidate or call the report complete. Promote the more precise explanation to canonical product documentation at integration if useful.
+
+## Same-task Follow-up: Cancelled and Unassigned Is Not a Room-completion Task
+
+- User challenges the direction: a cancelled/unassigned reservation reasonably has no room. The previous repeated instruction to obtain/fill a real room was not an adequate product resolution. Original code's required-room validation explains the technical block but does not establish that the expanded API population should be subjected to an impossible business completion step. Do not invent a room or assume cancellation universally removes historical room assignments.
+- Concurrent gate resumed the same owned feature task/worktree/branch successfully with no peers; retained project context applies. Planning gate Passed for read-only current-query/capture scope analysis, not approval to add a status exclusion. The user had earlier deferred broad XML/API population alignment; this challenge authorizes investigating the immediate cancellation/room contradiction, not silently adopting a universal new status filter.
+- Current search passes only arrivalStartDate/arrivalEndDate/limit/offset. Saved actual result contains6 Cancelled rows;2 are in the rate whitelist. Confirmation300007418 is the only whitelisted cancelled/unassigned row; confirmation300755329 is whitelisted Cancelled but retains a valid room. Total58 source rows/40 rate candidates. Excluding only the row that causes a room error would be an arbitrary pass-making rule; a consistent proposed Cancelled exclusion affects2 rate candidates, not only1.
+- Independent Oracle contract check confirms arrival dates, reservationStatuses, actualArrivals, expectedArrivals and dayOfArrivalCancels are separate parameters. The latter is only arrival-day cancellations, not a general exclude-cancelled switch. Saved production data proves this particular date-only request returned Cancelled; do not assert an undocumented universal API default or infer RES_DETAIL equivalence/CheckedOut-only selection. Official references: https://raw.githubusercontent.com/oracle/hospitality-api-docs/main/rest-api-specs/property/v1/rsv.json and https://docs.oracle.com/cd/E98457_01/opera_5_6_core_help/res_detail_help.htm.
+- Product proposal made concrete from saved evidence: preserve all original58 source rows and their audit, but exclude cancelled reservations from daily candidates if the user selects that new reporting policy. Under the captured10/7 rate selection it affects2 of40 candidates. No claim is made that the remaining daily/monthly already passes pricing or full acceptance.
+- Asked one business-scope clarification with the concrete2-row impact and reason: original code does not filter status, so the user must establish whether Cancelled belongs in the daily report. Options are exclude Cancelled while retaining original data, or retain current scope and leave10/7 unresolved. This is a business requirement, not a new skill/security permission requirement. While pending, no dependent status-filter implementation, real data edits, new Oracle reads, review finalization or report generation are authorized by elapsed time.
+- Verification: read original source contract/current parameter sheet, inspect saved request and all58 captured reservation statuses, join them by typed Reservation ID to the unchanged reanalyzed data and original rate whitelist; independent official-document check. No program/business code or service change. Earlier in-scope missing-room copy is not a completed solution to this user objection; do not present it as such.
+- Follow-up/promotion candidate: once the reporting scope is established, implement the selected consistent status policy before room completion, with explicit excluded-row lineage and unchanged source bytes. Separate entity eligibility from mandatory-field validation and distinguish cancelled with retained room from cancelled without room; neither may be guessed from the room field alone. Broad remaining XML/API report equivalence stays a distinct deferred issue.
+
+
+## Same-task Follow-up: Approved Cancellation Exclusion
+
+- User explicitly selected “日报排除已取消预订” on 2026-10-08 after the scope explanation. This supersedes only the prior lack of a cancellation filter; no NoShow/actual-arrival/CheckedOut-only or broad XML/API population alignment is authorized.
+- Concurrent Task Gate passed: same task/feature/codex, owned worktree and branch match; base2417b1a; no peers or overlaps. Primary unknown dirty documentation remains untouched.
+- Project Context Loaded: retained task record, read-before-planning/planning-gate, memory-index, project-positioning, current-state, decision-index, ADR006/007, system-overview/module-map/data-flow, business-rules/success-criteria, evidence/reflection/commitments/stale indexes. Goal remains two input paths sharing deterministic Finance/monthly logic. Accepted new cancellation exclusion overrides the old status-blind classification only; preserve raw rows, rate/price/date/dedup rules and audited staff changes. Canonical September deployment/delegation restrictions are stale where October explicit production/multi-agent authorization applies. Gate Passed.
+- Plan: carry verified reservation status from OHIP; use one cancellation predicate before field validation for both input paths; independently record excluded_cancelled without mislabelling rate exclusions; update schemas/Finance forward migration; append saved-status interpretation to pending reviews without changing source bytes or staff decisions; explicitly migrate only unprocessed local tasks to the new rule identity with backups and receipt; replay saved10/7 without Oracle calls, validate and activate local service.
+- Boundary: ambiguous XML short codes such as CA are not guessed as cancellation. Clearly identified Cancelled/CXL values may be excluded; missing/unknown/conflicting evidence keeps existing behavior. Broad source-population alignment remains deferred.
+
+
+### Approved Cancellation Exclusion — Outcome and Verification
+
+- Implemented processor4.3.0 for XML4/OHIP5: exclude explicit cancellations before missing-rate, required business values, pricing and room/date deduplication. Source rows remain in the structured audit under `excluded_cancelled` with exactly `RESERVATION_CANCELLED_EXCLUDED` and null price/channel facts. Rate exclusions keep their separate count. Unknown/missing/conflicting status remains governed by existing validation; no new NoShow or CheckedOut-only rule. Frozen legacy direct-MCP3 keeps its old scope.
+- OHIP acquisition carries the verified `reservation_status` outside the15 business fields. Internal saved-status interpretation verifies every original capture file, request/response identity, both complete search rounds, sequence and search/detail status/version agreement, then appends immutable evidence to the derived source. There is no HTTP mutation route for statuses. Original acquisition bytes, source reanalysis and staff decisions remain intact; cancelled-row historical decisions stay audited but are no longer editable or blocking.
+- Finance keeps schema4/5 compatibility by processor identity, adds forward migration020 with a separate nonnegative cancellation count and balanced total/outcome constraints, and checks020 before enabling either processing path. Old rows receive0 without changing released facts; rollback rejects existing cancellation facts. No old migration was edited.
+- UI shows the cancellation count and retains original-data notice. Removed review items clear obsolete unsaved drafts; edits still in the review are preserved. Both processor distribution packages were rebuilt byte-identically to source (SHA256 `2017aaa192571f8e2869ede0a1e86310e6d05cdf517f53e94b17a0a9359bc3ba`).
+- Verification:23 processor/XML/schema/package regressions passed;14 saved-status evidence tests passed;5 cancellation-review tests passed;114 review/source/direct processing/local wrapper regressions passed with disposable real PostgreSQL, no skipped checks. Two grouped root test commands also named a nonexistent runtime module and ended with import errors; no actual loaded test failed, and the intended `test_arr_direct_runtime` plus cancellation-review/status suite was subsequently run correctly:21 tests passed. Finance agent ran62 distinct scoped tests in batches, including019→020 migration preservation, restricted-role/constraint checks, XML and JSON cancellation handling, failed-batch audit, monthly formulas and no-reacquisition retry.48 JavaScript tests and syntax/whitespace checks passed. Counts overlap and must not be added as distinct coverage.
+- Local activation: verified queue idle, stopped LaunchAgent and private PostgreSQL, made a cold full backup outside the live root, and dry-ran an explicit policy-pin upgrade on copied pending state. Only unprocessed/unfrozen tasks were eligible. The durable intent records every parent/new file hash and target policy; previous revisions, source bytes and receipts are retained. Old optional-source receipt was separately rebound under the new context; no staff decision changed. Both10/7 and9/17 received statuses from their existing complete captures. Migration020 applied only to the owned private local database, Finance version count stayed0; updated private launcher and restarted8875.
+- Actual10/7 result:58 original rows;6 cancellations excluded (2 were among prior40 rate candidates),14 non-whitelist exclusions,0 duplicates,36 controlled candidates and2 unmatched-price rows. Source review pending0/finalized revision4 with0 manual field decisions. Authenticated browser explicitly continued the original request to the existing `needs_review` state; price items are LIAN TAI/WHO1/Opera0 and QBD/GRPA3/Opera0, each1 record/3 nights. No price was entered and no real daily/monthly was committed. The user chose to enter these two prices directly in the page; it is left open at the price-review panel for that action. No price value was supplied to the agent, so it cannot finalize pricing. Generating the actual report remains dependent on the user's saved price decision.
+- Actual9/17 maintenance result:18 raw cancellations excluded from review; active decisions38→30, original unchanged, revision2. Its prior optional source omissions remain a separate pending follow-up, and this date was not submitted/processed or refetched.
+- Private evidence: cold backup `/Users/chillishark/Library/Application Support/ARR2.0/policy43-service-backup-20261008-121313`; live `policy43-upgrade-intent.json`, `policy43-upgrade-completed.json`, `policy43-activation.json`; dry run `production-validation-20261007/policy43-dryrun-da8e24ea`; private executable `upgrade_cancelled_scope.py`. Browser screenshots `cancelled-scope-no-room-review-20261008.png` and `cancelled-scope-price-review-20261008.png`. The authenticated result tab is retained. No credential/environment/grant change, Oracle business read, production deployment, remote push or primary-checkout integration occurred.
+- Promotion candidate: update canonical source-selection/business-rules/current-state with user-approved cancellation exclusion and processor4.3/020 requirements, retaining the deferred broad population alignment. Human confirmation for the cancellation policy is satisfied by this conversation; integration ownership is still required for canonical writes.
diff --git a/.project-docs/60-reflection/cases/20261008-production-review-9e7b__eligibility-before-completion.md b/.project-docs/60-reflection/cases/20261008-production-review-9e7b__eligibility-before-completion.md
new file mode 100644
index 0000000..cfcf41f
--- /dev/null
+++ b/.project-docs/60-reflection/cases/20261008-production-review-9e7b__eligibility-before-completion.md
@@ -0,0 +1,9 @@
+# Check report eligibility before requesting missing values
+
+The repeated room-completion explanation was technically accurate about the old status-blind validator but led to the wrong product action: an unassigned Cancelled booking was treated as though staff must supply a room. Field requirements only apply after the record belongs in the report. A missing mandatory schema value does not by itself justify asking staff to fabricate or locate a value that may legitimately never exist.
+
+Inspect source status and report membership before opening field completion. If the old code has no relevant eligibility rule, distinguish the existing validator from the missing business decision. Explain the full population impact and obtain the business requirement; never drop just the row that fails validation. Here the user explicitly chose to exclude all cancelled bookings, affecting two rate candidates (six raw records), so both assigned and unassigned cancellations now follow the same audited decision before field checks and deduplication.
+
+Keep original evidence and source order, record exclusion separately from rate filtering, version the processing rules, and upgrade pending immutable identities explicitly. A safe code fix is not a final report: the real10/7 run subsequently reached legitimate two-key price review, requiring actual prices before Finance/monthly acceptance.
+
+Promotion candidate: add eligibility-before-completion as a project debugging/review principle at integration; no new blanket permission or approval workflow is proposed.
diff --git a/README.md b/README.md
index 18eb3ec..e2209b4 100644
--- a/README.md
+++ b/README.md
@@ -80,7 +80,7 @@ cp .env.example .env.local
均为 `true` 才表示页面处理和下载能力可用；未登录的容器只使用无详情的 `GET /healthz` readiness。
worker 是独立无端口进程，应由进程管理器单独保活。
-自动数据入口要求 `database/008_arr_mvp_v1_rebuild.sql` 加009–019增量迁移；019支持`ohip_json`来源和5.0结果，原XML保持4.0兼容。既有数据库实际版本须在部署时核对，不自动执行迁移。017必须在同一发布窗口先确认/应用016后才可应用；它增加价格复核case/item/event审计、`awaiting_review`生命周期、最终人工价格lineage和受保护回滚。018为冻结人工清单增加`manual_override_json`工件类型。012增加月报发布元数据和Finance日版本lineage；016允许新月报OSS工件并保留旧local记录；013保存用户上传XML文件名；014/015增加Booking当前整表指针和Excel提取草稿。ARR使用原有`artifact_callback`通用工件交付表，不读写009/010的grant/MCP submission表。
+自动数据入口要求 `database/008_arr_mvp_v1_rebuild.sql` 加009–020增量迁移；020记录已取消预订排除数量和原因；019支持`ohip_json`来源和5.0结果，原XML保持4.0兼容。既有数据库实际版本须在部署时核对，不自动执行迁移。017必须在同一发布窗口先确认/应用016后才可应用；它增加价格复核case/item/event审计、`awaiting_review`生命周期、最终人工价格lineage和受保护回滚。018为冻结人工清单增加`manual_override_json`工件类型。012增加月报发布元数据和Finance日版本lineage；016允许新月报OSS工件并保留旧local记录；013保存用户上传XML文件名；014/015增加Booking当前整表指针和Excel提取草稿。ARR使用原有`artifact_callback`通用工件交付表，不读写009/010的grant/MCP submission表。
## Booking Excel 房型提取
diff --git a/arr-opera-daily-ingest.skill b/arr-opera-daily-ingest.skill
index 4b600c5..0ca58d9 100644
Binary files a/arr-opera-daily-ingest.skill and b/arr-opera-daily-ingest.skill differ
diff --git a/arr-opera-daily-ingest.zip b/arr-opera-daily-ingest.zip
index 4b600c5..0ca58d9 100644
Binary files a/arr-opera-daily-ingest.zip and b/arr-opera-daily-ingest.zip differ
diff --git a/arr-opera-daily-ingest/SKILL.md b/arr-opera-daily-ingest/SKILL.md
index 046ee5c..c563223 100644
--- a/arr-opera-daily-ingest/SKILL.md
+++ b/arr-opera-daily-ingest/SKILL.md
@@ -63,6 +63,11 @@ Read `result.json` only after the process exits.
If Python or `openpyxl` is unavailable, stop with an infrastructure failure. Do not switch to a different spreadsheet implementation.
-## Direct data entry (processor 4.2.0)
+## Direct data entry (processor 4.3.0)
Use `--data-json` instead of `--xml` for the frozen OHIP data source. Both inputs use the same classification, pricing, review and daily workbook rules. JSON is never converted into XML. XML keeps result schema 4.0; direct data uses schema 5.0 with `ingestion_mode=ohip_data` and an explicit `source_data` / `ohip_json` artifact. The independent validator accepts the same source switch. See `references/data-result.schema.json` and `references/data-structured-result.schema.json`. Trace remains blank for the 15-field data input.
+
+
+## Cancelled reservation scope (processor 4.3.0)
+
+Exclude explicitly cancelled reservations before rate-code, required-field, pricing and room/date deduplication checks. Preserve every source row in the audit with `excluded_cancelled` and `RESERVATION_CANCELLED_EXCLUDED`; never request a room for an excluded reservation. Both XML and OHIP data use the same predicate. XML recognizes only explicit CXL/CANCELLED/CANCELED status values; unknown abbreviations and conflicting status fields are not inferred. Retired direct-MCP v3 replay retains its original scope.
diff --git a/arr-opera-daily-ingest/references/business-rules.md b/arr-opera-daily-ingest/references/business-rules.md
index 9cf3feb..50c9ca8 100644
--- a/arr-opera-daily-ingest/references/business-rules.md
+++ b/arr-opera-daily-ingest/references/business-rules.md
@@ -4,7 +4,7 @@
1. Validate the invocation and parse one fixed `RES_DETAIL` XML.
2. Require exactly one XML business date and keep one audit record for every `G_RESERVATION` in XML order.
-3. Require `RATE_CODE` so whitelist membership is knowable.
+3. First exclude explicitly cancelled reservations as `excluded_cancelled` with `RESERVATION_CANCELLED_EXCLUDED`; preserve the original row and source sequence. Other records require `RATE_CODE` so whitelist membership is knowable.
4. Classify a trimmed, uppercased rate code outside the whitelist as `excluded_rate_code`.
5. Validate every whitelist candidate; classify invalid rows as `validation_failed`.
6. Deduplicate valid candidates by `DISP_ROOM_NO + ARRIVAL`; retain the first XML occurrence and classify later occurrences as `duplicate` pointing to the first source sequence.
@@ -92,3 +92,10 @@ For other companies, trim, remove `: \ / ? * [ ]`, and truncate to 31 characters
## Prohibited behavior
Do not access OSS, embed credentials, generate/update a monthly workbook, query/write a database, infer Group Code from `BLOCK_CODE`, or derive structured facts by reopening the generated XLSX.
+
+
+## Cancellation exclusion approved 2026-10-08
+
+The active processor4.3.0 excludes cancelled reservations whether or not a room is assigned, before all business-field checks and room/date deduplication. OHIP uses `reservation_status`, retained from matching search/detail responses. XML uses the explicitly named `RESV_STATUS`, `RESERVATION_STATUS`, `SHORT_RESV_STATUS` fields. Trimmed, case-insensitive `CXL`, `CANCELLED`, `CANCELED` identify cancellation. A cancellation marker conflicting with a different nonempty status is not sufficient to exclude. Missing/unknown statuses (including CA/CD), NoShow and reservation types are not interpreted as cancellation. The retired direct-MCP v3 replay keeps its old rules.
+
+Every source row stays in the structured audit. `outcome_counts.excluded_cancelled` is always present for4.3.0, including zero; it is separate from `removed_by_rate_code`. Excluded cancellations have no pricing, amount or channel facts and never participate in daily or monthly totals. Original source files and captured HTTP responses remain read-only.
diff --git a/arr-opera-daily-ingest/references/data-structured-result.schema.json b/arr-opera-daily-ingest/references/data-structured-result.schema.json
index ec244e6..d92e503 100644
--- a/arr-opera-daily-ingest/references/data-structured-result.schema.json
+++ b/arr-opera-daily-ingest/references/data-structured-result.schema.json
@@ -17,7 +17,7 @@
"activation_eligible": { "type": "boolean" },
"ingestion_mode": { "type": "string", "const": "ohip_data" },
"business_date": { "type": ["string", "null"], "format": "date" },
- "processor_version": { "type": "string", "const": "4.2.0" },
+ "processor_version": { "type": "string", "const": "4.3.0" },
"rule_set_sha256": { "$ref": "#/$defs/sha256" },
"source_rows": { "type": "integer", "minimum": 0 },
"removed_by_rate_code": { "type": "integer", "minimum": 0 },
@@ -32,10 +32,11 @@
"manually_priced_rows": { "type": "integer", "minimum": 0 },
"outcome_counts": {
"type": "object", "additionalProperties": false,
- "required": ["candidate", "duplicate", "excluded_rate_code", "price_unmatched", "retained", "validation_failed"],
+ "required": ["candidate", "duplicate", "excluded_cancelled", "excluded_rate_code", "price_unmatched", "retained", "validation_failed"],
"properties": {
"candidate": { "type": "integer", "minimum": 0 },
"duplicate": { "type": "integer", "minimum": 0 },
+ "excluded_cancelled": { "type": "integer", "minimum": 0 },
"excluded_rate_code": { "type": "integer", "minimum": 0 },
"price_unmatched": { "type": "integer", "minimum": 0 },
"retained": { "type": "integer", "minimum": 0 },
@@ -147,7 +148,7 @@
"properties": {
"source_sequence": { "type": "integer", "minimum": 1 }, "source_location": { "type": "string", "minLength": 1 },
"source_worksheet": { "type": "null" }, "source_row_no": { "type": "null" },
- "outcome": { "type": "string", "enum": ["retained", "candidate", "excluded_rate_code", "duplicate", "validation_failed", "price_unmatched"] },
+ "outcome": { "type": "string", "enum": ["retained", "candidate", "excluded_rate_code", "duplicate", "validation_failed", "price_unmatched", "excluded_cancelled"] },
"decision_codes": { "type": "array", "uniqueItems": true, "items": { "type": "string", "minLength": 1 } },
"duplicate_of_source_sequence": { "type": ["integer", "null"], "minimum": 1 },
"adults": { "type": ["integer", "null"] }, "children": { "type": ["integer", "null"] }, "block_code": { "type": "string" },
@@ -166,6 +167,14 @@
"pricing_method": { "type": ["string", "null"], "enum": ["zero_price_exception", "price_reference_exact", "manual_review", null] }
},
"allOf": [
+ {
+ "if": { "properties": { "outcome": { "const": "excluded_cancelled" } }, "required": ["outcome"] },
+ "then": { "properties": {
+ "decision_codes": { "const": ["RESERVATION_CANCELLED_EXCLUDED"] },
+ "real_price": { "type": "null" }, "total_price": { "type": "null" },
+ "kb_amount": { "type": "null" }, "channel_key": { "type": "null" }, "pricing_method": { "type": "null" }
+ } }
+ },
{
"if": { "properties": { "outcome": { "const": "duplicate" } }, "required": ["outcome"] },
"then": { "properties": { "duplicate_of_source_sequence": { "type": "integer", "minimum": 1 } } },
diff --git a/arr-opera-daily-ingest/references/field-contracts.md b/arr-opera-daily-ingest/references/field-contracts.md
index 1850d62..09f4de0 100644
--- a/arr-opera-daily-ingest/references/field-contracts.md
+++ b/arr-opera-daily-ingest/references/field-contracts.md
@@ -107,8 +107,11 @@ When trimmed `RES_COMMENT` is empty, `group_code_key` is null and `booking_sourc
The complete record field list and conditional nullability rules are authoritative in [structured-result.schema.json](structured-result.schema.json).
-## Direct OHIP data input (processor 4.2.0)
+## Direct OHIP data input (processor 4.3.0)
`--data-json` accepts frozen `arr-ohip-data/v1` field observations with complete collection, exact hotel/date context, unique reservation IDs and continuous source order. Classification, whitelist, room/arrival deduplication, pricing, zero-price exceptions, nights, channel assignment and formulas are shared with XML. An unresolved field on a retained candidate fails the batch; it is not treated as a missing price. Non-whitelist rows are excluded before business-field validation, as in XML.
The reservation comment uses the first nonempty note in the supplied order after emoji cleanup. PRODUCTS displays package codes in supplied order, separated by comma and space; duplicates are retained. All notes, package schedules/details and source-response references remain unchanged in the immutable source JSON. This is the explicit new data display contract, not a claim of byte-identical Oracle report formatting. Optional confirmed-empty values display blank. Trace is not fetched and its existing workbook column remains blank.
+
+
+An OHIP row may carry a nonempty string `reservation_status` outside the15 business `fields`. New acquisition retains it from independently matched search/detail states. For pre-upgrade pending captures, an internal saved-response evidence receipt can append statuses to a derived source without changing original bytes or staff decisions. Cancelled rows are excluded before candidate validation. A missing status in an old source is not silently interpreted as cancelled.
diff --git a/arr-opera-daily-ingest/references/structured-output.md b/arr-opera-daily-ingest/references/structured-output.md
index bea6000..d261c02 100644
--- a/arr-opera-daily-ingest/references/structured-output.md
+++ b/arr-opera-daily-ingest/references/structured-output.md
@@ -31,7 +31,7 @@ The payload includes:
- `removed_by_rate_code`
- `removed_as_duplicates`
- `output_rows`, `candidate_rows`, `review_required_rows`, and `review_issue_count`
-- six-outcome reconciliation
+- seven-outcome reconciliation (processor4.3.0)
- grouped, privacy-minimized `review_issues` with fixed-table candidate-price comparisons
- `review_case_id`, `manual_override_sha256`, and `manually_priced_rows` on final manual replay only
- channel counts
@@ -90,3 +90,6 @@ It may then expose `booking.booking_source_rows.hotel_raw` as the original booki
ARR must revalidate Schema, hashes and the independent replay before it writes file/version/record/channel/lookup rows in one transaction and switches the current daily version last. Review receipt writes only a case/items/events audit record. A frozen manual replay creates the first Finance version and one commit event only after that final validation succeeds.
Processor 4.2.0 adds explicit direct-data schema 5.0 (`ohip_data`, `source_data`, `ohip_json`, application/json). XML stays schema 4.0 and frozen direct-MCP stays 3.0. Manual overrides bind the original source SHA-256 for either source type.
+
+
+Processor4.3.0 preserves result schema4.0/5.0 and adds `excluded_cancelled` to `outcome_counts` and record outcomes. It requires `RESERVATION_CANCELLED_EXCLUDED` with no price/channel facts; all source rows still reconcile. Old processor results retain their original outcome contract; the new outcome cannot be declared under an old processor identity. Finance requires migration020.
diff --git a/arr-opera-daily-ingest/references/structured-result.schema.json b/arr-opera-daily-ingest/references/structured-result.schema.json
index 22ac7a7..e531d23 100644
--- a/arr-opera-daily-ingest/references/structured-result.schema.json
+++ b/arr-opera-daily-ingest/references/structured-result.schema.json
@@ -17,7 +17,7 @@
"activation_eligible": { "type": "boolean" },
"ingestion_mode": { "type": "string", "const": "opera_xml" },
"business_date": { "type": ["string", "null"], "format": "date" },
- "processor_version": { "type": "string", "const": "4.2.0" },
+ "processor_version": { "type": "string", "const": "4.3.0" },
"rule_set_sha256": { "$ref": "#/$defs/sha256" },
"source_rows": { "type": "integer", "minimum": 0 },
"removed_by_rate_code": { "type": "integer", "minimum": 0 },
@@ -32,10 +32,11 @@
"manually_priced_rows": { "type": "integer", "minimum": 0 },
"outcome_counts": {
"type": "object", "additionalProperties": false,
- "required": ["candidate", "duplicate", "excluded_rate_code", "price_unmatched", "retained", "validation_failed"],
+ "required": ["candidate", "duplicate", "excluded_cancelled", "excluded_rate_code", "price_unmatched", "retained", "validation_failed"],
"properties": {
"candidate": { "type": "integer", "minimum": 0 },
"duplicate": { "type": "integer", "minimum": 0 },
+ "excluded_cancelled": { "type": "integer", "minimum": 0 },
"excluded_rate_code": { "type": "integer", "minimum": 0 },
"price_unmatched": { "type": "integer", "minimum": 0 },
"retained": { "type": "integer", "minimum": 0 },
@@ -147,7 +148,7 @@
"properties": {
"source_sequence": { "type": "integer", "minimum": 1 }, "source_location": { "type": "string", "minLength": 1 },
"source_worksheet": { "type": "null" }, "source_row_no": { "type": "null" },
- "outcome": { "type": "string", "enum": ["retained", "candidate", "excluded_rate_code", "duplicate", "validation_failed", "price_unmatched"] },
+ "outcome": { "type": "string", "enum": ["retained", "candidate", "excluded_rate_code", "duplicate", "validation_failed", "price_unmatched", "excluded_cancelled"] },
"decision_codes": { "type": "array", "uniqueItems": true, "items": { "type": "string", "minLength": 1 } },
"duplicate_of_source_sequence": { "type": ["integer", "null"], "minimum": 1 },
"adults": { "type": ["integer", "null"] }, "children": { "type": ["integer", "null"] }, "block_code": { "type": "string" },
@@ -166,6 +167,14 @@
"pricing_method": { "type": ["string", "null"], "enum": ["zero_price_exception", "price_reference_exact", "manual_review", null] }
},
"allOf": [
+ {
+ "if": { "properties": { "outcome": { "const": "excluded_cancelled" } }, "required": ["outcome"] },
+ "then": { "properties": {
+ "decision_codes": { "const": ["RESERVATION_CANCELLED_EXCLUDED"] },
+ "real_price": { "type": "null" }, "total_price": { "type": "null" },
+ "kb_amount": { "type": "null" }, "channel_key": { "type": "null" }, "pricing_method": { "type": "null" }
+ } }
+ },
{
"if": { "properties": { "outcome": { "const": "duplicate" } }, "required": ["outcome"] },
"then": { "properties": { "duplicate_of_source_sequence": { "type": "integer", "minimum": 1 } } },
diff --git a/arr-opera-daily-ingest/scripts/process_daily.py b/arr-opera-daily-ingest/scripts/process_daily.py
index 94278ba..cbf4026 100755
--- a/arr-opera-daily-ingest/scripts/process_daily.py
+++ b/arr-opera-daily-ingest/scripts/process_daily.py
@@ -27,7 +27,7 @@ from openpyxl.utils import get_column_letter
RESULT_VERSION = "4.0"
-PROCESSOR_VERSION = "4.2.0"
+PROCESSOR_VERSION = "4.3.0"
DATA_RESULT_VERSION = "5.0"
STRUCTURED_RESULT_SCHEMA_VERSION = "4.0"
# Retired direct-MCP/callback compatibility. This is deliberately an internal,
@@ -65,6 +65,7 @@ RULE_SET_PATHS = (
FINAL_OUTCOMES = {
"retained",
+ "excluded_cancelled",
"excluded_rate_code",
"duplicate",
"validation_failed",
@@ -78,6 +79,8 @@ LEGACY_DIRECT_FINAL_OUTCOMES = {
"validation_failed",
"price_unmatched",
}
+CANCELLED_RESERVATION_STATUSES = frozenset({"CXL", "CANCELLED", "CANCELED"})
+XML_RESERVATION_STATUS_FIELDS = ("RESV_STATUS", "RESERVATION_STATUS", "SHORT_RESV_STATUS")
DAILY_HEADERS = [
"BLOCK_CODE",
@@ -813,6 +816,9 @@ def read_data_source(path: Path, *, input_cleanup=None) -> Tuple[date, List[Dict
or not isinstance(row.get("fields"), dict) or set(row["fields"]) != fields):
invalid("接口预订身份、顺序或字段不完整")
ids.add(row["reservation_id"])
+ reservation_status = row.get("reservation_status", "")
+ if "reservation_status" in row and (not isinstance(reservation_status, str) or not reservation_status.strip()):
+ invalid("接口预订状态必须为非空文本")
values, gaps = {}, []
for key, observation in row["fields"].items():
if not isinstance(observation, dict) or observation.get("state") not in {
@@ -860,6 +866,7 @@ def read_data_source(path: Path, *, input_cleanup=None) -> Tuple[date, List[Dict
record = normalize_source_record(values, index)
record["_DATA_GAPS"] = gaps
record["_SOURCE_KIND"] = "ohip_data"
+ record["_RESERVATION_STATUS"] = reservation_status.strip()
records.append(record)
if input_cleanup is not None:
input_cleanup["ignored_emoji_count"] = ignored
@@ -872,10 +879,10 @@ def read_source(path: Path, *, input_cleanup=None, ignore_emojis=True):
return read_xml(path, input_cleanup=input_cleanup, ignore_emojis=ignore_emojis)
-def classify_input_records(reservations, business_date):
+def classify_input_records(reservations, business_date, *, apply_scope=True):
if reservations and isinstance(reservations[0], dict):
- return classify_normalized_records(reservations, business_date)
- return classify_source_records(reservations, business_date)
+ return classify_normalized_records(reservations, business_date, apply_scope=apply_scope)
+ return classify_source_records(reservations, business_date, apply_scope=apply_scope)
def required_text(record: ET.Element, tag: str, index: int, code: str) -> str:
@@ -917,7 +924,17 @@ def source_record(node: ET.Element, index: int) -> Dict[str, Any]:
DEPARTURE=node.findtext("TRUNC_END"),
RES_COMMENT=first_nonempty(node.findall("./LIST_G_COMMENT_RESV_NAME_ID/G_COMMENT_RESV_NAME_ID/RES_COMMENT")),
TRACE_TEXT=first_nonempty(node.findall("./LIST_G_DEPT_ID/G_DEPT_ID/TRACE_TEXT")))
- return normalize_source_record(fields, index)
+ # Reservation type (e.g. CA/TA/GC) is not a cancellation signal. Only these
+ # explicit status fields and the approved exact cancellation values count.
+ statuses = [text_or_blank(status.text) for tag in XML_RESERVATION_STATUS_FIELDS
+ for status in node.findall(tag) if text_or_blank(status.text)]
+ fields["RESERVATION_STATUS"] = statuses[0] if statuses else ""
+ record = normalize_source_record(fields, index)
+ normalized_statuses = {status.upper() for status in statuses}
+ record["_RESERVATION_STATUS_CONFLICT"] = bool(
+ normalized_statuses & CANCELLED_RESERVATION_STATUSES
+ and normalized_statuses - CANCELLED_RESERVATION_STATUSES)
+ return record
def normalize_source_record(values: Mapping[str, Any], index: int) -> Dict[str, Any]:
@@ -962,6 +979,8 @@ def normalize_source_record(values: Mapping[str, Any], index: int) -> Dict[str,
"_SOURCE_ROW_NO": None,
"_RAW_RATE_CODE": raw_rate_code,
"_NORMALIZED_RATE_CODE": raw_rate_code.upper(),
+ "_RESERVATION_STATUS": text_or_blank(values.get("RESERVATION_STATUS")),
+ "_RESERVATION_STATUS_CONFLICT": False,
"_GROUP_CODE_KEY": normalize_group_code(res_comment),
"_COMPANY_KEY": company_key,
"_OUTCOME": "pending",
@@ -1082,19 +1101,29 @@ def build_record(node: ET.Element, index: int, business_date: date, rate_code: s
def classify_source_records(
- reservations: Sequence[ET.Element], business_date: date
+ reservations: Sequence[ET.Element], business_date: date, *, apply_scope: bool = True
) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]], int, int, List[ErrorItem]]:
"""Classify every XML reservation and retain source order and duplicate lineage."""
all_records = [source_record(node, index) for index, node in enumerate(reservations, 1)]
- return classify_normalized_records(all_records, business_date)
+ return classify_normalized_records(all_records, business_date, apply_scope=apply_scope)
-def classify_normalized_records(all_records, business_date):
+def is_cancelled_record(record: Mapping[str, Any]) -> bool:
+ """Recognize explicit cancellation only; absent, unknown or conflicting status stays in scope."""
+ return (not record.get("_RESERVATION_STATUS_CONFLICT", False)
+ and text_or_blank(record.get("_RESERVATION_STATUS")).upper() in CANCELLED_RESERVATION_STATUSES)
+
+
+def classify_normalized_records(all_records, business_date, *, apply_scope=True):
candidates: List[Dict[str, Any]] = []
errors: List[ErrorItem] = []
removed_by_rate = 0
for record in all_records:
+ if apply_scope and is_cancelled_record(record):
+ record["_OUTCOME"] = "excluded_cancelled"
+ append_decision(record, "RESERVATION_CANCELLED_EXCLUDED")
+ continue
normalized_rate = record["_NORMALIZED_RATE_CODE"]
if not normalized_rate:
error = xml_record_error(
@@ -2048,7 +2077,7 @@ def validate_structured_completeness(payload: Dict[str, Any]) -> None:
failures.append("source_sequence必须从1开始连续且保持XML顺序")
counts = payload.get("outcome_counts", {})
if not isinstance(counts, dict) or set(counts) != FINAL_OUTCOMES:
- failures.append("outcome_counts必须覆盖六种固定outcome")
+ failures.append("outcome_counts必须覆盖七种固定outcome")
counts = {}
for outcome in FINAL_OUTCOMES:
actual = sum(1 for record in records if record.get("outcome") == outcome)
@@ -2074,6 +2103,7 @@ def validate_structured_completeness(payload: Dict[str, Any]) -> None:
payload.get("removed_by_rate_code", 0)
+ payload.get("removed_as_duplicates", 0)
+ payload.get("output_rows", 0)
+ + counts.get("excluded_cancelled", 0)
):
failures.append("成功payload的源行分解不平衡")
if payload.get("review_required_rows") or payload.get("review_issue_count"):
@@ -2091,7 +2121,7 @@ def validate_structured_completeness(payload: Dict[str, Any]) -> None:
failures.append("待复核payload必须包含缺价行")
if payload.get("candidate_rows", 0) + payload.get("review_required_rows", 0) + counts.get(
"excluded_rate_code", 0
- ) + counts.get("duplicate", 0) != source_rows:
+ ) + counts.get("duplicate", 0) + counts.get("excluded_cancelled", 0) != source_rows:
failures.append("待复核payload的源行分解不平衡")
issues = payload.get("review_issues")
if not isinstance(issues, list) or payload.get("review_issue_count") != len(issues):
@@ -2182,6 +2212,12 @@ def validate_structured_completeness(payload: Dict[str, Any]) -> None:
or len(decisions) != len(dict.fromkeys(decisions))
):
failures.append("记录decision_codes必须是非空且不重复的数组")
+ if record.get("outcome") == "excluded_cancelled":
+ if decisions != ["RESERVATION_CANCELLED_EXCLUDED"]:
+ failures.append("取消排除记录必须保留明确的取消排除原因")
+ if any(record.get(field) is not None for field in (
+ "real_price", "total_price", "kb_amount", "channel_key", "pricing_method")):
+ failures.append("取消排除记录不得参与定价或渠道归属")
if record.get("normalized_rate_code") != (
text_or_blank(record.get("rate_code")).upper() or None
):
@@ -2426,7 +2462,7 @@ def process(args: argparse.Namespace) -> int:
removed_duplicates,
classification_errors,
) = classify_input_records(
- reservations, business_date
+ reservations, business_date, apply_scope=not legacy_v3_output
)
metrics["removed_by_rate_code"] = removed_rate
metrics["removed_as_duplicates"] = removed_duplicates
diff --git a/arr-opera-daily-ingest/scripts/validate_daily.py b/arr-opera-daily-ingest/scripts/validate_daily.py
index e68e6b8..9693d0c 100755
--- a/arr-opera-daily-ingest/scripts/validate_daily.py
+++ b/arr-opera-daily-ingest/scripts/validate_daily.py
@@ -457,6 +457,7 @@ def validate_structured_result_contract(
manual_manifest: Optional[core.ManualOverrideManifest] = None,
manual_override_path: Optional[Path] = None,
ignore_emojis: bool = True,
+ apply_scope: bool = True,
) -> None:
required = {
"result_schema_version",
@@ -536,6 +537,7 @@ def validate_structured_result_contract(
)
expected_outcome_counts = {
"duplicate": removed_duplicates,
+ "excluded_cancelled": source_rows - removed_by_rate - removed_duplicates - len(expected_records),
"excluded_rate_code": removed_by_rate,
"candidate": 0,
"price_unmatched": 0,
@@ -609,7 +611,7 @@ def validate_structured_result_contract(
_date, reservations = core.read_source(xml_path, ignore_emojis=ignore_emojis)
all_records, retained, _removed_rate, _removed_duplicates, classification_errors = (
- core.classify_input_records(reservations, business_date)
+ core.classify_input_records(reservations, business_date, apply_scope=apply_scope)
)
if classification_errors:
errors.append(
@@ -763,6 +765,7 @@ def validate_review_structured_result_contract(
)
expected_outcome_counts = {
"candidate": candidate_rows,
+ "excluded_cancelled": sum(record.get("_OUTCOME") == "excluded_cancelled" for record in all_records),
"duplicate": removed_duplicates,
"excluded_rate_code": removed_by_rate,
"price_unmatched": review_rows,
@@ -978,6 +981,7 @@ def validate_legacy_direct_success_contracts(
)
replay_outcomes = dict(outcomes)
replay_outcomes["candidate"] = 0
+ replay_outcomes["excluded_cancelled"] = 0
replay_structured["outcome_counts"] = replay_outcomes
replay_artifacts = dict(artifacts)
replay_artifacts["manual_override_json"] = None
@@ -1008,6 +1012,7 @@ def validate_legacy_direct_success_contracts(
replay_result,
errors,
ignore_emojis=False,
+ apply_scope=False,
)
@@ -1201,7 +1206,7 @@ def validate(args: argparse.Namespace) -> List[core.ErrorItem]:
xml_path, input_cleanup=input_cleanup, ignore_emojis=not legacy_v3_output
)
all_records, records, removed_rate, removed_duplicates, classification_errors = (
- core.classify_input_records(reservations, business_date)
+ core.classify_input_records(reservations, business_date, apply_scope=not legacy_v3_output)
)
if classification_errors:
raise core.ProcessingFailure(classification_errors)
diff --git a/arr_ingestion/direct_postgres.py b/arr_ingestion/direct_postgres.py
index 7de1d2b..3e39091 100644
--- a/arr_ingestion/direct_postgres.py
+++ b/arr_ingestion/direct_postgres.py
@@ -756,12 +756,13 @@ class PostgresDirectIngestionRepository(PostgresIngestionRepository):
duplicate_rows,
validation_failed_rows,
price_unmatched_rows,
+ excluded_cancelled_rows,
validated_at,
result_delivery_mode
)
VALUES (
%s, %s, %s, %s, 'validated', %s, %s, %s, %s,
- %s, %s, %s, %s, %s, %s, now(), 'direct_mcp'
+ %s, %s, %s, %s, %s, %s, %s, now(), 'direct_mcp'
)
RETURNING id
""",
diff --git a/arr_ingestion/postgres.py b/arr_ingestion/postgres.py
index 41bdb15..d72bd19 100644
--- a/arr_ingestion/postgres.py
+++ b/arr_ingestion/postgres.py
@@ -158,7 +158,7 @@ def _is_transient_database_error(error: Exception) -> bool:
return isinstance(sqlstate, str) and sqlstate in TRANSIENT_SQLSTATES
-def _outcome_counts(payload: Mapping[str, Any]) -> Tuple[int, int, int, int, int]:
+def _outcome_counts(payload: Mapping[str, Any]) -> Tuple[int, int, int, int, int, int]:
values = payload.get("outcome_counts")
if not isinstance(values, Mapping):
raise IngestionError(
@@ -167,13 +167,14 @@ def _outcome_counts(payload: Mapping[str, Any]) -> Tuple[int, int, int, int, int
)
try:
counts = tuple(
- int(values[name])
+ int(values.get(name, 0) if name == "excluded_cancelled" else values[name])
for name in (
"retained",
"excluded_rate_code",
"duplicate",
"validation_failed",
"price_unmatched",
+ "excluded_cancelled",
)
)
except (KeyError, TypeError, ValueError):
@@ -337,6 +338,32 @@ class PostgresIngestionRepository(IngestionRepository):
if any(value not in definitions.get(key, "") for key, value in expected.items()):
raise IngestionError("DATABASE_MIGRATION_MISSING", "direct data entry requires migration 019")
self._run_transaction(check, "direct data schema could not be checked")
+ self.assert_cancelled_scope_schema()
+
+ def assert_cancelled_scope_schema(self) -> None:
+ """Require the count and outcome constraints before either input is enabled."""
+ def check(cursor):
+ cursor.execute("""SELECT EXISTS (
+ SELECT 1 FROM information_schema.columns
+ WHERE table_schema = 'finance' AND table_name = 'daily_versions'
+ AND column_name = 'excluded_cancelled_rows'
+ AND data_type = 'integer' AND is_nullable = 'NO'
+ ), (
+ SELECT pg_get_constraintdef(oid) FROM pg_constraint
+ WHERE conrelid = 'finance.daily_versions'::regclass
+ AND conname = 'daily_versions_counts_reconcile'
+ ), (
+ SELECT pg_get_constraintdef(oid) FROM pg_constraint
+ WHERE conrelid = 'finance.daily_records'::regclass
+ AND conname = 'daily_records_outcome_check'
+ )""")
+ row = cursor.fetchone()
+ if (not row or row[0] is not True
+ or "excluded_cancelled_rows" not in str(row[1] or "")
+ or "excluded_cancelled" not in str(row[2] or "")):
+ raise IngestionError("DATABASE_MIGRATION_MISSING",
+ "ARR processing requires migration 020")
+ self._run_transaction(check, "ARR cancellation schema could not be checked")
def register_job(self, registration: JobRegistration) -> None:
if (
@@ -1809,13 +1836,14 @@ class PostgresIngestionRepository(IngestionRepository):
duplicate_rows,
validation_failed_rows,
price_unmatched_rows,
+ excluded_cancelled_rows,
failure_code,
failure_message,
validated_at
)
VALUES (
NULL, NULL, %s, %s, %s, %s, %s, 'rejected',
- %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
+ %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
%s, 'deterministic processing failed', now()
)
RETURNING id
@@ -2047,11 +2075,12 @@ class PostgresIngestionRepository(IngestionRepository):
duplicate_rows,
validation_failed_rows,
price_unmatched_rows,
+ excluded_cancelled_rows,
validated_at
)
VALUES (
%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 'validated',
- %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, now()
+ %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, now()
)
RETURNING id
""",
diff --git a/arr_ingestion/validation.py b/arr_ingestion/validation.py
index fea6d22..6f454e9 100644
--- a/arr_ingestion/validation.py
+++ b/arr_ingestion/validation.py
@@ -483,6 +483,18 @@ V4_OUTCOME_FIELDS = {
"retained",
"validation_failed",
}
+CANCELLED_OUTCOME_PROCESSOR_VERSIONS = {"4.3.0"}
+
+
+def _v4_outcome_fields(processor_version: str) -> set[str]:
+ # Schema 4/5 remain readable for older immutable deliveries. The new
+ # outcome is part of the explicitly approved 4.3 rule identity only; the
+ # delivery validator also binds that identity to the approved rule hash.
+ if processor_version in CANCELLED_OUTCOME_PROCESSOR_VERSIONS:
+ return V4_OUTCOME_FIELDS | {"excluded_cancelled"}
+ return V4_OUTCOME_FIELDS
+
+
V4_RESULT_METRIC_FIELDS = {
"source_rows",
"removed_by_rate_code",
@@ -519,7 +531,8 @@ def _validate_v4_records(payload: Mapping[str, Any]) -> Tuple[Dict[str, int], in
records = payload.get("records")
if not isinstance(records, list) or len(records) != source_rows:
raise IngestionError("RESULT_CONTRACT_INVALID", "structured record count is invalid")
- actual_outcomes = {outcome: 0 for outcome in V4_OUTCOME_FIELDS}
+ outcome_fields = _v4_outcome_fields(str(payload.get("processor_version")))
+ actual_outcomes = {outcome: 0 for outcome in outcome_fields}
manual_rows = 0
for index, record in enumerate(records, 1):
values = _require_exact_mapping(record, RECORD_FIELDS, "structured record")
@@ -528,7 +541,7 @@ def _validate_v4_records(payload: Mapping[str, Any]) -> Tuple[Dict[str, int], in
if values.get("source_worksheet") is not None or values.get("source_row_no") is not None:
raise IngestionError("RESULT_CONTRACT_INVALID", "direct XML source coordinates are invalid")
outcome = values.get("outcome")
- if outcome not in V4_OUTCOME_FIELDS:
+ if outcome not in outcome_fields:
raise IngestionError("RESULT_CONTRACT_INVALID", "structured outcome is invalid")
actual_outcomes[str(outcome)] += 1
duplicate_of = values.get("duplicate_of_source_sequence")
@@ -635,8 +648,9 @@ def _validate_structured_payload_v4(
"manually_priced_rows",
)
}
- outcomes = _require_exact_mapping(payload.get("outcome_counts"), V4_OUTCOME_FIELDS, "outcome counts")
- for outcome in V4_OUTCOME_FIELDS:
+ outcome_fields = _v4_outcome_fields(envelope.processor_version)
+ outcomes = _require_exact_mapping(payload.get("outcome_counts"), outcome_fields, "outcome counts")
+ for outcome in outcome_fields:
_require_nonnegative_integer(outcomes.get(outcome), f"outcome_counts.{outcome}")
actual_outcomes, manual_rows = _validate_v4_records(payload)
if dict(outcomes) != actual_outcomes:
@@ -694,6 +708,7 @@ def _validate_structured_payload_v4(
or outcomes.get("candidate") != 0
or counts["source_rows"]
!= counts["removed_by_rate_code"] + counts["removed_as_duplicates"] + counts["output_rows"]
+ + outcomes.get("excluded_cancelled", 0)
or channel_rows != counts["output_rows"]
or review_keys
):
@@ -721,6 +736,7 @@ def _validate_structured_payload_v4(
+ counts["removed_as_duplicates"]
+ counts["candidate_rows"]
+ counts["review_required_rows"]
+ + outcomes.get("excluded_cancelled", 0)
or channels
or not review_keys
or manual_ref is not None
diff --git a/arr_web/DIRECT_DATA_ENTRY.md b/arr_web/DIRECT_DATA_ENTRY.md
index bab6418..36de111 100644
--- a/arr_web/DIRECT_DATA_ENTRY.md
+++ b/arr_web/DIRECT_DATA_ENTRY.md
@@ -27,7 +27,7 @@
## 字段完善的保存与继续处理
`DataFieldReviews`只接受`collection_complete=true`的完整采集，以当前冻结处理器的费率白名单和字段验证识别阻塞项。
-预订列表仍保留全部取得的记录及原顺序；非白名单记录按原规则处理，不因Cancelled/NoShow等状态在取数阶段被丢弃。
+预订列表仍保留全部取得的记录及原顺序；非白名单记录按原规则处理，不因Cancelled/NoShow等状态在取数阶段被丢弃。处理阶段按用户已确认的规则排除已取消预订，再识别其余候选的待完善字段；NoShow没有新增排除规则。
费率本身未确定时先完善费率，再重新识别该记录的其余问题。查询失败或采集不完整仍走原重试流程。
人工仅能完善任务列出的异常字段。团队代码、预订备注、套餐和房型允许明确“确认无此项”；
@@ -112,7 +112,7 @@ LC_ALL=C LANG=C .venv/bin/python -m arr_web.local_ohip serve \
## 正常运行配置
-数据库须已应用019迁移；2026-09-18首次验证只在独立本机测试库应用，未修改当时既有数据库。
+数据库须已应用020迁移；2026-09-18首次验证只在独立本机测试库应用，未修改当时既有数据库。
该迁移允许真实的 `ohip_json` 源制品和5.0结果进入已有复核流程。
启动会先检查迁移，未具备条件时不开放下载服务。原来的月报后台程序仍须正常运行。
@@ -125,8 +125,7 @@ LC_ALL=C LANG=C .venv/bin/python -m arr_web.local_ohip serve \
以上三个 OHIP 参数必须一起提供；数据库、对象存储、网页登录沿用现有私有配置。
凭据文件仅在实际点击获取数据时读取，不在启动时查询酒店。
-私有任务目录须在仓库之外，并绑定酒店、来源类型、查询上限和处理规则；规则更换后使用新目录，
-不将旧任务悄悄改绑。旧版本未完成的价格复核遵循原有“规则已变更”保护。
+私有任务目录须在仓库之外，并绑定酒店、来源类型、查询上限和处理规则；规则更换后通常使用新目录；同环境尚未进入处理且未冻结的待完善任务，可在停服务、备份和逐文件升级收据保护下显式接续，不能悄悄改绑。旧版本未完成的价格复核遵循原有“规则已变更”保护。
也支持 `DirectARRSource` 注入共享运行依赖。原有 `CapturedARRSource` 为历史 XML 模式保留，
它不再是新数据入口的前置条件。不得把模拟传输当成实际平台传输。
@@ -142,10 +141,17 @@ ARR_TEST_LOCAL_POSTGRES=1 .venv/bin/python -m unittest tests.test_arr_direct_dat
```
检查覆盖6笔正常订单总额18200、字段完善后继续处理、既有缺价复核、必填及可空字段验证、去重、原始数据保留、独立报表核对、
-字段修订冲突与冻结恢复、页面请求、月报公式、提交后断线恢复、来源版本隔离，以及019迁移检查和回退保护。
+字段修订冲突与冻结恢复、页面请求、月报公式、提交后断线恢复、来源版本隔离，以及019/020迁移检查和回退保护。
浏览器检查：`tests/browser/arr_direct_data.cjs`，只接受明确指定的本机模拟实例。
历史8875沙箱曾交给用户会话后台服务运行，避免对话结束后登录页无法连接。仅在用户会话运行；
该历史OHIP沙箱的停止命令为 `launchctl bootout gui/$(id -u)/com.arr.local-ohip.8875`。服务配置位于私有实例的
`service.plist`；重新加载使用 `launchctl bootstrap gui/$(id -u) /absolute/private/instance/service.plist`。
后台环境已设 `LC_ALL=C`、`LANG=C`。先停后台服务再使用前台serve命令，不同时启动两份。
+
+
+## 已取消预订（2026-10-08确认）
+
+日报排除已取消预订，无论是否已有房号。取消单不要求补房号或其他报表字段，不参与去重、定价及月报；完整原始数据和排除原因保留。
+接口使用已核对一致的查询与详情预订状态；旧待完善任务可由内部维护操作从原始完整捕获补充状态，不重新查询Oracle，不修改原始数据或人工决定。XML使用明确的取消状态值CXL/CANCELLED/CANCELED；CA等未确认缩写不猜测。
+处理器版本4.3.0，Finance需先应用`database/020_daily_cancelled_exclusion.sql`。旧已发布结果不重新处理。
diff --git a/arr_web/arr_data_review.py b/arr_web/arr_data_review.py
index db47da2..35016ad 100644
--- a/arr_web/arr_data_review.py
+++ b/arr_web/arr_data_review.py
@@ -30,6 +30,7 @@ from integrations.ohip.collect_arr_source import CollectionError
VERSION = "arr-source-field-review/v1"
REANALYSIS_POLICY = "oracle-optional-association/v1"
REANALYSIS_VERSION = "arr-source-reanalysis/v1"
+STATUS_POLICY = "arr-exclude-cancelled/v1"
LIMIT = 100 * 1024 * 1024
OPTIONAL = frozenset({"BLOCK_CODE", "RES_COMMENT", "PRODUCTS", "ROOM_CATEGORY_LABEL"})
LABELS = {
@@ -136,6 +137,7 @@ class DataFieldReviews:
and reference.get("source_manifest_sha256") == meta["source_manifest_sha256"],
"data_review_reanalysis_context_changed")
self._reanalyzed_source(directory, original, state)
+ self._status_evidence(directory, meta, original, state)
return meta, original, state
def _publish(self, directory, state):
@@ -221,9 +223,47 @@ class DataFieldReviews:
require(fingerprint(receipt) == fingerprint(expected_receipt), "data_review_reanalysis_receipt_changed")
return reanalyzed, receipt
+ def _status_evidence(self, directory, meta, original, state):
+ reference = state.get("reservation_status_evidence")
+ if reference is None:
+ return None
+ require(type(reference) is dict and reference.get("policy_id") == STATUS_POLICY,
+ "data_review_status_reference_invalid")
+ digest = reference.get("sha256", "")
+ require(type(digest) is str and bool(re.fullmatch(r"[0-9a-f]{64}", digest))
+ and reference.get("file") == f"reservation-status-{digest}.json",
+ "data_review_status_reference_invalid")
+ raw = protected_read(directory / reference["file"], LIMIT)
+ require(_hash(raw) == digest, "data_review_status_evidence_changed")
+ evidence = strict_json(raw)
+ require(evidence.get("version") == "arr-reservation-status-evidence/v1"
+ and evidence.get("policy_id") == STATUS_POLICY
+ and evidence.get("original_sha256") == meta["original_sha256"]
+ and evidence.get("source_manifest_sha256") == meta["source_manifest_sha256"]
+ and evidence.get("report_date") == meta["report_date"]
+ and evidence.get("hotel_id") == original["hotel_id"]
+ and type(evidence.get("records")) is list
+ and len(evidence["records"]) == len(original["records"]),
+ "data_review_status_context_changed")
+ for row, status in zip(original["records"], evidence["records"]):
+ require(type(status) is dict and status.get("source_sequence") == row["source_sequence"]
+ and status.get("reservation_id") == row["reservation_id"]
+ and type(status.get("reservation_status")) is str
+ and bool(status["reservation_status"].strip())
+ and ("reservation_status" not in row
+ or row["reservation_status"] == status["reservation_status"])
+ and type(status.get("sources")) is list and bool(status["sources"]),
+ "data_review_status_records_changed")
+ return evidence
+
def _derive(self, directory, original, state):
source, _receipt = self._reanalyzed_source(directory, original, state)
data = copy.deepcopy(source)
+ meta = strict_json(protected_read(directory / "identity.json", 65536))
+ evidence = self._status_evidence(directory, meta, original, state)
+ if evidence is not None:
+ for row, status in zip(data["records"], evidence["records"]):
+ row["reservation_status"] = status["reservation_status"]
for item_id, decision in state["decisions"].items():
sequence, field = item_id.split(":")
row = data["records"][int(sequence) - 1]
@@ -246,6 +286,8 @@ class DataFieldReviews:
business_date, records = self.rules.read_data_source(Path(stream.name))
issues = {}
for row, record in zip(data["records"], records):
+ if self.rules.is_cancelled_record(record):
+ continue
rate = record["_NORMALIZED_RATE_CODE"]
fields = {}
if not rate or "RATE_CODE" in record["_DATA_GAPS"]:
@@ -273,7 +315,10 @@ class DataFieldReviews:
def _public(self, directory, meta, original, state):
data = self._derive(directory, original, state)
issues = self._issues(directory, data)
- keys = set(issues) | set(state["decisions"])
+ cancelled_sequences = {row["source_sequence"] for row in data["records"]
+ if self.rules.is_cancelled_record({"_RESERVATION_STATUS": row.get("reservation_status", "")})}
+ keys = set(issues) | {key for key in state["decisions"]
+ if int(key.split(":")[0]) not in cancelled_sequences}
items = []
for item_id in sorted(keys, key=lambda key: (int(key.split(":")[0]), key.split(":")[1])):
sequence, field = item_id.split(":")
@@ -289,6 +334,7 @@ class DataFieldReviews:
return {"request_id": meta["request_id"], "report_date": meta["report_date"],
"revision": state["revision"], "status": state["status"], "items": items,
"pending_count": len(issues), "total_count": len(items),
+ "excluded_cancelled_count": len(cancelled_sequences),
"can_finalize": not issues and state["status"] == "editing"}
def prepare(self, request_id, payload, manifest_sha256, report_date):
@@ -322,13 +368,48 @@ class DataFieldReviews:
# A complete ordinary source needs no human step or derived source.
# A reanalysis remains an explicit review until finalized, even if
# its validated source interpretation resolves every listed gap.
- return review if review["total_count"] or "source_reanalysis" in state else None
+ return review if review["total_count"] or "source_reanalysis" in state or "reservation_status_evidence" in state else None
def get(self, request_id):
directory = self._directory(request_id)
with self._lock(directory):
return self._public(directory, *self._read(directory))
+ def apply_saved_status_evidence(self, request_id, capture_root, *, expected_revision):
+ """Bind saved search/detail statuses; no HTTP route or new business read.
+
+ Evidence is reconstructed from the complete original capture, never from
+ staff-entered status values. Fields, order and original bytes stay intact.
+ """
+ from integrations.ohip.reservation_status import saved_status_evidence
+ directory = self._directory(request_id)
+ with self._lock(directory):
+ meta, original, state = self._read(directory)
+ if state["status"] != "editing" or (directory / "finalize-intent.json").exists():
+ raise _error("FROZEN", "已确认生成，预订状态不能再补充", 409)
+ evidence = saved_status_evidence(capture_root, meta["source_manifest_sha256"],
+ protected_read(directory / "original.json", LIMIT))
+ raw = json_bytes(evidence)
+ digest = _hash(raw)
+ if type(expected_revision) is not int:
+ self._revision(state, expected_revision)
+ prior = state.get("reservation_status_evidence")
+ if prior is not None:
+ if prior.get("sha256") == digest:
+ return self._public(directory, meta, original, state)
+ raise _error("CONFLICT", "该来源已补充预订状态，请核对现有记录", 409)
+ self._revision(state, expected_revision)
+ reference = {"policy_id": STATUS_POLICY, "file": f"reservation-status-{digest}.json", "sha256": digest}
+ _write_once(directory / reference["file"], raw)
+ state["reservation_status_evidence"] = reference
+ state["revision"] += 1
+ state["events"].append({"revision": state["revision"], "action": "reservation_status_evidence",
+ "policy_id": STATUS_POLICY, "evidence_sha256": digest,
+ "at": datetime.now(timezone.utc).isoformat()})
+ result = self._public(directory, meta, original, state)
+ self._publish(directory, state)
+ return result
+
def apply_source_reanalysis(self, request_id, payload, manifest_sha256, policy_id, *, expected_revision):
"""Attach a validated offline reinterpretation, without changing acquisition or staff decisions.
@@ -441,6 +522,10 @@ class DataFieldReviews:
self._revision(state, revision)
actor = self._actor(actor)
data = self._derive(directory, original, state)
+ sequence = int(item_id.split(":")[0]) if isinstance(item_id, str) and re.fullmatch(r"[1-9][0-9]*:[A-Z_]+", item_id) else 0
+ if 1 <= sequence <= len(data["records"]) and self.rules.is_cancelled_record(
+ {"_RESERVATION_STATUS": data["records"][sequence - 1].get("reservation_status", "")}):
+ raise _error(message="已取消预订不参与日报，无需完善字段")
if item_id not in self._issues(directory, data) and item_id not in state["decisions"]:
raise _error(message="只能完善当前任务列出的异常字段")
sequence, field = item_id.split(":")
@@ -470,6 +555,10 @@ class DataFieldReviews:
_source, receipt = self._reanalyzed_source(directory, original, state)
if receipt is not None:
audit["source_reanalysis"] = {**receipt, "receipt_sha256": state["source_reanalysis"]["receipt_sha256"]}
+ evidence = self._status_evidence(directory, meta, original, state)
+ if evidence is not None:
+ audit["reservation_status_evidence"] = {**evidence,
+ "evidence_sha256": state["reservation_status_evidence"]["sha256"]}
data = self._derive(directory, original, state)
data["manual_data_review"] = {"manifest": audit, "manifest_sha256": fingerprint(audit)}
binding = fingerprint({"source_manifest_sha256": meta["source_manifest_sha256"],
diff --git a/arr_web/local_ohip.py b/arr_web/local_ohip.py
index 71dd6c3..21f5633 100644
--- a/arr_web/local_ohip.py
+++ b/arr_web/local_ohip.py
@@ -215,7 +215,7 @@ def open_portal(root, port):
load_key(credential_file)
access = AccessState(root)
with job_lock(root):
- database = ReplayDatabase(root, schema_version=19)
+ database = ReplayDatabase(root, schema_version=20)
try:
database.start()
def factory(*, root, snapshot, **dependencies):
diff --git a/arr_web/local_replay_database.py b/arr_web/local_replay_database.py
index cb89cf6..788a6b6 100644
--- a/arr_web/local_replay_database.py
+++ b/arr_web/local_replay_database.py
@@ -14,8 +14,8 @@ from integrations.ohip.capture_job import atomic_json, private_directory
class ReplayDatabase:
- def __init__(self, root: Path, *, schema_version: int = 18):
- if schema_version not in (18, 19):
+ def __init__(self, root: Path, *, schema_version: int = 20):
+ if schema_version not in (18, 19, 20):
raise ValueError("unsupported_local_schema")
self.schema_version = schema_version
self.root = root
diff --git a/arr_web/processing_runtime.py b/arr_web/processing_runtime.py
index e957a9e..e5daffc 100644
--- a/arr_web/processing_runtime.py
+++ b/arr_web/processing_runtime.py
@@ -79,6 +79,7 @@ def compose_programmatic_processing(
database_config,
connect=connect,
)
+ repository.assert_cancelled_scope_schema()
policy = load_processor_policy(project_root)
ingestion = IngestionService(DeliveryValidator(object_store, policy), repository)
processor = LocalDailyProcessor(policy)
diff --git a/arr_web/static/app.js b/arr_web/static/app.js
index f483c50..4518f5b 100644
--- a/arr_web/static/app.js
+++ b/arr_web/static/app.js
@@ -682,6 +682,9 @@
$("#arr-data-review-progress").textContent = review ? I18N.t("data_review.progress", {
date: review.report_date, completed: Number(review.total_count) - Number(review.pending_count), total: review.total_count,
}) : I18N.t("data_review.loading");
+ if (Number.isInteger(review?.excluded_cancelled_count) && review.excluded_cancelled_count > 0) {
+ $("#arr-data-review-progress").textContent += ` · ${I18N.t("data_review.excluded_cancelled", { count: review.excluded_cancelled_count })}`;
+ }
$("#arr-data-review-refresh").textContent = I18N.t("data_review.refresh");
$("#arr-data-review-refresh").disabled = busy;
const finalize = $("#arr-data-review-finalize");
@@ -731,6 +734,10 @@
state.arrDataReview = review;
state.arrDataReviewDisconnected = false;
state.arrDataReviewFinalizing = review.status === "finalized";
+ const currentItems = new Set((review.items || []).map((item) => item.item_id));
+ Object.keys(state.arrDataReviewDrafts).forEach((itemId) => {
+ if (!currentItems.has(itemId)) delete state.arrDataReviewDrafts[itemId];
+ });
review.items?.forEach((item) => {
if (item.confirmed && state.arrDataReviewDrafts[item.item_id] === arrDataReviewInputValue(item)) delete state.arrDataReviewDrafts[item.item_id];
});
diff --git a/arr_web/static/i18n.js b/arr_web/static/i18n.js
index cd1a73f..6f5bc69 100644
--- a/arr_web/static/i18n.js
+++ b/arr_web/static/i18n.js
@@ -170,6 +170,7 @@
"data_review.refresh": ["刷新字段", "Refresh fields", "รีเฟรชข้อมูล"],
"data_review.loading": ["正在读取待完善字段…", "Loading fields for review…", "กำลังโหลดข้อมูลที่ต้องตรวจสอบ…"],
"data_review.no_items": ["没有待完善字段。", "No fields need review.", "ไม่มีข้อมูลที่ต้องตรวจสอบ"],
+ "data_review.excluded_cancelled": ["已排除 {count} 笔取消预订，原始数据保留", "{count} cancelled reservations excluded; original data retained", "ไม่รวมการจองที่ยกเลิก {count} รายการ โดยเก็บข้อมูลต้นฉบับไว้"],
"data_review.region": ["待完善报表字段", "Report fields to review", "ข้อมูลรายงานที่ต้องตรวจสอบ"],
"data_review.disconnected": ["字段读取中断，请刷新字段后继续。", "Field connection lost. Refresh the fields to continue.", "การเชื่อมต่อข้อมูลขัดข้อง โปรดรีเฟรชข้อมูลเพื่อดำเนินการต่อ"],
"data_review.saving": ["正在保存并确认字段…", "Saving and confirming field…", "กำลังบันทึกและยืนยันข้อมูล…"],
diff --git a/database/020_daily_cancelled_exclusion.down.sql b/database/020_daily_cancelled_exclusion.down.sql
new file mode 100644
index 0000000..6cbe5e9
--- /dev/null
+++ b/database/020_daily_cancelled_exclusion.down.sql
@@ -0,0 +1,26 @@
+-- Rollback is safe only before any cancellation-exclusion facts are committed.
+BEGIN;
+DO $$
+BEGIN
+ IF current_database() <> 'booking_test' THEN
+ RAISE EXCEPTION 'ARR migration is allowed only in booking_test';
+ END IF;
+ IF EXISTS (SELECT 1 FROM finance.daily_versions WHERE excluded_cancelled_rows <> 0)
+ OR EXISTS (SELECT 1 FROM finance.daily_records WHERE outcome = 'excluded_cancelled') THEN
+ RAISE EXCEPTION 'refusing rollback: cancellation-exclusion facts exist';
+ END IF;
+END $$;
+
+ALTER TABLE finance.daily_records DROP CONSTRAINT daily_records_outcome_check;
+ALTER TABLE finance.daily_records
+ ADD CONSTRAINT daily_records_outcome_check CHECK (outcome IN (
+ 'retained', 'excluded_rate_code', 'duplicate', 'validation_failed', 'price_unmatched'
+ ));
+ALTER TABLE finance.daily_versions DROP CONSTRAINT daily_versions_counts_reconcile;
+ALTER TABLE finance.daily_versions
+ ADD CONSTRAINT daily_versions_counts_reconcile CHECK (
+ source_rows = retained_rows + excluded_rate_code_rows + duplicate_rows
+ + validation_failed_rows + price_unmatched_rows
+ );
+ALTER TABLE finance.daily_versions DROP COLUMN excluded_cancelled_rows;
+COMMIT;
diff --git a/database/020_daily_cancelled_exclusion.sql b/database/020_daily_cancelled_exclusion.sql
new file mode 100644
index 0000000..8532b05
--- /dev/null
+++ b/database/020_daily_cancelled_exclusion.sql
@@ -0,0 +1,43 @@
+-- Add an auditable cancellation exclusion count without changing released facts.
+-- PostgreSQL 15+. Migrations 008 through 019 remain immutable.
+BEGIN;
+DO $$
+BEGIN
+ IF current_database() <> 'booking_test' THEN
+ RAISE EXCEPTION 'ARR migration is allowed only in booking_test';
+ END IF;
+ IF to_regclass('finance.daily_versions') IS NULL
+ OR to_regclass('finance.daily_records') IS NULL
+ OR NOT EXISTS (
+ SELECT 1 FROM pg_constraint
+ WHERE conrelid = 'ingestion.artifacts'::regclass
+ AND conname = 'artifacts_artifact_kind_check'
+ AND pg_get_constraintdef(oid) LIKE '%ohip_json%'
+ ) THEN
+ RAISE EXCEPTION 'ARR migration 019 must be applied first';
+ END IF;
+END $$;
+
+ALTER TABLE finance.daily_versions
+ ADD COLUMN excluded_cancelled_rows integer NOT NULL DEFAULT 0
+ CHECK (excluded_cancelled_rows >= 0);
+
+ALTER TABLE finance.daily_versions
+ DROP CONSTRAINT daily_versions_counts_reconcile;
+ALTER TABLE finance.daily_versions
+ ADD CONSTRAINT daily_versions_counts_reconcile CHECK (
+ source_rows = retained_rows + excluded_rate_code_rows + duplicate_rows
+ + validation_failed_rows + price_unmatched_rows + excluded_cancelled_rows
+ );
+
+ALTER TABLE finance.daily_records
+ DROP CONSTRAINT daily_records_outcome_check;
+ALTER TABLE finance.daily_records
+ ADD CONSTRAINT daily_records_outcome_check CHECK (outcome IN (
+ 'retained', 'excluded_rate_code', 'duplicate', 'validation_failed',
+ 'price_unmatched', 'excluded_cancelled'
+ ));
+
+COMMENT ON COLUMN finance.daily_versions.excluded_cancelled_rows IS
+ 'Source rows explicitly excluded as cancelled under the pinned processor rules; older versions remain zero.';
+COMMIT;
diff --git a/deploy/OHIP_RELEASE_HANDOVER.md b/deploy/OHIP_RELEASE_HANDOVER.md
index 53ae99a..e7cf978 100644
--- a/deploy/OHIP_RELEASE_HANDOVER.md
+++ b/deploy/OHIP_RELEASE_HANDOVER.md
@@ -1,3 +1,5 @@
+2026-10-08补充：用户已确认日报排除已取消预订。当前处理器4.3.0需在019后应用020；Web与处理器同时发布，月报复用保留的日报事实。完整原始数据保留，取消单不补房号。已有未处理任务接续必须备份并显式记录新旧规则身份，不能复用旧身份直接运行新规则。下文2026-09-18部署/测试情况为历史记录，不代表当前本机生产连接状态。
+
# ARR 自动取数交付与生产启用说明
交付日期：2026-09-18。状态：**开发收尾完成，生产配置和真实数据验收待完成。**
diff --git a/integrations/ohip/arr_data.py b/integrations/ohip/arr_data.py
index 96d6d53..aad1048 100644
--- a/integrations/ohip/arr_data.py
+++ b/integrations/ohip/arr_data.py
@@ -375,6 +375,7 @@ def _record(search, detail, reader, sequence):
"DEPARTURE": observe(lambda: dates().departure.isoformat()),
}
return {"source_sequence": sequence, "reservation_id": base.reservation_id(detail),
+ "reservation_status": detail["reservationStatus"],
"fields": values, "related": related}
diff --git a/integrations/ohip/reservation_status.py b/integrations/ohip/reservation_status.py
new file mode 100644
index 0000000..069ee54
--- /dev/null
+++ b/integrations/ohip/reservation_status.py
@@ -0,0 +1,227 @@
+"""Read-only reservation status evidence from a pinned, complete data capture.
+
+This module never creates a transport or changes acquired data. Unknown nonempty
+Oracle status strings remain verbatim evidence; no abbreviations are interpreted.
+"""
+from __future__ import annotations
+
+import hashlib
+import os
+from pathlib import Path
+import re
+import stat
+from urllib.parse import parse_qs, urlsplit
+
+from . import collect_arr_source as base
+from .audit_arr_capture import protected_read
+from .data_client import FETCH, typed_id
+
+
+VERSION = "arr-reservation-status-evidence/v1"
+POLICY_ID = "arr-exclude-cancelled/v1"
+SOURCE_VERSION = "arr-ohip-data/v1"
+MAX_DATA_BYTES = 25 * 1024 * 1024
+FILE_PATTERN = re.compile(r"(?:capture\.json|arr-data\.json|replay-provenance\.json|"
+ r"data-request-[0-9]{6}\.(?:json|meta\.json|response\.bin))")
+REQUEST_PATTERN = re.compile(r"data-request-([0-9]{6})\.json")
+RESPONSE_PATTERN = re.compile(r"data-request-[0-9]{6}\.response\.bin")
+require = base.require
+
+
+def _hash(raw):
+ return hashlib.sha256(raw).hexdigest()
+
+
+def saved_status_evidence(capture_root, expected_manifest_sha256, original_payload: bytes) -> dict:
+ """Verify saved HTTP evidence and bind one unchanged status to every source row."""
+ require(type(expected_manifest_sha256) is str
+ and bool(re.fullmatch(r"[0-9a-f]{64}", expected_manifest_sha256)), "status_evidence_manifest_pin_invalid")
+ require(type(original_payload) is bytes and len(original_payload) <= MAX_DATA_BYTES,
+ "status_evidence_original_invalid")
+ root = Path(capture_root)
+ info = root.lstat()
+ require(stat.S_ISDIR(info.st_mode) and info.st_uid == os.getuid()
+ and stat.S_IMODE(info.st_mode) == 0o700, "status_evidence_directory_unsafe")
+ manifest_raw = protected_read(root / "result.json", base.MAX_MANIFEST_BYTES)
+ require(_hash(manifest_raw) == expected_manifest_sha256, "status_evidence_manifest_changed")
+ manifest = base.strict_json(manifest_raw)
+ require(manifest.get("version") == SOURCE_VERSION and manifest.get("collection_complete") is True
+ and manifest.get("status") in {"collected", "collected_with_gaps"}
+ and manifest.get("error") is None and manifest.get("finance_ready") is False,
+ "status_evidence_capture_incomplete")
+ entries = manifest.get("files")
+ require(type(entries) is list and 2 <= len(entries) <= 300002, "status_evidence_inventory_invalid")
+ inventory, raw_files, total = {}, {}, 0
+ for entry in entries:
+ require(type(entry) is dict and set(entry) == {"name", "bytes", "sha256"},
+ "status_evidence_inventory_invalid")
+ name, size, digest = entry["name"], entry["bytes"], entry["sha256"]
+ require(type(name) is str and bool(FILE_PATTERN.fullmatch(name)) and name not in inventory,
+ "status_evidence_inventory_name_invalid")
+ require(type(size) is int and 0 <= size <= MAX_DATA_BYTES and type(digest) is str
+ and bool(re.fullmatch(r"[0-9a-f]{64}", digest)), "status_evidence_inventory_invalid")
+ total += size
+ require(total <= base.MAX_ARCHIVE_BYTES, "status_evidence_archive_too_large")
+ raw = protected_read(root / name, size)
+ require(len(raw) == size and _hash(raw) == digest, "status_evidence_archive_changed")
+ inventory[name], raw_files[name] = entry, raw
+ require({p.name for p in root.iterdir()} == set(inventory) | {"result.json"},
+ "status_evidence_inventory_changed")
+ require(raw_files.get("arr-data.json") == original_payload
+ and manifest.get("data_sha256") == _hash(original_payload), "status_evidence_original_changed")
+ original = base.strict_json(original_payload)
+ require(original.get("version") == SOURCE_VERSION and original.get("collection_complete") is True
+ and original.get("status") == manifest["status"]
+ and original.get("source_kind") == manifest.get("source_kind"), "status_evidence_original_invalid")
+ require("capture.json" in raw_files, "status_evidence_capture_missing")
+ capture = base.strict_json(raw_files["capture.json"])
+ require(capture.get("version") == SOURCE_VERSION and capture.get("service_url") == base.SERVICE
+ and capture.get("application_id") == base.APPLICATION
+ and capture.get("source_kind") == original.get("source_kind")
+ and capture.get("source_kind") in {"ohip_platform", "test_transport"}, "status_evidence_context_mismatch")
+ require(type(capture.get("options")) is dict
+ and set(capture["options"]) == {"arrival_date", "hotel_id", "page_size", "max_pages", "max_records"},
+ "status_evidence_context_mismatch")
+ options = base.Options(**capture["options"])
+ options.validate()
+ require(original.get("hotel_id") == options.hotel_id and original.get("report_date") == options.arrival_date,
+ "status_evidence_context_mismatch")
+ rows = original.get("records")
+ require(type(rows) is list and 1 <= len(rows) <= options.max_records
+ and type(manifest.get("records")) is int and manifest["records"] == len(rows),
+ "status_evidence_records_invalid")
+ identities = []
+ for sequence, row in enumerate(rows, 1):
+ require(type(row) is dict and type(row.get("source_sequence")) is int
+ and row["source_sequence"] == sequence and type(row.get("sources")) is list
+ and bool(row["sources"]) and type(row.get("fields")) is dict, "status_evidence_records_invalid")
+ identity = row.get("reservation_id")
+ require(type(identity) is str and bool(re.fullmatch(r"[A-Za-z0-9_-]{1,128}", identity))
+ and identity not in identities, "status_evidence_records_invalid")
+ identities.append(identity)
+ refs = row["sources"]
+ require(all(type(ref) is str and bool(RESPONSE_PATTERN.fullmatch(ref)) and ref in inventory for ref in refs)
+ and len(set(refs)) == len(refs), "status_evidence_source_reference_invalid")
+
+ requests = sorted(name for name in inventory if REQUEST_PATTERN.fullmatch(name))
+ require(type(manifest.get("http_attempts")) is int and len(requests) == manifest["http_attempts"]
+ and requests == [f"data-request-{i:06d}.json" for i in range(1, len(requests) + 1)],
+ "status_evidence_request_inventory_invalid")
+ request_prefixes = {name[:-5] for name in requests}
+ require(all(name.rsplit(".", 2)[0] in request_prefixes
+ for name in inventory if name.endswith((".meta.json", ".response.bin"))),
+ "status_evidence_request_inventory_invalid")
+ documents, searches, details, operations = {}, [], {}, {}
+ for request_name in requests:
+ prefix = request_name[:-5]
+ meta_name, response_name = prefix + ".meta.json", prefix + ".response.bin"
+ require(meta_name in raw_files, "status_evidence_request_metadata_missing")
+ request, meta = base.strict_json(raw_files[request_name]), base.strict_json(raw_files[meta_name])
+ operation = request.get("operation_id")
+ operations[response_name] = operation
+ status = meta.get("http_status")
+ if type(status) is not int or not 200 <= status < 300:
+ continue # Retried failures cannot be evidence, but their bytes are still pinned.
+ require(response_name in raw_files and meta.get("error") is None and meta.get("oversized", False) is False,
+ "status_evidence_response_invalid")
+ if operation not in {base.SEARCH, base.DETAIL}:
+ continue
+ envelope = base.strict_json(raw_files[response_name])
+ require(envelope.get("operation_id") == operation and envelope.get("hotel_id") == options.hotel_id
+ and type(envelope.get("oracle_request_id")) is str and bool(envelope["oracle_request_id"].strip())
+ and type(envelope.get("data")) is dict, "status_evidence_response_context_mismatch")
+ if "upstream_status" in envelope:
+ require(type(envelope["upstream_status"]) is int and 200 <= envelope["upstream_status"] < 300,
+ "status_evidence_upstream_failure")
+ base.check_warnings(envelope)
+ item = {"request": request, "data": envelope["data"], "response": response_name,
+ "number": int(REQUEST_PATTERN.fullmatch(request_name)[1])}
+ documents[response_name] = item
+ if operation == base.SEARCH:
+ require(request.get("method") == "POST" and request.get("path") == "/api/v1/reservations/searches"
+ and type(request.get("body")) is dict, "status_evidence_search_request_invalid")
+ searches.append(item)
+ else:
+ require(type(request.get("path")) is str, "status_evidence_detail_request_mismatch")
+ path = urlsplit(request["path"])
+ collection = envelope["data"].get("reservations")
+ reservations = collection.get("reservation") if type(collection) is dict else None
+ require(type(reservations) is list and len(reservations) == 1 and type(reservations[0]) is dict,
+ "status_evidence_detail_ambiguous")
+ detail = reservations[0]
+ identity = base.validate_row(detail, options)
+ require(typed_id(detail.get("reservationIdList"), "Reservation") == identity
+ and request.get("method") == "GET" and request.get("body") is None
+ and not path.scheme and not path.netloc and not path.fragment
+ and path.path == f"/api/v1/reservations/{identity}"
+ and parse_qs(path.query, keep_blank_values=True) == {"fetchInstructions": list(FETCH)},
+ "status_evidence_detail_request_mismatch")
+ require(identity in identities and identity not in details, "status_evidence_detail_ambiguous")
+ item["row"] = detail
+ details[identity] = item
+ # The complete collector performs two identical paginated searches. Recheck
+ # them offline rather than trusting a new interpretation of row membership.
+ rounds = []
+ for item in searches:
+ if item["request"]["body"].get("offset") == 0:
+ rounds.append([])
+ require(bool(rounds), "status_evidence_search_round_invalid")
+ rounds[-1].append(item)
+ require(len(rounds) == 2 and set(details) == set(identities), "status_evidence_search_round_invalid")
+
+ def replay_search(items):
+ class SavedReader:
+ index = 0
+
+ def read(self, operation, *, body):
+ require(self.index < len(items), "status_evidence_search_round_invalid")
+ item = items[self.index]
+ self.index += 1
+ require(operation == base.SEARCH and item["request"]["body"] == body,
+ "status_evidence_search_request_invalid")
+ return item["data"]
+
+ reader = SavedReader()
+ found = base.search_day(reader, options, server_sort=False)
+ require(reader.index == len(items), "status_evidence_search_round_invalid")
+ return found
+
+ initial, final = (replay_search(items) for items in rounds)
+ require(initial == final and [base.reservation_id(row) for row in initial] == identities,
+ "status_evidence_search_changed")
+ result = []
+ for source_row, search in zip(rows, initial):
+ identity = source_row["reservation_id"]
+ detail_item = details[identity]
+ detail = detail_item["row"]
+ require(rounds[0][-1]["number"] < detail_item["number"] < rounds[1][0]["number"],
+ "status_evidence_request_order_invalid")
+ status = search.get("reservationStatus")
+ require(type(status) is str and bool(status.strip()) and status == detail.get("reservationStatus"),
+ "status_evidence_status_conflict")
+ modified = search.get("lastModifyDateTime")
+ require(type(modified) is str and bool(modified.strip()) and modified == detail.get("lastModifyDateTime"),
+ "status_evidence_state_conflict")
+ if "reservation_status" in source_row:
+ require(source_row["reservation_status"] == status, "status_evidence_existing_status_conflict")
+ require(all(operations.get(ref) not in {base.SEARCH, base.DETAIL} or ref in documents
+ for ref in source_row["sources"]), "status_evidence_source_binding_failed")
+ initial_refs = [item["response"] for item in rounds[0]
+ if any(base.reservation_id(row) == identity for row in item["data"]["reservations"]["reservationInfo"])]
+ require(len(initial_refs) == 1 and initial_refs[0] in source_row["sources"]
+ and detail_item["response"] in source_row["sources"], "status_evidence_source_binding_missing")
+ claimed_details = [ref for ref in source_row["sources"]
+ if ref in documents and documents[ref]["request"]["operation_id"] == base.DETAIL]
+ require(claimed_details == [detail_item["response"]], "status_evidence_source_binding_ambiguous")
+ final_refs = [item["response"] for item in rounds[1]
+ if any(base.reservation_id(row) == identity for row in item["data"]["reservations"]["reservationInfo"])]
+ proof_names = []
+ for ref in initial_refs + [detail_item["response"]] + final_refs:
+ prefix = ref.removesuffix(".response.bin")
+ proof_names.extend((prefix + ".json", prefix + ".meta.json", ref))
+ result.append({"source_sequence": source_row["source_sequence"], "reservation_id": identity,
+ "reservation_status": status,
+ "sources": [{"file": name, "sha256": inventory[name]["sha256"]} for name in proof_names]})
+ return {"version": VERSION, "policy_id": POLICY_ID, "original_sha256": _hash(original_payload),
+ "source_manifest_sha256": expected_manifest_sha256, "report_date": options.arrival_date,
+ "hotel_id": options.hotel_id, "records": result}
diff --git a/tests/javascript/arr_data_review.cjs b/tests/javascript/arr_data_review.cjs
index 10859c3..6a0674e 100644
--- a/tests/javascript/arr_data_review.cjs
+++ b/tests/javascript/arr_data_review.cjs
@@ -10,6 +10,31 @@ const item=(extra={})=>({item_id:'1:BLOCK_CODE',source_sequence:1,confirmation_n
const review=(items,revision=1)=>({request_id:requestId,report_date:task.report_date,status:'editing',revision,items,pending_count:items.filter(x=>!x.confirmed).length,total_count:items.length,can_finalize:items.every(x=>x.confirmed)});
const plain=value=>JSON.parse(JSON.stringify(value));
+test('cancelled source rows need no fields and the exclusion stays visible before continuation',async()=>{
+ const h=harness(()=>response(200,{...review([]),excluded_cancelled_count:6}));
+ await h.acceptARRDownloadTask(task,{sync:false});
+ assert.equal(h.element('#arr-data-review-body').innerHTML,'');
+ assert.match(h.element('#arr-data-review-progress').textContent,/data_review\.excluded_cancelled/);
+ 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));
+ await h.acceptARRDownloadTask(task,{sync:false});
+ h.state.arrDataReviewDrafts['1:BLOCK_CODE']='CANCELLED-DRAFT';
+ h.state.arrDataReviewDrafts['2:DISP_ROOM_NO']='ROOM-DRAFT';
+ current=review([current.items[1]],2);
+ await h.loadARRDataReview(requestId);
+ assert.equal(h.state.arrDataReviewDrafts['1:BLOCK_CODE'],undefined);
+ assert.equal(h.state.arrDataReviewDrafts['2:DISP_ROOM_NO'],'ROOM-DRAFT');
+ current={...review([],3),excluded_cancelled_count:2};
+ await h.loadARRDataReview(requestId);
+ assert.equal(Object.keys(h.state.arrDataReviewDrafts).length,0);
+ assert.equal(h.element('#arr-data-review-finalize').disabled,false);
+});
+
for (const fixture of [
{name:'missing room retains the original required rule', field:'DISP_ROOM_NO', can_be_empty:false,
source_state:'missing', reason_code:'calendar_missing_rooms', expected:'room_missing_in_scope'},
diff --git a/tests/local_postgres.py b/tests/local_postgres.py
index d005cf3..f066806 100644
--- a/tests/local_postgres.py
+++ b/tests/local_postgres.py
@@ -76,7 +76,9 @@ class TemporaryPostgres:
connection.close()
raise
- def reset_database(self):
+ def reset_database(self, *, schema_version=20):
+ if schema_version not in (19, 20):
+ 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:
connection.execute("DROP DATABASE booking_test WITH (FORCE)")
@@ -86,8 +88,8 @@ class TemporaryPostgres:
if existing:
raise RuntimeError("schema bootstrap requires an empty fixture database")
paths = sorted(p for p in (PROJECT / "database").glob("[0-9][0-9][0-9]_*.sql")
- if ".down." not in p.name and 8 <= int(p.name[:3]) <= 19)
- if len(paths) != 12:
+ if ".down." not in p.name and 8 <= int(p.name[:3]) <= schema_version)
+ if len(paths) != schema_version - 7:
raise RuntimeError("unexpected fixture migration set")
for path in paths:
connection.execute(path.read_text(), prepare=False)
diff --git a/tests/test_arr_cancelled_ingestion.py b/tests/test_arr_cancelled_ingestion.py
new file mode 100644
index 0000000..a2356ff
--- /dev/null
+++ b/tests/test_arr_cancelled_ingestion.py
@@ -0,0 +1,169 @@
+"""Cancellation outcome contracts and additive Finance migration, synthetic only."""
+from __future__ import annotations
+
+import copy
+from dataclasses import replace
+import json
+import os
+from pathlib import Path
+import tempfile
+import unittest
+
+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 tests.local_postgres import TemporaryPostgres
+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 cancelled_xml(*, rate_amount="900"):
+ cancelled = reservation(1).replace(
+ "SYNTHETIC-ROOM-1", ""
+ ).replace("", "CANCELLED")
+ return xml_document(cancelled, reservation(2, rate_amount=rate_amount))
+
+
+class CancelledIngestionContractTests(unittest.TestCase):
+ def generated(self, root):
+ raw, store, _ = build_delivery(cancelled_xml(), 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_new_outcome_is_validated_and_replayed_without_room(self):
+ with tempfile.TemporaryDirectory() as temporary:
+ raw, store, _, payload = self.generated(Path(temporary))
+ verified = DeliveryValidator(store, policy()).validate(raw)
+ self.assertEqual(verified.envelope.status, "success")
+ self.assertEqual(payload["outcome_counts"]["excluded_cancelled"], 1)
+ self.assertEqual(payload["output_rows"], 1)
+ self.assertEqual(payload["records"][0]["outcome"], "excluded_cancelled")
+ self.assertEqual(payload["records"][0]["disp_room_no"], "")
+ self.assertNotIn("removed_as_cancelled", payload)
+
+ def test_new_identity_requires_zero_count_and_old_identity_rejects_new_key(self):
+ with tempfile.TemporaryDirectory() as temporary:
+ raw, store, _ = build_delivery(xml_document(reservation(1)), Path(temporary))
+ envelope = DeliveryEnvelope.from_dict(json.loads(raw))
+ payload = json.loads(store.objects[envelope.artifacts["structured_result_json"].object_key])
+ self.assertEqual(payload["outcome_counts"]["excluded_cancelled"], 0)
+ _validate_structured_payload_v4(payload, envelope)
+ missing = copy.deepcopy(payload)
+ del missing["outcome_counts"]["excluded_cancelled"]
+ with self.assertRaisesRegex(IngestionError, "outcome counts contract"):
+ _validate_structured_payload_v4(missing, envelope)
+ old_envelope = replace(envelope, processor_version="4.2.0")
+ old_payload = copy.deepcopy(payload)
+ old_payload["processor_version"] = "4.2.0"
+ with self.assertRaisesRegex(IngestionError, "outcome counts contract"):
+ _validate_structured_payload_v4(old_payload, old_envelope)
+ del old_payload["outcome_counts"]["excluded_cancelled"]
+ _validate_structured_payload_v4(old_payload, old_envelope)
+
+ def test_old_identity_cannot_disguise_cancelled_record_or_rule_hash(self):
+ with tempfile.TemporaryDirectory() as temporary:
+ raw, store, envelope, payload = self.generated(Path(temporary))
+ old = copy.deepcopy(payload)
+ old["processor_version"] = "4.2.0"
+ 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"))
+ bad_hash = json.loads(raw)
+ bad_hash["rule_set_sha256"] = "0" * 64
+ with self.assertRaises(IngestionError) as caught:
+ DeliveryValidator(store, policy()).validate(json.dumps(bad_hash).encode())
+ self.assertEqual(caught.exception.code, "PROCESSOR_NOT_ALLOWED")
+
+ 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))
+ counts["excluded_cancelled"] = 6
+ self.assertEqual(_outcome_counts({"outcome_counts": counts}), (2, 1, 3, 4, 5, 6))
+
+ def test_cancelled_count_balances_pending_price_review(self):
+ with tempfile.TemporaryDirectory() as temporary:
+ raw, store, _ = build_delivery(cancelled_xml(rate_amount="1800"), Path(temporary))
+ verified = DeliveryValidator(store, policy()).validate(raw)
+ self.assertEqual(verified.envelope.status, "review_required")
+ self.assertEqual(verified.structured_payload["outcome_counts"]["excluded_cancelled"], 1)
+ self.assertEqual(verified.structured_payload["review_required_rows"], 1)
+
+
+@unittest.skipUnless(os.environ.get("ARR_TEST_LOCAL_POSTGRES") == "1", "owned PostgreSQL opt-in required")
+class CancelledFinanceMigrationTests(unittest.TestCase):
+ @classmethod
+ def setUpClass(cls):
+ cls.database = TemporaryPostgres().__enter__()
+ cls.addClassCleanup(cls.database.__exit__, None, None, None)
+
+ def test_upgrade_preserves_history_permissions_guards_and_count_constraints(self):
+ import psycopg
+ self.database.reset_database(schema_version=19)
+ repository = PostgresIngestionRepository(DatabaseConfig("owned-fixture"), connect=self.database.connect)
+ with self.assertRaises(IngestionError) as missing:
+ repository.assert_cancelled_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]
+ run = connection.execute("""INSERT INTO ingestion.processing_runs
+ (run_key,pipeline_type,source_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-success','opera_daily',%s,'accepted','direct_mcp','2026-07-27',
+ '3.0.0',%s,'3.0',%s,now(),now()) RETURNING id""",
+ (artifact, "b" * 64, "c" * 64)).fetchone()[0]
+ version = connection.execute("""INSERT INTO finance.daily_versions
+ (business_date,version_no,processing_run_id,source_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,validated_at,result_delivery_mode)
+ VALUES ('2026-07-27',1,%s,%s,'validated','3.0.0',%s,'3.0',%s,
+ 1,0,1,0,0,0,now(),'direct_mcp') RETURNING id""",
+ (run, artifact, "b" * 64, "c" * 64)).fetchone()[0]
+ before = connection.execute("SELECT to_jsonb(v) FROM finance.daily_versions v WHERE id=%s", (version,)).fetchone()[0]
+ guards = connection.execute("SELECT tgname,pg_get_triggerdef(oid) FROM pg_trigger WHERE NOT tgisinternal ORDER BY tgname").fetchall()
+ connection.execute("CREATE ROLE arr_cancelled_fixture")
+ connection.execute("GRANT USAGE ON SCHEMA finance TO arr_cancelled_fixture")
+ connection.execute("GRANT SELECT, INSERT ON finance.daily_versions TO arr_cancelled_fixture")
+ connection.execute((PROJECT / "database/020_daily_cancelled_exclusion.sql").read_text(), prepare=False)
+ repository.assert_cancelled_scope_schema()
+ after = connection.execute("SELECT to_jsonb(v) FROM finance.daily_versions v WHERE id=%s", (version,)).fetchone()[0]
+ self.assertEqual(after.pop("excluded_cancelled_rows"), 0)
+ self.assertEqual(after, 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_cancelled_fixture','finance.daily_versions','excluded_cancelled_rows','INSERT'),has_table_privilege('arr_cancelled_fixture','finance.daily_versions','UPDATE')").fetchone(), (True, False))
+ with self.assertRaises(psycopg.errors.CheckViolation):
+ connection.execute("UPDATE finance.daily_versions SET excluded_cancelled_rows=1 WHERE id=%s", (version,))
+ connection.execute("UPDATE finance.daily_versions SET excluded_rate_code_rows=0, excluded_cancelled_rows=1 WHERE id=%s", (version,))
+ with self.assertRaisesRegex(psycopg.errors.RaiseException, "cancellation-exclusion facts exist"):
+ connection.execute((PROJECT / "database/020_daily_cancelled_exclusion.down.sql").read_text(), prepare=False)
+ connection.execute("ROLLBACK")
+ connection.execute("SET ROLE arr_cancelled_fixture")
+ with self.assertRaises(psycopg.errors.InsufficientPrivilege):
+ connection.execute("UPDATE finance.daily_versions SET excluded_cancelled_rows=0 WHERE id=%s", (version,))
+ connection.execute("RESET ROLE")
+
+ def test_empty_database_allows_guarded_rollback_and_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/020_daily_cancelled_exclusion.down.sql").read_text(), prepare=False)
+ with self.assertRaises(IngestionError) as missing:
+ repository.assert_data_source_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()
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/test_arr_cancelled_review.py b/tests/test_arr_cancelled_review.py
new file mode 100644
index 0000000..827e940
--- /dev/null
+++ b/tests/test_arr_cancelled_review.py
@@ -0,0 +1,119 @@
+"""Cancellation eligibility before field review; synthetic, offline sources only."""
+import copy
+import hashlib
+import json
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from arr_processing.policy import load_processor_policy
+from arr_web.arr_data_review import DataFieldReviews
+from arr_web.contracts import PortalError
+from integrations.ohip.collect_arr_source import CollectionError
+from tests.test_arr_data_review import source_document, raw, gap, CONTEXT, DAY, REQUEST, ACTOR, SOURCE_MANIFEST
+
+
+class CancelledReviewTests(unittest.TestCase):
+ def setUp(self):
+ temporary = tempfile.TemporaryDirectory(prefix="arr-cancelled-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 evidence(self, statuses=("Cancelled", "InHouse")):
+ return {"version": "arr-reservation-status-evidence/v1", "policy_id": "arr-exclude-cancelled/v1",
+ "original_sha256": hashlib.sha256(raw(self.document)).hexdigest(),
+ "source_manifest_sha256": SOURCE_MANIFEST, "hotel_id": CONTEXT["hotel_id"], "report_date": DAY,
+ "records": [{"source_sequence": row["source_sequence"], "reservation_id": row["reservation_id"],
+ "reservation_status": status, "sources": [{"file": "synthetic.response.bin", "sha256": "f" * 64}]}
+ for row, status in zip(self.document["records"], statuses)]}
+
+ def apply(self, revision, evidence=None):
+ with patch("integrations.ohip.reservation_status.saved_status_evidence", return_value=evidence or self.evidence()):
+ return self.service.apply_saved_status_evidence(REQUEST, Path("unused-synthetic-capture"),
+ expected_revision=revision)
+
+ def test_cancelled_fields_and_rate_are_not_manual_requirements_but_other_rows_still_are(self):
+ self.document["records"][0]["reservation_status"] = "Cancelled"
+ 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_cancelled_count"], 1)
+ self.assertFalse(review["can_finalize"])
+
+ def test_saved_statuses_remove_room_task_preserve_original_and_freeze_audited_source(self):
+ gap(self.document, "DISP_ROOM_NO")
+ before = self.prepare()
+ original = self.service.original(REQUEST)
+ review = self.apply(before["revision"])
+ self.assertEqual(review["pending_count"], 0)
+ self.assertEqual(review["items"], [])
+ self.assertTrue(review["can_finalize"])
+ self.assertEqual(self.service.original(REQUEST), original)
+ self.assertIsNone(self.service.payload(REQUEST))
+ self.assertEqual(self.prepare(), review, "a saved-status overlay must not bypass explicit continuation")
+ self.assertEqual(self.apply(before["revision"]), review, "exact retry is idempotent")
+ final = self.service.finalize(REQUEST, review["revision"], ACTOR)
+ restored = DataFieldReviews(root=self.root, policy=self.policy, context=CONTEXT)
+ payload, binding = restored.payload(REQUEST)
+ derived = json.loads(payload)
+ self.assertEqual(derived["records"][0]["fields"], self.document["records"][0]["fields"])
+ self.assertEqual(derived["records"][0]["reservation_status"], "Cancelled")
+ audit = derived["manual_data_review"]["manifest"]
+ self.assertEqual(audit["changes"], [])
+ self.assertEqual(audit["reservation_status_evidence"]["original_sha256"], hashlib.sha256(original[0]).hexdigest())
+ self.assertEqual([event["action"] for event in audit["events"]], ["reservation_status_evidence", "finalize"])
+ self.assertEqual(restored.get(REQUEST), final)
+ self.assertEqual(restored.original(REQUEST), original)
+ self.assertEqual(len(binding), 64)
+ with self.assertRaises(PortalError):
+ self.apply(final["revision"])
+
+ def test_prior_staff_decision_on_cancelled_row_stays_audited_but_cannot_be_edited(self):
+ gap(self.document, "BLOCK_CODE")
+ before = self.prepare()
+ saved = self.service.update(REQUEST, "1:BLOCK_CODE", before["revision"], "VERIFIED", ACTOR)
+ review = self.apply(saved["revision"])
+ self.assertEqual(review["items"], [])
+ with self.assertRaises(PortalError):
+ 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(derived["manual_data_review"]["manifest"]["changes"][0]["value"], "VERIFIED")
+ self.assertEqual(derived["records"][0]["fields"]["BLOCK_CODE"]["value"], "VERIFIED")
+
+ def test_overlay_rejects_wrong_order_context_and_revision_without_publishing(self):
+ gap(self.document, "DISP_ROOM_NO")
+ review = self.prepare()
+ with self.assertRaises(PortalError):
+ self.apply(review["revision"] + 1)
+ for alter in (lambda e: e["records"].reverse(),
+ lambda e: e.update(original_sha256="0" * 64),
+ lambda e: e["records"].pop(),
+ lambda e: e["records"][0].update(reservation_status="")):
+ evidence = self.evidence()
+ alter(evidence)
+ with self.assertRaises(CollectionError):
+ self.apply(review["revision"], evidence)
+ self.assertEqual(self.service.get(REQUEST), review)
+
+ def test_status_evidence_tamper_is_rejected(self):
+ gap(self.document, "DISP_ROOM_NO")
+ review = self.prepare()
+ self.apply(review["revision"])
+ evidence_file = next((self.root / REQUEST).glob("reservation-status-*.json"))
+ evidence_file.write_bytes(evidence_file.read_bytes() + b" ")
+ with self.assertRaises(CollectionError):
+ self.service.get(REQUEST)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/test_arr_cancelled_scope.py b/tests/test_arr_cancelled_scope.py
new file mode 100644
index 0000000..e2e9cc3
--- /dev/null
+++ b/tests/test_arr_cancelled_scope.py
@@ -0,0 +1,213 @@
+"""Offline cancellation scope checks for both daily report input paths."""
+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, xml_document
+from tests.test_arr_data_review import source_document
+
+
+def xml_row(sequence, statuses=(), *, blank_room=False, blank_rate=False, invalid_fields=False):
+ node = ET.fromstring(reservation(sequence))
+ for tag, value in statuses:
+ ET.SubElement(node, tag).text = value
+ if blank_room:
+ node.find("DISP_ROOM_NO").text = ""
+ if blank_rate:
+ node.find("RATE_CODE").text = ""
+ if invalid_fields:
+ for tag, value in (("COMPANY_NAME", ""), ("FULL_NAME", ""), ("ADULTS", "-1"),
+ ("NO_OF_ROOMS", "0"), ("TRUNC_END", "2026-07-26")):
+ node.find(tag).text = value
+ return ET.tostring(node, encoding="unicode")
+
+
+class CancelledScopeTests(unittest.TestCase):
+ def setUp(self):
+ temporary = tempfile.TemporaryDirectory(prefix="arr-cancelled-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 read_data(self, document):
+ path = self.root / "source.json"
+ path.write_text(json.dumps(document))
+ return core.read_data_source(path)
+
+ def run_data(self, document, output_name="output"):
+ path = self.root / "source.json"
+ path.write_text(json.dumps(document))
+ output = self.root / output_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()):
+ result = core.process(args)
+ self.assertEqual(path.read_bytes(), before)
+ return result, json.loads((output / "structured-result.json").read_text())
+
+ def test_xml_cancellation_precedes_rate_room_other_validation_and_keeps_lineage(self):
+ all_rows, eligible, rate_removed, duplicates, errors = self.classify_xml(
+ xml_row(1, (("RESV_STATUS", " CxL "),), blank_room=True, blank_rate=True, invalid_fields=True),
+ reservation(2))
+ self.assertEqual(errors, [])
+ self.assertEqual((rate_removed, duplicates), (0, 0))
+ self.assertEqual([row["_SOURCE_INDEX"] for row in eligible], [2])
+ self.assertEqual(all_rows[0]["_OUTCOME"], "excluded_cancelled")
+ self.assertEqual(all_rows[0]["_DECISION_CODES"], ["RESERVATION_CANCELLED_EXCLUDED"])
+ self.assertEqual(all_rows[0]["CONFIRMATION_NO"], "SYNTHETIC-CONF-1")
+ self.assertEqual(all_rows[0]["_SOURCE_LOCATION"], "reservation[1]")
+
+ def test_cancelled_with_a_room_never_occupies_duplicate_key(self):
+ rows = [ET.fromstring(reservation(index, room="SAME-ROOM")) for index in (1, 2, 3)]
+ ET.SubElement(rows[0], "SHORT_RESV_STATUS").text = "CANCELLED"
+ all_rows, eligible, _, duplicates, errors = self.classify_xml(
+ *(ET.tostring(node, encoding="unicode") for node in rows))
+ self.assertEqual(errors, [])
+ self.assertEqual([row["_SOURCE_INDEX"] for row in eligible], [2])
+ self.assertEqual([row["_OUTCOME"] for row in all_rows], ["excluded_cancelled", "pending", "duplicate"])
+ self.assertEqual(all_rows[2]["_DUPLICATE_OF_SOURCE_SEQUENCE"], 2)
+ self.assertEqual(duplicates, 1)
+
+ def test_only_exact_cancellation_values_in_explicit_status_fields_are_excluded(self):
+ for field in ("RESV_STATUS", "RESERVATION_STATUS", "SHORT_RESV_STATUS"):
+ for status in ("CXL", " cancelled ", "CaNcElEd"):
+ with self.subTest(field=field, status=status):
+ rows, _, _, _, errors = self.classify_xml(
+ xml_row(1, ((field, status),), blank_room=True), reservation(2))
+ self.assertEqual(errors, [])
+ self.assertTrue(core.is_cancelled_record(rows[0]))
+ for status in ("CA", "CD", "GC", "TA", "NoShow", "CKOT", "CancelledByGuest", ""):
+ with self.subTest(status=status):
+ rows, _, _, _, errors = self.classify_xml(
+ xml_row(1, (("SHORT_RESV_STATUS", status),), blank_room=True), reservation(2))
+ self.assertFalse(core.is_cancelled_record(rows[0]))
+ self.assertIn("XML_ROOM_MISSING", {error.code for error in errors})
+ for field in ("RESERVATION_TYPE", "RESV_TYPE", "BOOKING_TYPE"):
+ with self.subTest(type_field=field):
+ rows, _, _, _, errors = self.classify_xml(
+ xml_row(1, ((field, "CANCELLED"),), blank_room=True), reservation(2))
+ self.assertFalse(core.is_cancelled_record(rows[0]))
+ self.assertIn("XML_ROOM_MISSING", {error.code for error in errors})
+
+ def test_conflicting_xml_status_is_retained_and_cancel_aliases_agree(self):
+ for fields in ((("RESV_STATUS", "CXL"), ("SHORT_RESV_STATUS", "CKIN")),
+ (("RESERVATION_STATUS", "CANCELLED"), ("SHORT_RESV_STATUS", "CA")),
+ (("RESV_STATUS", "CXL"), ("RESV_STATUS", "RESERVED"))):
+ with self.subTest(fields=fields):
+ rows, _, _, _, errors = self.classify_xml(xml_row(1, fields, blank_room=True), reservation(2))
+ self.assertTrue(rows[0]["_RESERVATION_STATUS_CONFLICT"])
+ self.assertFalse(core.is_cancelled_record(rows[0]))
+ self.assertIn("XML_ROOM_MISSING", {error.code for error in errors})
+ rows, _, _, _, errors = self.classify_xml(xml_row(1,
+ (("RESV_STATUS", "CXL"), ("SHORT_RESV_STATUS", "CANCELED")), blank_room=True), reservation(2))
+ self.assertEqual(errors, [])
+ self.assertFalse(rows[0]["_RESERVATION_STATUS_CONFLICT"])
+ self.assertTrue(core.is_cancelled_record(rows[0]))
+
+ def test_data_cancellation_is_shared_and_unknown_or_absent_status_keeps_validation(self):
+ for status in ("Cancelled", " CANCELED ", "cxl"):
+ with self.subTest(status=status):
+ document = source_document(2)
+ document["records"][0]["reservation_status"] = status
+ for field in document["records"][0]["fields"]:
+ document["records"][0]["fields"][field] = {"state": "missing", "value": None}
+ business_date, rows = self.read_data(document)
+ all_rows, _, rate_removed, _, errors = core.classify_input_records(rows, business_date)
+ self.assertEqual(errors, [])
+ self.assertEqual(rate_removed, 0)
+ self.assertEqual(all_rows[0]["_OUTCOME"], "excluded_cancelled")
+ for status in (None, "NoShow", "InHouse", "CA"):
+ with self.subTest(status=status):
+ document = source_document(2)
+ if status is not None:
+ document["records"][0]["reservation_status"] = status
+ document["records"][0]["fields"]["DISP_ROOM_NO"] = {"state": "missing", "value": None}
+ business_date, rows = self.read_data(document)
+ _, _, _, _, errors = core.classify_input_records(rows, business_date)
+ self.assertIn("DATA_ROOM_MISSING", {error.code for error in errors})
+
+ def test_data_status_rejects_invalid_types_and_empty_strings(self):
+ for value in (None, True, 1, [], {}, "", " "):
+ with self.subTest(value=value):
+ document = source_document()
+ document["records"][0]["reservation_status"] = value
+ with self.assertRaises(core.ProcessingFailure) as raised:
+ self.read_data(document)
+ self.assertEqual(raised.exception.errors[0].code, "INPUT_DATA_INVALID")
+
+ def test_xml_success_and_independent_validation_audit_cancellation_separately(self):
+ result, source, _, _, payload = run_processor(xml_document(
+ xml_row(1, (("SHORT_RESV_STATUS", "CXL"),), blank_room=True, blank_rate=True),
+ reservation(2)), self.root)
+ self.assertEqual(result, 0, payload["errors"])
+ 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([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")
+ self.assertEqual(payload["artifacts"]["source_xml"]["sha256"], core.sha256_file(source))
+ for field, value in (("decision_codes", ["RATE_CODE_NOT_WHITELISTED"]),
+ ("real_price", 900), ("channel_key", "QBD")):
+ with self.subTest(tampered=field):
+ changed = copy.deepcopy(payload)
+ changed["records"][0][field] = value
+ with self.assertRaises(core.ProcessingFailure):
+ core.validate_structured_completeness(changed)
+ for name in ("structured-result.schema.json", "data-structured-result.schema.json"):
+ schema = json.loads((core.SKILL_ROOT / "references" / name).read_text())
+ self.assertIn("excluded_cancelled", schema["properties"]["outcome_counts"]["required"])
+ self.assertIn("excluded_cancelled", schema["$defs"]["record"]["properties"]["outcome"]["enum"])
+
+ def test_data_success_review_and_failure_keep_cancelled_rows_outside_business_rules(self):
+ baseline = source_document(2)
+ baseline["records"][0]["reservation_status"] = "Cancelled"
+ for field in baseline["records"][0]["fields"]:
+ baseline["records"][0]["fields"][field] = {"state": "missing", "value": None}
+ result, payload = self.run_data(baseline)
+ self.assertEqual(result, 0, payload["errors"])
+ self.assertEqual(payload["outcome_counts"]["excluded_cancelled"], 1)
+ self.assertEqual(payload["removed_by_rate_code"], 0)
+ self.assertEqual(payload["output_rows"], 1)
+ review = copy.deepcopy(baseline)
+ review["records"][1]["fields"]["EFFECTIVE_RATE_AMOUNT"]["value"] = "8765"
+ result, payload = self.run_data(review, "review")
+ self.assertEqual(result, 0, payload["errors"])
+ self.assertEqual(payload["status"], "review_required", payload["errors"])
+ self.assertEqual(payload["outcome_counts"]["excluded_cancelled"], 1)
+ self.assertEqual(payload["outcome_counts"]["price_unmatched"], 1)
+ failed = copy.deepcopy(baseline)
+ failed["records"][1]["fields"]["DISP_ROOM_NO"] = {"state": "missing", "value": None}
+ result, payload = self.run_data(failed, "failure")
+ self.assertNotEqual(result, 0)
+ self.assertEqual(payload["status"], "failed")
+ self.assertEqual(payload["outcome_counts"]["excluded_cancelled"], 1)
+ self.assertEqual(payload["outcome_counts"]["validation_failed"], 1)
+
+ def test_legacy_v3_replay_retains_original_scope_and_original_outcome_contract(self):
+ rows = (xml_row(1, (("SHORT_RESV_STATUS", "CXL"),)), reservation(2))
+ result, _, _, _, payload = run_processor(xml_document(*rows), self.root, legacy_v3_output=True)
+ self.assertEqual(result, 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"])
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/test_arr_direct_data.py b/tests/test_arr_direct_data.py
index f5814d2..97df005 100644
--- a/tests/test_arr_direct_data.py
+++ b/tests/test_arr_direct_data.py
@@ -291,6 +291,33 @@ class DirectDataPostgresTests(unittest.TestCase):
self.assertEqual(self.monthly_worker().process_next().status, "published")
self.assertEqual(self.coordinator.get_price_review(outcome.job_id, 50, 0)["case_status"], "completed")
+ def test_cancelled_direct_source_keeps_audit_but_no_room_review_or_monthly_row(self):
+ row = self.transport.rows[0]
+ row["reservationStatus"] = "Cancelled"
+ row["roomStay"].pop("roomId", None)
+ row["roomStay"].pop("currentRoomInfo", None)
+ for rate in row["roomStay"]["roomRates"]:
+ rate.pop("roomId", None)
+ self.transport.calendar_rooms = []
+ self.assertEqual(self.execute().status, "succeeded")
+ self.assertEqual(self.sql("SELECT result_schema_version,source_rows,retained_rows,excluded_cancelled_rows FROM finance.daily_versions"),
+ [("5.0", 6, 5, 1)])
+ self.assertEqual(self.sql("SELECT outcome FROM finance.daily_records WHERE source_sequence=1"), [("excluded_cancelled",)])
+ 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(c.data_type == "f" for sheet in workbook for row in sheet for c 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_commit_ack_loss_retry_has_one_result_and_no_new_fetch(self):
state = {"armed": True}
self.repository._connect = lambda dsn: pg_helpers.LoseCommitConnection(self.database.connect(dsn), state)
diff --git a/tests/test_arr_download_postgres_integration.py b/tests/test_arr_download_postgres_integration.py
index e3017f4..cdb4fe6 100644
--- a/tests/test_arr_download_postgres_integration.py
+++ b/tests/test_arr_download_postgres_integration.py
@@ -193,6 +193,41 @@ class PostgresDownloadIntegrationTests(unittest.TestCase):
self.assertEqual(json.loads(response.body)["data"]["job_id"], task["job_id"])
self.assertEqual((self.reader_calls, self.adapter.calls, self.validator.calls, self.processor.calls), (1, 1, 1, 1))
+ def source_with_cancelled_first(self, **kwargs):
+ cancelled = FixtureExecutor.xml(room="SYNTHETIC-CANCELLED").replace(
+ b"SYNTHETIC-CANCELLED", b""
+ ).replace(b"", b"CANCELLED")
+ start = cancelled.index(b"")
+ end = cancelled.index(b"") + len(b"")
+ return FixtureExecutor.xml(**kwargs).replace(
+ b"", b"" + cancelled[start:end]
+ )
+
+ def test_cancelled_without_room_is_audited_but_daily_monthly_use_retained_only(self):
+ self.adapter.payload = self.source_with_cancelled_first()
+ self.execute()
+ self.assert_one_commit()
+ self.assertEqual(self.sql("SELECT source_rows,retained_rows,excluded_cancelled_rows FROM finance.daily_versions"), [(2, 1, 1)])
+ self.assertEqual(self.sql("SELECT outcome,disp_room_no FROM finance.daily_records ORDER BY source_sequence"),
+ [("excluded_cancelled", ""), ("retained", "SYNTHETIC-ROOM-1")])
+ self.assertEqual(self.count("finance.v_active_daily_facts"), 1)
+ self.assertEqual(self.monthly_worker().process_next().status, "published")
+ self.assertEqual(self.count("reporting.monthly_runs"), 1)
+ self.execute()
+ self.assert_one_commit()
+ self.assertEqual(self.sql("SELECT excluded_cancelled_rows FROM finance.daily_versions"), [(1,)])
+
+ def test_failed_batch_retains_cancelled_audit_count_without_monthly_commit(self):
+ self.adapter.payload = self.source_with_cancelled_first(departure="2026-09-14")
+ self.execute()
+ self.assertEqual(self.sql("SELECT version_status,source_rows,excluded_cancelled_rows,validation_failed_rows FROM finance.daily_versions"),
+ [("rejected", 2, 1, 1)])
+ self.assertEqual(self.sql("SELECT outcome FROM finance.daily_records ORDER BY source_sequence"),
+ [("excluded_cancelled",), ("validation_failed",)])
+ self.assertEqual(self.daily_event_count(), 0)
+ self.assertEqual(self.count("finance.current_daily_versions"), 0)
+ self.assertEqual(self.monthly_worker().process_next().status, "idle")
+
def test_rejected_commit_lost_ack_recovers_same_failure_without_new_capture(self):
self.adapter.payload = FixtureExecutor.xml(rate_code="")
state = {"armed": True}
diff --git a/tests/test_arr_download_runtime.py b/tests/test_arr_download_runtime.py
index 9055405..04c5913 100644
--- a/tests/test_arr_download_runtime.py
+++ b/tests/test_arr_download_runtime.py
@@ -40,6 +40,7 @@ class RuntimeTests(unittest.TestCase):
self.root = Path(temporary.name)
self.store = ManagedObjectStore(FilesystemObjectBackend(self.root / 'objects', create=True))
self.repository = InMemoryIngestionRepository()
+ self.repository.assert_cancelled_scope_schema = Mock()
self.processor = CountingProcessor(self.policy)
self.client = Mock()
with patch.object(processing_runtime, 'compose_object_store', return_value=
@@ -56,6 +57,23 @@ class RuntimeTests(unittest.TestCase):
mapping_validator=FixtureValidator(), reader_factory=self.readers,
page_size=2, max_pages=10, max_records=10)
+ def test_processing_composition_checks_finance_cancellation_schema(self):
+ self.repository.assert_cancelled_scope_schema.assert_called_once_with()
+
+ def test_missing_cancellation_migration_closes_storage_and_disables_composition(self):
+ from arr_ingestion.contracts import IngestionError
+ self.repository.assert_cancelled_scope_schema.side_effect = IngestionError(
+ "DATABASE_MIGRATION_MISSING", "ARR processing requires migration 020")
+ 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 readers(self, archive, hotel):
self.factory_calls += 1
return (source.Reader(archive, hotel, self.transport, sleep=lambda _: None),
diff --git a/tests/test_arr_ingestion_postgres.py b/tests/test_arr_ingestion_postgres.py
index 1eb519f..04aa1af 100644
--- a/tests/test_arr_ingestion_postgres.py
+++ b/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"), 20)
+ self.assertEqual(version_insert.count("%s"), 21)
class Migration008ContractTests(unittest.TestCase):
diff --git a/tests/test_arr_opera_daily_ingest.py b/tests/test_arr_opera_daily_ingest.py
index db224c7..b9fd2cd 100644
--- a/tests/test_arr_opera_daily_ingest.py
+++ b/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.2.0")
+ self.assertEqual(core.PROCESSOR_VERSION, "4.3.0")
self.assertEqual(len(core.DAILY_HEADERS), 19)
self.assertEqual(len(core.RATE_WHITELIST), 20)
@@ -302,13 +302,14 @@ class ArrOperaDailyIngestTests(unittest.TestCase):
)
self.assertEqual(payload["result_schema_version"], "4.0")
- self.assertEqual(payload["processor_version"], "4.2.0")
+ self.assertEqual(payload["processor_version"], "4.3.0")
self.assertEqual(payload["source_rows"], 5)
self.assertEqual(
payload["outcome_counts"],
{
"candidate": 0,
"duplicate": 1,
+ "excluded_cancelled": 0,
"excluded_rate_code": 1,
"price_unmatched": 0,
"retained": 3,
@@ -643,9 +644,11 @@ class ArrOperaDailyIngestTests(unittest.TestCase):
"monthly_report",
structured_schema["properties"]["artifacts"]["properties"],
)
- retained = structured_schema["$defs"]["record"]["allOf"][1]["then"][
- "properties"
- ]
+ retained = next(
+ condition["then"]["properties"]
+ for condition in structured_schema["$defs"]["record"]["allOf"]
+ if "retained" in condition["if"]["properties"]["outcome"].get("enum", [])
+ )
self.assertEqual(retained["nights"]["minimum"], 0)
def test_cli_has_no_legacy_monthly_surface(self):
@@ -799,7 +802,7 @@ class ArrOperaDailyIngestTests(unittest.TestCase):
},
}
)
- self.assertEqual(delivery.processor_version, "4.2.0")
+ self.assertEqual(delivery.processor_version, "4.3.0")
self.assertEqual(delivery.result_schema_version, "4.0")
self.assertEqual(delivery.business_date, date(2026, 7, 27))
diff --git a/tests/test_ohip_reservation_status.py b/tests/test_ohip_reservation_status.py
new file mode 100644
index 0000000..1a4d320
--- /dev/null
+++ b/tests/test_ohip_reservation_status.py
@@ -0,0 +1,310 @@
+"""Offline status evidence checks against the real synthetic ARR collector."""
+import copy
+import hashlib
+import json
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from integrations.ohip import arr_data as data
+from integrations.ohip import collect_arr_source as base
+from integrations.ohip import reservation_status as status
+from tests.test_ohip_arr_data import DAY, HOTEL, REQUEST, SimulatedOHIP
+
+
+def digest(raw):
+ return hashlib.sha256(raw).hexdigest()
+
+
+class SavedStatusEvidenceTests(unittest.TestCase):
+ def setUp(self):
+ self.temp = tempfile.TemporaryDirectory()
+ self.addCleanup(self.temp.cleanup)
+ self.root = Path(self.temp.name)
+ self.service = SimulatedOHIP(2)
+ self.capture_number = 0
+ self.collect()
+ self.snapshot = {file.name: file.read_bytes() for file in self.capture.iterdir()}
+
+ def collect(self):
+ self.capture_number += 1
+ source = data.ARRDataSource(self.root / f"source-{self.capture_number}", HOTEL,
+ transport_factory=lambda: self.service, sleep=lambda _: None, page_size=1)
+ summary = source.fetch(DAY, REQUEST)
+ self.assertTrue(summary["collection_complete"], summary)
+ self.capture = Path(summary["data_path"]).parent
+ self.original = (self.capture / "arr-data.json").read_bytes()
+ self.pin = summary["manifest_sha256"]
+
+ def restore(self):
+ for file in self.capture.iterdir():
+ file.unlink()
+ for name, raw in self.snapshot.items():
+ file = self.capture / name
+ file.write_bytes(raw)
+ file.chmod(0o600)
+ self.original = self.snapshot["arr-data.json"]
+ self.pin = digest(self.snapshot["result.json"])
+ self.capture.chmod(0o700)
+
+ def document(self, name):
+ return json.loads((self.capture / name).read_bytes())
+
+ def write(self, name, value):
+ (self.capture / name).write_bytes(base.json_bytes(value))
+
+ def repin(self, *, original_changed=False):
+ manifest = self.document("result.json")
+ for entry in manifest["files"]:
+ raw = (self.capture / entry["name"]).read_bytes()
+ entry.update(bytes=len(raw), sha256=digest(raw))
+ manifest["data_sha256"] = digest((self.capture / "arr-data.json").read_bytes())
+ self.write("result.json", manifest)
+ self.pin = digest((self.capture / "result.json").read_bytes())
+ if original_changed:
+ self.original = (self.capture / "arr-data.json").read_bytes()
+
+ def requests(self, operation):
+ return sorted(file.name for file in self.capture.glob("data-request-*.json")
+ if not file.name.endswith(".meta.json")
+ and self.document(file.name)["operation_id"] == operation)
+
+ def response(self, request):
+ return request.removesuffix(".json") + ".response.bin"
+
+ def evidence(self):
+ return status.saved_status_evidence(self.capture, self.pin, self.original)
+
+ def assert_rejected(self, code=None):
+ with self.assertRaises(base.CollectionError) as caught:
+ self.evidence()
+ if code:
+ self.assertEqual(str(caught.exception), code)
+
+ def test_complete_paginated_capture_is_read_only_and_offline(self):
+ calls = len(self.service.calls)
+ with patch.object(base, "HTTPTransport", side_effect=AssertionError("network forbidden")), \
+ patch.object(base, "load_key", side_effect=AssertionError("credentials forbidden")):
+ receipt = self.evidence()
+ self.assertEqual(receipt["version"], status.VERSION)
+ self.assertEqual(receipt["policy_id"], status.POLICY_ID)
+ self.assertEqual(receipt["original_sha256"], digest(self.original))
+ self.assertEqual(receipt["source_manifest_sha256"], self.pin)
+ self.assertEqual((receipt["report_date"], receipt["hotel_id"]), (DAY, HOTEL))
+ self.assertEqual([(r["source_sequence"], r["reservation_id"], r["reservation_status"])
+ for r in receipt["records"]], [(1, "res0", "InHouse"), (2, "res1", "InHouse")])
+ for row in receipt["records"]:
+ self.assertEqual(len(row["sources"]), 9)
+ self.assertEqual(len({item["file"] for item in row["sources"]}), 9)
+ for item in row["sources"]:
+ self.assertEqual(item["sha256"], digest((self.capture / item["file"]).read_bytes()))
+ self.assertEqual(len(self.service.calls), calls)
+ self.assertEqual({file.name: file.read_bytes() for file in self.capture.iterdir()}, self.snapshot)
+
+ def test_legacy_original_without_status_is_supported(self):
+ payload = self.document("arr-data.json")
+ for row in payload["records"]:
+ row.pop("reservation_status", None)
+ self.write("arr-data.json", payload)
+ self.repin(original_changed=True)
+ self.assertEqual([r["reservation_status"] for r in self.evidence()["records"]], ["InHouse", "InHouse"])
+ self.assertNotIn("reservation_status", self.document("arr-data.json")["records"][0])
+
+ def test_unknown_nonempty_status_is_preserved_without_interpretation(self):
+ for row, value in zip(self.service.rows, ("CA", "FutureOracleStatus")):
+ row["reservationStatus"] = value
+ self.collect()
+ self.assertEqual([r["reservation_status"] for r in self.evidence()["records"]], ["CA", "FutureOracleStatus"])
+
+ def test_retry_failure_is_never_selected_as_evidence(self):
+ self.service.calls.clear()
+ self.service.status = lambda operation, count: 503 if operation == base.SEARCH and count == 1 else 200
+ self.collect()
+ failed = self.response(self.requests(base.SEARCH)[0])
+ receipt = self.evidence()
+ self.assertTrue(all(failed not in {item["file"] for item in row["sources"]} for row in receipt["records"]))
+ # A source row cannot claim that failed response as its status evidence.
+ payload = self.document("arr-data.json")
+ payload["records"][0]["sources"].append(failed)
+ self.write("arr-data.json", payload)
+ self.repin(original_changed=True)
+ self.assert_rejected("status_evidence_source_binding_failed")
+
+ def test_pins_detect_manifest_raw_data_and_source_format_changes(self):
+ for mutation, code in (
+ (lambda: setattr(self, "pin", "0" * 64), "status_evidence_manifest_changed"),
+ (lambda: setattr(self, "original", self.original + b"\n"), "status_evidence_original_changed"),
+ (lambda: (self.capture / self.response(self.requests(base.DETAIL)[0])).write_bytes(b"{}"),
+ "status_evidence_archive_changed"),
+ ):
+ with self.subTest(code=code):
+ self.restore()
+ mutation()
+ self.assert_rejected(code)
+
+ def test_inventory_rejects_unsafe_duplicate_unlisted_and_orphan_files(self):
+ for name in ("../escape", "/tmp/escape", "data-request-000001.json/child"):
+ with self.subTest(name=name):
+ self.restore()
+ manifest = self.document("result.json")
+ manifest["files"][0]["name"] = name
+ self.write("result.json", manifest)
+ self.pin = digest((self.capture / "result.json").read_bytes())
+ self.assert_rejected("status_evidence_inventory_name_invalid")
+ self.restore()
+ manifest = self.document("result.json")
+ manifest["files"].append(copy.deepcopy(manifest["files"][0]))
+ self.write("result.json", manifest)
+ self.pin = digest((self.capture / "result.json").read_bytes())
+ self.assert_rejected("status_evidence_inventory_name_invalid")
+ self.restore()
+ extra = self.capture / "data-request-999999.response.bin"
+ extra.write_bytes(b"{}")
+ extra.chmod(0o600)
+ self.assert_rejected("status_evidence_inventory_changed")
+ manifest = self.document("result.json")
+ manifest["files"].append({"name": extra.name, "bytes": 2, "sha256": digest(b"{}")})
+ self.write("result.json", manifest)
+ self.pin = digest((self.capture / "result.json").read_bytes())
+ self.assert_rejected("status_evidence_request_inventory_invalid")
+
+ def test_incomplete_manifest_or_invalid_options_rejected(self):
+ for target, key, value in (("result.json", "collection_complete", False),
+ ("result.json", "error", "http_failure"),
+ ("result.json", "status", "failed"),
+ ("result.json", "finance_ready", True),
+ ("capture.json", "options", {"unexpected": 1})):
+ with self.subTest(target=target, key=key):
+ self.restore()
+ document = self.document(target)
+ document[key] = value
+ self.write(target, document)
+ self.repin()
+ self.assert_rejected()
+
+ def test_request_metadata_and_envelope_fail_closed(self):
+ request_name = self.requests(base.DETAIL)[0]
+ response_name = self.response(request_name)
+ meta_name = request_name.removesuffix(".json") + ".meta.json"
+ for target, key, value in ((meta_name, "http_status", 500), (meta_name, "error", "transport_failure"),
+ (meta_name, "oversized", True), (response_name, "operation_id", base.SEARCH),
+ (response_name, "hotel_id", "OTHER_HOTEL"),
+ (response_name, "oracle_request_id", " "),
+ (response_name, "upstream_status", 500),
+ (response_name, "warnings", [{"message": "partial result"}]),
+ (request_name, "method", "POST"),
+ (request_name, "path", "/api/v1/reservations/res0"),
+ (request_name, "path", None)):
+ with self.subTest(target=target, key=key, value=value):
+ self.restore()
+ document = self.document(target)
+ document[key] = value
+ self.write(target, document)
+ self.repin()
+ self.assert_rejected()
+
+ def test_typed_identity_hotel_date_and_detail_cardinality_must_match(self):
+ response_name = self.response(self.requests(base.DETAIL)[0])
+ for key, value in (("reservationIdList", [{"type": "Reservation", "id": "res99"}]),
+ ("reservationIdList", [{"type": "Confirmation", "id": "res0"}]),
+ ("reservationIdList", [{"type": "Reservation", "id": "res0"}] * 2),
+ ("hotelId", "OTHER_HOTEL"),
+ ("roomStay", {"arrivalDate": "2026-09-16"})):
+ with self.subTest(key=key, value=value):
+ self.restore()
+ envelope = self.document(response_name)
+ envelope["data"]["reservations"]["reservation"][0][key] = value
+ self.write(response_name, envelope)
+ self.repin()
+ self.assert_rejected()
+ for count in (0, 2):
+ with self.subTest(count=count):
+ self.restore()
+ envelope = self.document(response_name)
+ envelope["data"]["reservations"]["reservation"] *= count
+ self.write(response_name, envelope)
+ self.repin()
+ self.assert_rejected("status_evidence_detail_ambiguous")
+
+ def test_conflicting_missing_empty_and_nonstring_status_rejected(self):
+ response_name = self.response(self.requests(base.DETAIL)[0])
+ for value in (None, "", " ", 0, False, {}, [], "Cancelled"):
+ with self.subTest(value=value):
+ self.restore()
+ envelope = self.document(response_name)
+ envelope["data"]["reservations"]["reservation"][0]["reservationStatus"] = value
+ self.write(response_name, envelope)
+ self.repin()
+ self.assert_rejected("status_evidence_status_conflict")
+ self.restore()
+ envelope = self.document(response_name)
+ envelope["data"]["reservations"]["reservation"][0].pop("reservationStatus")
+ self.write(response_name, envelope)
+ self.repin()
+ self.assert_rejected("status_evidence_status_conflict")
+
+ def test_modified_timestamp_and_final_search_drift_rejected(self):
+ detail_name = self.response(self.requests(base.DETAIL)[0])
+ envelope = self.document(detail_name)
+ envelope["data"]["reservations"]["reservation"][0]["lastModifyDateTime"] = "later"
+ self.write(detail_name, envelope)
+ self.repin()
+ self.assert_rejected("status_evidence_state_conflict")
+ self.restore()
+ final_name = self.response(self.requests(base.SEARCH)[-1])
+ envelope = self.document(final_name)
+ envelope["data"]["reservations"]["reservationInfo"][0]["reservationStatus"] = "Cancelled"
+ self.write(final_name, envelope)
+ self.repin()
+ self.assert_rejected("status_evidence_search_changed")
+
+ def test_original_order_existing_status_and_source_refs_are_bound(self):
+ for mutate in (
+ lambda rows: rows.reverse(),
+ lambda rows: rows[0].update(source_sequence=2),
+ lambda rows: rows[0].update(reservation_id="res99"),
+ lambda rows: rows[0].update(reservation_status="Cancelled"),
+ lambda rows: rows[0].update(sources=rows[1]["sources"]),
+ lambda rows: rows[0]["sources"].append(self.response(self.requests(base.DETAIL)[1])),
+ lambda rows: rows[0]["sources"].append(rows[0]["sources"][0]),
+ ):
+ with self.subTest(mutation=mutate):
+ self.restore()
+ payload = self.document("arr-data.json")
+ mutate(payload["records"])
+ self.write("arr-data.json", payload)
+ self.repin(original_changed=True)
+ self.assert_rejected()
+
+ def test_search_body_cannot_change_date_status_scope_or_pagination(self):
+ request_name = self.requests(base.SEARCH)[0]
+ for key, value in (("arrivalStartDate", "2026-09-16"), ("reservationStatus", ["InHouse"]),
+ ("offset", 1), ("limit", 100)):
+ with self.subTest(key=key):
+ self.restore()
+ request = self.document(request_name)
+ request["body"][key] = value
+ self.write(request_name, request)
+ self.repin()
+ self.assert_rejected()
+
+ def test_archive_private_permissions_and_no_symlink_files(self):
+ self.capture.chmod(0o755)
+ self.assert_rejected("status_evidence_directory_unsafe")
+ self.capture.chmod(0o700)
+ target = self.capture / "arr-data.json"
+ target.chmod(0o644)
+ self.assert_rejected("unsafe_archive_file")
+ target.chmod(0o600)
+ outside = self.root / "outside.json"
+ outside.write_bytes(target.read_bytes())
+ outside.chmod(0o600)
+ target.unlink()
+ target.symlink_to(outside)
+ with self.assertRaises(OSError):
+ self.evidence()
+
+
+if __name__ == "__main__":
+ unittest.main()