From e35a97d5d882027eea6428d13c3b7746a74fd593 Mon Sep 17 00:00:00 2001 From: YZJF <195568136+YZJF@users.noreply.github.com> Date: Fri, 18 Sep 2026 20:04:53 +0800 Subject: [PATCH] feat(quota): give the blocked priority transition a typed owner notice #4543 made the quota projection admit that a higher-priority Todo blocked ahead of an executable fallback should be told to the owner. That admission is an intent, not a delivery: the owner-facing reason was still one generic sentence with no task, cause, recovery condition or next step. This slice ships the notice itself and nothing that has no caller: * build_blocked_transition_notice turns every first material entry into blocked into a typed notice - task, cause, evidence, impact, responsible party, recovery condition and next action - keeping owner_must_know and owner_must_act as separate decisions, plus a blocker_revision digest over the cause, evidence, recovery condition and responsible party so a later ledger can dedup "same blocker, same cause" from "materially changed". * should_run_prepare attaches at most three notices to the projection. * blocked_priority_fallback_owner_reason renders the first notice as the owner-facing sentence and falls back to the previous prose when the payload carries none, so the projection stays a thin caller instead of growing a second copy of the rules. Deliberately out of scope: the emission decision, the persisted delivery and readback state and the reconciliation of resolved and superseded blockers. Those were part of an earlier revision of this branch (661-line module) and were removed after review: a production-reachability grep showed every one of those entry points was exercised only by tests, and AGENTS.md asks that a structure with no real call site wait in docs or todo until a caller exists. They arrive with the successor that binds the ledger to an authorized delivery surface. Until then every notice reports delivery.state == pending: no surface has been authorized, so nothing has reached anybody. Refs #4381. Arrives with #4543. Signed-off-by: YZJF <195568136+YZJF@users.noreply.github.com> --- loopx/canary/module_metric_baseline.json | 2 +- .../quota/blocked_transition_notice.py | 322 ++++++++++++++++++ .../control_plane/quota/should_run_prepare.py | 10 + .../work_items/interaction_contract.py | 8 +- .../test_blocked_transition_notice.py | 245 +++++++++++++ 5 files changed, 584 insertions(+), 3 deletions(-) create mode 100644 loopx/control_plane/quota/blocked_transition_notice.py create mode 100644 tests/control_plane/test_blocked_transition_notice.py diff --git a/loopx/canary/module_metric_baseline.json b/loopx/canary/module_metric_baseline.json index ccbc6ab8af..8bde5c33e2 100644 --- a/loopx/canary/module_metric_baseline.json +++ b/loopx/canary/module_metric_baseline.json @@ -68,7 +68,7 @@ "loopx/control_plane/work_items/interaction_contract.py": { "any_count": 85, "dict_any_count": 0, - "lines": 1544 + "lines": 1548 }, "loopx/extensions/lark/goal_topic_runtime.py": { "any_count": 46, diff --git a/loopx/control_plane/quota/blocked_transition_notice.py b/loopx/control_plane/quota/blocked_transition_notice.py new file mode 100644 index 0000000000..8e8b44c1b5 --- /dev/null +++ b/loopx/control_plane/quota/blocked_transition_notice.py @@ -0,0 +1,322 @@ +"""Typed blocked-transition notice contract (#4381). + +Refs #4381. Arrives with #4543, which made the quota projection admit that a +higher-priority Agent Todo blocked ahead of an executable fallback should be +told to the owner. That admission is an *intent*, not a delivery: it answers +"the owner should know" but says nothing about what the owner is actually told. + +This module closes the first of the gaps #4543 left open: the notice itself. +Every first material entry into ``blocked`` carries the task, the concrete cause +and evidence, the impact, the party that can resolve it, the recovery condition +and the next action. The contract keeps ``owner_must_know`` and ``owner_must_act`` +as separate decisions: an agent-owned blocker states that no owner action is +required, while an owner gate asks one concrete question. + +Each notice also carries a ``blocker_revision`` digest over the cause, evidence, +recovery condition, responsible party and supersession marker. The digest is the +dedup key a later ledger needs — unchanged means "already told", changed means +"tell again" — but this module does not decide emission: that decision, the +persisted delivery/readback state and the reconciliation of resolved and +superseded blockers arrive with the successor that owns a real caller. Until +then every notice reports ``delivery.state == "pending"``: no delivery surface +has been authorized, so nothing has reached anybody, and a NOTIFY intent must +not be reported as a delivery. +""" + +from __future__ import annotations + +import hashlib +import json +from collections.abc import Mapping +from typing import Any + +from ..runtime.public_safety import public_safe_compact_text +from ..todos.contract import ( + TODO_STATUS_BLOCKED, + TODO_TASK_CLASS_USER_ACTION, + TODO_TASK_CLASS_USER_GATE, + normalize_todo_role, + normalize_todo_status, +) +from ..todos.resume_condition import ( + TODO_RESUME_KIND_CAPACITY_AVAILABLE, + TODO_RESUME_KIND_MONITOR_CHANGED, + TODO_RESUME_KIND_PR_MERGED, + TODO_RESUME_KIND_RESUME_AT, + TODO_RESUME_KIND_TODO_DONE, + normalize_todo_resume_when, +) + +BLOCKED_TRANSITION_NOTICE_SCHEMA_VERSION = "blocked_transition_notice_v0" +BLOCKED_TRANSITION_NOTICE_KIND = "blocked_transition_notice" + +# The only delivery state this slice can honestly report: no surface has been +# authorized yet. "delivered" and "readback_verified" arrive with the successor +# that records an actual handover. +NOTICE_DELIVERY_PENDING = "pending" + +RESPONSIBLE_AGENT = "agent" +RESPONSIBLE_OWNER = "owner" +RESPONSIBLE_EXTERNAL = "external_dependency" + + +def _compact(value: Any, limit: int = 180) -> str: + text = public_safe_compact_text(value, limit=limit) + if text is None: + return "" + return str(text).strip() + + +def _digest(payload: Mapping[str, Any]) -> str: + encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + return f"sha256:{hashlib.sha256(encoded.encode('utf-8')).hexdigest()}" + + +def blocked_transition_notice_identity(item: Mapping[str, Any]) -> str | None: + """Return the stable identity of a blocker, or ``None`` when unidentifiable.""" + + todo_id = _compact(item.get("todo_id"), limit=120) + if todo_id: + return f"todo:{todo_id}" + text = _compact(item.get("text")) + if not text: + return None + return f"text:{hashlib.sha256(text.encode('utf-8')).hexdigest()[:16]}" + + +def _resume_kind(resume_when: str | None) -> str | None: + if not resume_when: + return None + return resume_when.split(":", 1)[0] or None + + +def _resume_target(resume_when: str | None) -> str: + if not resume_when: + return "" + return resume_when.split(":", 1)[1] if ":" in resume_when else "" + + +def _recovery_description(kind: str | None, resume_when: str | None, cause: str) -> str: + target = _resume_target(resume_when) + if kind == TODO_RESUME_KIND_TODO_DONE: + return f"todo {target or 'unknown'} must reach done" + if kind == TODO_RESUME_KIND_PR_MERGED: + return f"pull request {target or 'unknown'} must merge" + if kind == TODO_RESUME_KIND_CAPACITY_AVAILABLE: + return "a runtime with the required capability must become available" + if kind == TODO_RESUME_KIND_MONITOR_CHANGED: + return f"monitor {target or 'unknown'} must report a material change" + if kind == TODO_RESUME_KIND_RESUME_AT: + return f"the scheduled resume time {target or 'unknown'} must arrive" + return f"the recorded blocker must clear: {cause}" + + +def _responsible_party( + item: Mapping[str, Any], + *, + resume_kind: str | None, +) -> str: + role = normalize_todo_role(item.get("role")) + task_class = _compact(item.get("task_class"), limit=60).lower() + if role == "user" or task_class in { + TODO_TASK_CLASS_USER_GATE, + TODO_TASK_CLASS_USER_ACTION, + }: + return RESPONSIBLE_OWNER + if resume_kind in {TODO_RESUME_KIND_PR_MERGED, TODO_RESUME_KIND_CAPACITY_AVAILABLE}: + return RESPONSIBLE_EXTERNAL + return RESPONSIBLE_AGENT + + +def _next_action( + *, + owner_must_act: bool, + responsible_party: str, + recovery: str, + fallback_text: str, +) -> str: + if owner_must_act: + question = ( + "Confirm whether to proceed or cancel this blocked step" + if responsible_party == RESPONSIBLE_OWNER + else f"Decide how to resolve this blocker: {recovery}" + ) + return f"Ask the owner one concrete question — {question}." + if fallback_text: + return ( + "No owner action is required: the agent keeps the fallback " + f"'{fallback_text}' moving and reports again if this blocker changes." + ) + return ( + "No owner action is required: the agent keeps independent work moving " + "and reports again if this blocker changes." + ) + + +def build_blocked_transition_notice( + item: Mapping[str, Any], + *, + selected_executable: Mapping[str, Any] | None = None, +) -> dict[str, Any] | None: + """Build the typed first-transition notice for one blocked Todo. + + Returns ``None`` when the item is not in a blocked state: a scheduled + future monitor window is a deferral, and an open item is simply work that + has not started. Only a real blocker earns a notice. + + The returned notice is a plain JSON-serializable mapping. It carries a + ``blocker_revision`` digest so a later ledger can dedup "same blocker, same + cause" from "same blocker, materially changed cause", but it does not itself + decide whether to emit: emission, delivery recording and reconciliation need +an authorized delivery surface and belong to the successor slice. + +The module also owns how a notice is rendered for the owner +(:func:`blocked_priority_fallback_owner_reason`), so the projection that shows +the reason stays a thin caller instead of growing a second copy of the rules. +""" + + if not isinstance(item, Mapping): + return None + identity = blocked_transition_notice_identity(item) + if identity is None: + return None + status = normalize_todo_status(item.get("status")) or "open" + resume_when = normalize_todo_resume_when(item.get("resume_when")) + resume_pending = bool(resume_when and item.get("resume_ready") is False) + if status != TODO_STATUS_BLOCKED and not resume_pending: + return None + + text = _compact(item.get("text")) + reason = _compact(item.get("reason"), limit=240) + resume_kind = _resume_kind(resume_when) + cause = ( + reason + if status == TODO_STATUS_BLOCKED and reason + else ( + f"the todo waits on {resume_when}" + if resume_pending + else "the todo is marked blocked without a recorded reason" + ) + ) + evidence: list[str] = [f"status={status}"] + if reason: + evidence.append(f"reason={reason}") + if resume_when: + evidence.append(f"resume_when={resume_when}") + evidence.append(f"resume_ready={str(item.get('resume_ready')).lower()}") + claimed_by = _compact(item.get("claimed_by"), limit=80) + if claimed_by: + evidence.append(f"claimed_by={claimed_by}") + + responsible_party = _responsible_party(item, resume_kind=resume_kind) + owner_must_act = responsible_party == RESPONSIBLE_OWNER + fallback_text = _compact( + selected_executable.get("text") if isinstance(selected_executable, Mapping) else None + ) + recovery = _recovery_description(resume_kind, resume_when, cause) + impact = ( + f"'{text}' will not advance until {recovery}." + + ( + f" The goal keeps moving on the lower-priority fallback '{fallback_text}'." + if fallback_text + else " No executable fallback is selected, so no agent work advances." + ) + ) + superseded_by = _compact(item.get("superseded_by"), limit=120) + + notice: dict[str, Any] = { + "schema_version": BLOCKED_TRANSITION_NOTICE_SCHEMA_VERSION, + "kind": BLOCKED_TRANSITION_NOTICE_KIND, + "blocker_identity": identity, + "task": { + "todo_id": _compact(item.get("todo_id"), limit=120) or None, + "text": text, + "status": status, + "task_class": _compact(item.get("task_class"), limit=60) or None, + "priority": _compact(item.get("priority"), limit=40) or None, + }, + "cause": cause, + "evidence": evidence, + "impact": impact, + "responsible_party": responsible_party, + "recovery_condition": { + "kind": resume_kind or "status_blocked", + "satisfied": False, + "resume_when": resume_when, + "description": recovery, + }, + "next_action": _next_action( + owner_must_act=owner_must_act, + responsible_party=responsible_party, + recovery=recovery, + fallback_text=fallback_text, + ), + "owner_must_know": True, + "owner_must_act": owner_must_act, + "superseded_by": superseded_by or None, + "delivery": { + "state": NOTICE_DELIVERY_PENDING, + "surface": None, + "delivered_at": None, + "readback_verified_at": None, + }, + } + notice["blocker_revision"] = _digest( + { + "blocker_identity": identity, + "cause": cause, + "evidence": evidence, + "recovery_condition": notice["recovery_condition"], + "responsible_party": responsible_party, + "owner_must_act": owner_must_act, + "superseded_by": notice["superseded_by"], + } + ) + return notice + + +def blocked_transition_notice_owner_reason(notice: Mapping[str, Any]) -> str | None: + """Render one typed notice as the single sentence the owner reads.""" + + if not isinstance(notice, Mapping): + return None + task = notice.get("task") + task_text = str(task.get("text") or "").strip() if isinstance(task, Mapping) else "" + cause = str(notice.get("cause") or "").strip() + impact = str(notice.get("impact") or "").strip() + next_action = str(notice.get("next_action") or "").strip() + parts = [ + part + for part in ( + f"'{task_text}' is blocked: {cause}." if task_text and cause else (cause or None), + impact or None, + next_action or None, + ) + if part + ] + return " ".join(parts) if parts else None + + +def blocked_priority_fallback_owner_reason(fallback: Mapping[str, Any]) -> str | None: + """Render the owner-facing reason for a blocked-priority-fallback payload. + + The typed notice wins over the generic fallback prose so the sentence the + owner reads is the notice that was contracted. The payload keeps every + notice; the reason renders the first one, because one sentence per + heartbeat beats a list the owner has to scan. + """ + + if not isinstance(fallback, Mapping): + return None + notices = fallback.get("blocked_transition_notices") + if isinstance(notices, list): + notice = next( + (item for item in notices if isinstance(item, Mapping)), + None, + ) + if notice is not None: + reason = blocked_transition_notice_owner_reason(notice) + if reason: + return reason + prose = str(fallback.get("reason") or "").strip() + return prose or None diff --git a/loopx/control_plane/quota/should_run_prepare.py b/loopx/control_plane/quota/should_run_prepare.py index 6804c25373..15f679ae62 100644 --- a/loopx/control_plane/quota/should_run_prepare.py +++ b/loopx/control_plane/quota/should_run_prepare.py @@ -30,6 +30,7 @@ from ..goals.goal_frontier import ( build_goal_frontier_projection_context_from_status, ) +from ..quota.blocked_transition_notice import build_blocked_transition_notice from ..quota.error_codes import HeartbeatReceiptIdentityConflictError from ..agents.capability_memory import resolve_agent_capabilities from ..quota.goal_boundary import ( @@ -228,6 +229,7 @@ def _blocked_priority_fallback( return None blocked_items: list[dict[str, Any]] = [] + transition_notices: list[dict[str, Any]] = [] owner_visible_blocker = False for item in first_open: if not isinstance(item, dict): @@ -269,6 +271,13 @@ def _blocked_priority_fallback( status == TODO_STATUS_BLOCKED or resume_condition_pending ): owner_visible_blocker = True + # The owner notice is a typed contract, not only a boolean: it + # carries the cause, evidence, impact, responsible party, recovery + # condition and next action that #4381 asks for, and it keeps + # "must know" separate from "must act". See blocked_transition_notice. + notice = build_blocked_transition_notice(item, selected_executable=selected) + if notice is not None: + transition_notices.append(notice) if not blocked_items: return None @@ -293,6 +302,7 @@ def _blocked_priority_fallback( ) ), "blocked_items": blocked_items[:3], + "blocked_transition_notices": transition_notices[:3], "selected_executable": selected_item, "recommended_action": ( "Keep the blocked core todo visible in status while selecting fallback; " diff --git a/loopx/control_plane/work_items/interaction_contract.py b/loopx/control_plane/work_items/interaction_contract.py index ef6818ecd6..a5ccbde6de 100644 --- a/loopx/control_plane/work_items/interaction_contract.py +++ b/loopx/control_plane/work_items/interaction_contract.py @@ -1,4 +1,5 @@ from __future__ import annotations +from ..quota.blocked_transition_notice import blocked_priority_fallback_owner_reason from ..quota.effective_action import EffectiveAction import shlex import typing @@ -1036,8 +1037,11 @@ def _blocked_priority_fallback_user_reason(payload: dict[str, Any]) -> str | Non and fallback.get("notify_user") is not True ): return None - reason = str(fallback.get("reason") or "").strip() - return reason or None + # The typed notice carries what #4381 asks the owner to be told — the task, + # the concrete cause, the impact, who can resolve it, the recovery + # condition and the next action — and it takes precedence over the generic + # fallback prose. The rendering lives with the notice contract. + return blocked_priority_fallback_owner_reason(fallback) def _interaction_must_attempt( diff --git a/tests/control_plane/test_blocked_transition_notice.py b/tests/control_plane/test_blocked_transition_notice.py new file mode 100644 index 0000000000..88740f4e62 --- /dev/null +++ b/tests/control_plane/test_blocked_transition_notice.py @@ -0,0 +1,245 @@ +"""Regression coverage for the typed blocked-transition notice (#4381). + +#4543 made the quota projection admit that a blocked higher-priority Todo +should be surfaced while a fallback runs. An admission is not a delivery: this +suite pins the contract that turns that intent into one owner-visible notice +with a cause, an impact, a responsible party and a recovery condition, plus the +revision digest a later ledger dedups on. + +Scope note: the emission decision, the persisted delivery/readback state and +the reconciliation of resolved and superseded blockers arrive with the +successor slice that owns a real caller. Only behaviour that is reachable from +production code is pinned here. +""" + +from __future__ import annotations + +from typing import Any + +from loopx.control_plane.quota.blocked_transition_notice import ( + NOTICE_DELIVERY_PENDING, + build_blocked_transition_notice, +) +from loopx.control_plane.quota.should_run_prepare import _blocked_priority_fallback +from loopx.control_plane.work_items.interaction_contract import ( + _blocked_priority_fallback_user_reason, +) + +BLOCKED_TODO_ID = "todo_aaaaaaaaaaaa" +FALLBACK_TODO_ID = "todo_bbbbbbbbbbbb" + + +def _advancement_item( + todo_id: str, + text: str, + *, + status: str, + **extra: Any, +) -> dict[str, Any]: + return { + "todo_id": todo_id, + "text": text, + "status": status, + "task_class": "advancement_task", + **extra, + } + + +def _fallback() -> dict[str, Any]: + return _advancement_item( + FALLBACK_TODO_ID, + "[P1] Prepare independent documentation", + status="open", + ) + + +def _agent_owned_blocker(reason: str = "Required input has not arrived") -> dict[str, Any]: + return _advancement_item( + BLOCKED_TODO_ID, + "[P0] Validate the primary deliverable", + status="blocked", + reason=reason, + ) + + +def _owner_gate_blocker() -> dict[str, Any]: + return { + "todo_id": BLOCKED_TODO_ID, + "text": "[P0] Confirm the production rollout window", + "status": "blocked", + "task_class": "user_gate", + "role": "user", + "reason": "The owner has not confirmed the rollout window", + } + + +def test_agent_owned_blocker_tells_the_owner_without_asking_for_action() -> None: + notice = build_blocked_transition_notice( + _agent_owned_blocker(), selected_executable=_fallback() + ) + + assert notice is not None + assert notice["owner_must_know"] is True + assert notice["owner_must_act"] is False + assert notice["responsible_party"] == "agent" + assert notice["task"]["todo_id"] == BLOCKED_TODO_ID + assert notice["cause"] == "Required input has not arrived" + assert "reason=Required input has not arrived" in notice["evidence"] + assert "status=blocked" in notice["evidence"] + assert "will not advance until" in notice["impact"] + assert notice["recovery_condition"]["satisfied"] is False + assert notice["recovery_condition"]["kind"] == "status_blocked" + assert notice["next_action"].startswith("No owner action is required") + # No surface has been authorized yet, so the notice is pending, not sent. + assert notice["delivery"]["state"] == NOTICE_DELIVERY_PENDING + + +def test_owner_gate_blocker_asks_one_concrete_question() -> None: + notice = build_blocked_transition_notice( + _owner_gate_blocker(), selected_executable=_fallback() + ) + + assert notice is not None + assert notice["owner_must_know"] is True + assert notice["owner_must_act"] is True + assert notice["responsible_party"] == "owner" + assert "one concrete question" in notice["next_action"] + + +def test_resume_condition_names_the_party_that_can_resolve_it() -> None: + waiting = _advancement_item( + BLOCKED_TODO_ID, + "[P0] Validate the primary deliverable", + status="open", + resume_when="pr_merged:acme/loopx#4543", + resume_ready=False, + ) + + notice = build_blocked_transition_notice(waiting, selected_executable=_fallback()) + + assert notice is not None + assert notice["recovery_condition"]["kind"] == "pr_merged" + assert "pull request acme/loopx#4543 must merge" in notice["impact"] + assert notice["responsible_party"] == "external_dependency" + assert notice["owner_must_act"] is False + + +def test_scheduled_future_monitor_earns_no_notice() -> None: + future_monitor = { + "todo_id": BLOCKED_TODO_ID, + "text": "[P1-monitor] Observe the stable public fixture", + "status": "open", + "task_class": "continuous_monitor", + "next_due_at": "2999-01-01T00:00:00Z", + } + + assert build_blocked_transition_notice(future_monitor) is None + + +def test_the_revision_digest_separates_an_unchanged_cause_from_a_new_one() -> None: + notice = build_blocked_transition_notice( + _agent_owned_blocker(), selected_executable=_fallback() + ) + repeat = build_blocked_transition_notice( + _agent_owned_blocker(), selected_executable=_fallback() + ) + changed = build_blocked_transition_notice( + _agent_owned_blocker(reason="Required input has not arrived (vendor retry 2)"), + selected_executable=_fallback(), + ) + superseded = build_blocked_transition_notice( + {**_agent_owned_blocker(), "superseded_by": "todo_cccccccccccc"}, + selected_executable=_fallback(), + ) + assert notice is not None and repeat is not None + assert changed is not None and superseded is not None + + # The digest is the dedup key the successor ledger consumes: same blocker + # and same cause is one notice, a materially changed cause is a new one. + assert notice["blocker_identity"] == repeat["blocker_identity"] + assert notice["blocker_revision"] == repeat["blocker_revision"] + assert changed["blocker_revision"] != notice["blocker_revision"] + assert superseded["blocker_revision"] != notice["blocker_revision"] + + +def test_blocked_primary_with_running_fallback_carries_the_notice() -> None: + fallback = _blocked_priority_fallback( + { + "first_open_items": [_agent_owned_blocker(), _fallback()], + "first_executable_items": [_fallback()], + } + ) + + assert fallback is not None + assert fallback["notify_user"] is True + assert fallback["requires_user_action"] is False + notices = fallback["blocked_transition_notices"] + assert len(notices) == 1 + assert notices[0]["blocker_identity"] == f"todo:{BLOCKED_TODO_ID}" + assert notices[0]["owner_must_act"] is False + assert "Prepare independent documentation" in notices[0]["impact"] + # The fallback keeps running; the notice does not gate delivery. + assert fallback["selected_executable"]["todo_id"] == FALLBACK_TODO_ID + + +def test_owner_facing_reason_carries_the_typed_notice() -> None: + fallback = _blocked_priority_fallback( + { + "first_open_items": [_agent_owned_blocker(), _fallback()], + "first_executable_items": [_fallback()], + } + ) + assert fallback is not None + + reason = _blocked_priority_fallback_user_reason( + {"blocked_priority_fallback": fallback} + ) + + assert reason is not None + assert "Validate the primary deliverable" in reason + assert "Required input has not arrived" in reason + assert "will not advance until" in reason + assert "No owner action is required" in reason + + +def test_owner_facing_reason_falls_back_to_the_generic_prose() -> None: + legacy = { + "kind": "blocked_priority_fallback", + "notify_user": True, + "reason": "a higher-priority agent todo is blocked", + } + + assert ( + _blocked_priority_fallback_user_reason({"blocked_priority_fallback": legacy}) + == "a higher-priority agent todo is blocked" + ) + + +def test_the_owner_reason_renders_one_notice_while_the_payload_keeps_every_one() -> None: + notices = [ + build_blocked_transition_notice( + _agent_owned_blocker(), selected_executable=_fallback() + ), + build_blocked_transition_notice( + _owner_gate_blocker(), selected_executable=_fallback() + ), + ] + assert all(notice is not None for notice in notices) + + payload = { + "kind": "blocked_priority_fallback", + "notify_user": True, + "reason": "a higher-priority agent todo is blocked", + "blocked_transition_notices": notices, + } + + reason = _blocked_priority_fallback_user_reason( + {"blocked_priority_fallback": payload} + ) + + assert reason is not None + # The reason is one sentence for one blocker, not a list to scan. + assert "Validate the primary deliverable" in reason + assert "Confirm the production rollout window" not in reason + # The payload keeps the full set for whoever needs to read it. + assert len(payload["blocked_transition_notices"]) == 2