From a2f341fc1210f7125ca69171d469ca6bec92e91c Mon Sep 17 00:00:00 2001 From: GeneAI Date: Wed, 19 Aug 2026 23:31:25 -0400 Subject: [PATCH] fix: guard get_question_batches and degrade malformed direct-built payloads (confirmation-pass-1) Two needs-a-look rows from the confirmation-pass-1 ledger (2026-08-20), both direct-Python-API reach only: - FormSchema.get_question_batches rejects non-positive batch_size with a named ValueError (before: -1 silently dropped every question, 0 raised a raw range() error). Reject over clamp: the size is code-authored, so a loud error is catchable where a clamp reshapes the bug. - Formatting helpers degrade malformed construct payloads like suggested_pick instead of crashing: confirm_consequences and endorsement_map are new one-place shape guards used by every surface; expansion_items skips non-dict rows. Parse-path rejection unchanged. Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 11 ++ src/attune_forms/markdown_surface.py | 6 +- src/attune_forms/models.py | 63 +++++++++-- src/attune_forms/widget.py | 6 +- tests/test_models_guards.py | 151 +++++++++++++++++++++++++++ 5 files changed, 226 insertions(+), 11 deletions(-) create mode 100644 tests/test_models_guards.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 0320848..1e7ee59 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -70,6 +70,17 @@ follow [SemVer](https://semver.org/). fallback, matching what the code reads ### Fixed +- Unknown DEFINITION keys are now named problems instead of silently + ignored (confirmation-pass-1 chair ruling, 2026-08-20): a typo'd + field key (`"maximun": 10`) built a bound-less field that validated + any answer clean. `form_from_dict` rejects every unrecognized + top-level and field-level key (`unknown definition key '...'`, + mirroring #37's answer-side wording; the `label`/`questions` aliases + stay accepted), and the MCP `inputSchema` declares + `additionalProperties: false` on both the form and field objects so + the SDK gate agrees with the parser. The schema is D3-mirrored to + attune-ai — the mirror must pick up `additionalProperties` at the + next release-gated re-sync - Unknown DEFINITION keys are now named problems instead of silently ignored (confirmation-pass-1 chair ruling, 2026-08-20): a typo'd field key (`"maximun": 10`) built a bound-less field that validated diff --git a/src/attune_forms/markdown_surface.py b/src/attune_forms/markdown_surface.py index 749371e..aa7b86e 100644 --- a/src/attune_forms/markdown_surface.py +++ b/src/attune_forms/markdown_surface.py @@ -29,6 +29,8 @@ FormQuestion, FormSchema, QuestionType, + confirm_consequences, + endorsement_map, expansion_items, ranking_slot_count, recommended_first, @@ -105,7 +107,7 @@ def _option_lines(q: FormQuestion, *, badge_for: dict[str, str] | None = None) - def _endorsement_suffix(q: FormQuestion, opt: str) -> str: """ " — endorsed by: a, b" for a deliberation option, or "".""" - names = (q.endorsements or {}).get(opt) + names = endorsement_map(q).get(opt) return f" — endorsed by: {', '.join(names)}" if names else "" @@ -221,7 +223,7 @@ def _control_lines(q: FormQuestion) -> list[str]: return _assumption_lines(q) if q.type == QuestionType.CONFIRM: lines = ["If approved:"] - for item in q.consequences or []: + for item in confirm_consequences(q): tag = f" `{item['severity']}`" if item.get("severity") else "" detail = f" — {item['detail']}" if item.get("detail") else "" lines.append(f"- {item.get('label', '')}{tag}{detail}") diff --git a/src/attune_forms/models.py b/src/attune_forms/models.py index ed43999..915eb64 100644 --- a/src/attune_forms/models.py +++ b/src/attune_forms/models.py @@ -142,18 +142,26 @@ def _consequences_summary(consequences: list[dict[str, str]] | None) -> str | No The flat-surface projection of the preview: label, severity in parentheses, detail after a dash — compact enough for help_text or an elicitation description without dumping a paragraph per item. + + A direct-built question carrying non-dict entries degrades to + skipping them (to ``None`` when nothing renderable remains) instead + of crashing the surface — the same norm as :func:`suggested_pick`; + the parse path (``form_from_dict``) rejects the shape with a named + problem before it ever reaches here. """ if not consequences: return None bits = [] for item in consequences: + if not isinstance(item, dict): + continue bit = item.get("label", "") if item.get("severity"): bit += f" ({item['severity']})" if item.get("detail"): bit += f" — {item['detail']}" bits.append(bit) - return "Will: " + "; ".join(bits) + return "Will: " + "; ".join(bits) if bits else None def triage_item_key(item: dict[str, str]) -> str: @@ -223,6 +231,11 @@ def expansion_items(question: "FormQuestion") -> list[tuple[str, dict[str, str]] validators) iterates through this one function, so the item set and its keys can never differ between surfaces (0.5.0 cleanup batch: triage-expansion unification). + + A direct-built question carrying non-dict rows degrades to skipping + them instead of crashing whichever surface iterates first — the same + norm as :func:`suggested_pick`; the parse path rejects the shape + with a named problem before it ever reaches here. """ if question.type is QuestionType.TRIAGE: items = question.triage_items @@ -230,7 +243,7 @@ def expansion_items(question: "FormQuestion") -> list[tuple[str, dict[str, str]] items = question.assumptions else: items = None - return [(triage_item_key(item), item) for item in items or []] + return [(triage_item_key(item), item) for item in items or [] if isinstance(item, dict)] def suggested_pick(question: "FormQuestion", key: str) -> str | None: @@ -247,6 +260,31 @@ def suggested_pick(question: "FormQuestion", key: str) -> str | None: return None +def endorsement_map(question: "FormQuestion") -> dict[str, list[str]]: + """A DELIBERATION's ``endorsements`` mapping, or ``{}``. + + The shape guard belongs in exactly one place, like + :func:`suggested_pick`: a direct-built question carrying a + list-typed ``endorsements`` degrades to "no endorsements" on every + surface (help-text fold, widget chips, markdown suffix) instead of + crashing whichever one read the mapping first; the parse path + rejects the shape with a named problem. + """ + return question.endorsements if isinstance(question.endorsements, dict) else {} + + +def confirm_consequences(question: "FormQuestion") -> list[dict[str, str]]: + """A CONFIRM's renderable ``consequences`` rows. + + Same one-place shape guard as :func:`endorsement_map`: a direct-built + question carrying non-dict entries degrades to skipping them on every + surface (widget rows, markdown rows, the flat-surface summary) + instead of crashing; the parse path rejects the shape with a named + problem. + """ + return [c for c in question.consequences or [] if isinstance(c, dict)] + + def item_context(question: "FormQuestion", item: dict[str, str]) -> str | None: """The one-line context of an expanded row, or ``None`` when bare. @@ -453,10 +491,9 @@ def to_ask_user_format(self) -> dict[str, Any]: # per-option endorsements fold into help_text — otherwise the # 2-1 split (the construct's whole payload) would be invisible. help_text = self._fallback_help() - if self.type is QuestionType.DELIBERATION and self.endorsements: - who = "; ".join( - f"{opt}: {', '.join(names)}" for opt, names in self.endorsements.items() - ) + endorsements = endorsement_map(self) + if self.type is QuestionType.DELIBERATION and endorsements: + who = "; ".join(f"{opt}: {', '.join(names)}" for opt, names in endorsements.items()) note = f"Endorsements — {who}" help_text = f"{help_text} · {note}" if help_text else note return { @@ -605,12 +642,24 @@ def get_question_batches(self, batch_size: int = 4) -> list[list[FormQuestion]]: """Batch questions for asking (AskUserQuestion supports max 4 at once). Args: - batch_size: Maximum questions per batch (default: 4) + batch_size: Maximum questions per batch (default: 4); must be + at least 1 Returns: List of question batches + Raises: + ValueError: For a non-positive ``batch_size`` — before the + guard, ``-1`` silently returned ``[]`` (every question + dropped) and ``0`` raised a raw ``range()`` error + (confirmation-pass-1 finding, 2026-08-20). Rejecting + beats clamping: the size is code-authored, not user + data, so a loud error at the misuse site is catchable + where a silent clamp-to-1 would just reshape the bug. + """ + if batch_size < 1: + raise ValueError(f"batch_size must be at least 1, got {batch_size}") batches = [] for i in range(0, len(self.questions), batch_size): batches.append(self.questions[i : i + batch_size]) diff --git a/src/attune_forms/widget.py b/src/attune_forms/widget.py index e81095d..ee3d1c1 100644 --- a/src/attune_forms/widget.py +++ b/src/attune_forms/widget.py @@ -34,6 +34,8 @@ FormQuestion, FormSchema, QuestionType, + confirm_consequences, + endorsement_map, expansion_items, ranking_slot_count, recommended_first, @@ -251,7 +253,7 @@ def _control_deliberation_html(q: FormQuestion) -> str: path as DECISION — the user chairs the pick. """ notes = q.option_notes or {} - endorse = q.endorsements or {} + endorse = endorsement_map(q) cards = "" for opt in recommended_first(q): is_rec = opt == q.recommended @@ -320,7 +322,7 @@ def _control_confirm_html(q: FormQuestion) -> str: is ever pre-selected or badged, so approving is an explicit act. """ rows = "" - for item in q.consequences or []: + for item in confirm_consequences(q): tag = ( f'{_esc(item["severity"])}' if item.get("severity") diff --git a/tests/test_models_guards.py b/tests/test_models_guards.py new file mode 100644 index 0000000..7c3e348 --- /dev/null +++ b/tests/test_models_guards.py @@ -0,0 +1,151 @@ +"""Tests for the confirmation-pass-1 models fixes (ledger 2026-08-20). + +Two needs-a-look rows, both direct-Python-API reach only (the parse +path validates these shapes before they reach the helpers): + +1. ``get_question_batches`` rejects a non-positive ``batch_size`` with a + named ``ValueError`` — before, ``-1`` silently returned ``[]`` (every + question dropped) and ``0`` raised a raw ``range()`` error. +2. The formatting helpers now degrade direct-built malformed shapes the + way ``suggested_pick`` always did, instead of crashing: non-dict + ``consequences`` entries are skipped, a list-typed ``endorsements`` + reads as no endorsements (``endorsement_map``), and non-dict + ``triage_items`` / ``assumptions`` rows are skipped by + ``expansion_items`` on every surface. +""" + +from __future__ import annotations + +import pytest + +from attune_forms import ( + form_to_askuserquestion, + form_to_markdown, + form_to_widget_html, +) +from attune_forms.elicitation_schema import form_to_elicitation_schema +from attune_forms.models import ( + FormQuestion, + FormSchema, + QuestionType, + _consequences_summary, + endorsement_map, + expansion_items, +) + + +def _form(q: FormQuestion) -> FormSchema: + return FormSchema(title="T", description="", questions=[q]) + + +def _render_every_surface(q: FormQuestion) -> None: + """Every flat + rich surface renders the question rather than raising.""" + form = _form(q) + form_to_widget_html(form, instance_id="x") + form_to_askuserquestion(form) + form_to_elicitation_schema(form) + form_to_markdown(form) + + +class TestGetQuestionBatchesGuard: + _QS = [FormQuestion(id=f"q{i}", text=f"Q{i}", type=QuestionType.TEXT_INPUT) for i in range(5)] + + def test_negative_batch_size_is_named_not_silent_empty(self) -> None: + """The silent-drop shape itself: -1 used to return [] — five + questions gone with no signal.""" + form = FormSchema(title="T", description="", questions=list(self._QS)) + with pytest.raises(ValueError, match="batch_size must be at least 1, got -1"): + form.get_question_batches(batch_size=-1) + + def test_zero_batch_size_is_named_not_raw_range_error(self) -> None: + form = FormSchema(title="T", description="", questions=list(self._QS)) + with pytest.raises(ValueError, match="batch_size must be at least 1, got 0"): + form.get_question_batches(batch_size=0) + + def test_positive_sizes_still_batch(self) -> None: + form = FormSchema(title="T", description="", questions=list(self._QS)) + assert [len(b) for b in form.get_question_batches()] == [4, 1] + assert [len(b) for b in form.get_question_batches(batch_size=1)] == [1] * 5 + assert [len(b) for b in form.get_question_batches(batch_size=2)] == [2, 2, 1] + + +class TestConsequencesDegrade: + def test_non_dict_entries_are_skipped_not_crashed(self) -> None: + """Before the guard: raw AttributeError from ``item.get`` on the + string entry — inconsistent with suggested_pick's degrade norm.""" + assert ( + _consequences_summary( + [{"label": "Tag pushed", "severity": "irreversible"}, "oops"] # type: ignore[list-item] + ) + == "Will: Tag pushed (irreversible)" + ) + + def test_all_bad_entries_degrade_to_none(self) -> None: + assert _consequences_summary(["oops", 3]) is None # type: ignore[list-item] + + def test_confirm_renders_on_every_surface(self) -> None: + q = FormQuestion( + id="gate", + text="Ship?", + type=QuestionType.CONFIRM, + consequences=[{"label": "Tag pushed"}, "oops"], # type: ignore[list-item] + ) + _render_every_surface(q) + payload = q.to_ask_user_format() + assert payload["help_text"] == "Will: Tag pushed" + + +class TestEndorsementsDegrade: + _BAD = FormQuestion( + id="pick", + text="Which?", + type=QuestionType.DELIBERATION, + options=["A", "B"], + endorsements=["claude", "codex"], # type: ignore[arg-type] — the wrong shape, deliberately + ) + + def test_list_typed_endorsements_read_as_none(self) -> None: + assert endorsement_map(self._BAD) == {} + + def test_mapping_passes_through(self) -> None: + q = FormQuestion( + id="pick", + text="Which?", + type=QuestionType.DELIBERATION, + options=["A", "B"], + endorsements={"A": ["claude"]}, + ) + assert endorsement_map(q) == {"A": ["claude"]} + + def test_every_surface_renders_without_the_fold(self) -> None: + """Before the guard: ``.items()`` (help-text fold), ``.get`` + (widget chips, markdown suffix) each raised AttributeError.""" + _render_every_surface(self._BAD) + payload = self._BAD.to_ask_user_format() + assert payload["help_text"] is None # no endorsement fold, no crash + + +class TestExpansionItemsDegrade: + def test_non_dict_triage_rows_are_skipped(self) -> None: + q = FormQuestion( + id="board", + text="Rule each", + type=QuestionType.TRIAGE, + triage_items=[{"id": "one", "label": "First"}, "oops"], # type: ignore[list-item] + dispositions=["keep", "drop"], + ) + assert [k for k, _ in expansion_items(q)] == ["one"] + _render_every_surface(q) + assert [p["question_id"] for p in q.to_ask_user_formats()] == ["board.one"] + + def test_non_dict_assumption_rows_are_skipped(self) -> None: + q = FormQuestion( + id="assume", + text="Rule these", + type=QuestionType.ASSUMPTION_REVIEW, + assumptions=[{"id": "py", "label": "Py floor"}, 42], # type: ignore[list-item] + ) + assert [k for k, _ in expansion_items(q)] == ["py"] + _render_every_surface(q) + ids = [p["question_id"] for p in q.to_ask_user_formats()] + assert ids == ["assume.py", "assume.py.text"]