fix(status-write): validate VFD/RM/PXA transitions in CreateParticipantStatusNode

notes/bt-pitfalls.md

@@ -1172,23 +1172,29 @@ governing spec.
 
 ---
 
-## State-Validation Bypass: `CreateParticipantStatusNode` Does Not Validate Transitions
+## State-Validation at `CreateParticipantStatusNode`
 
-(ISSUE-1825, 2026-07-30; see also `notes/case-state-model.md`)
+(ISSUE-1825 / #1896 concern tracked; fixed in #2081 / PR #2095;
+see also `notes/case-state-model.md`, `notes/status-dimension-objects.md`)
 
-`CreateParticipantStatusNode.update()` constructs `VfdDimension(state=<target>)`
-and persists it directly without calling `VfdDimension.transition()` or
-`is_valid_vfd_transition`. **State validity is enforced entirely by upstream BT
-guard nodes** — nothing at the persistence boundary rejects an invalid jump
-(e.g., `vfd → VFD`).
+`CreateParticipantStatusNode.update()` now validates VFD, RM, and PXA
+transitions **before persisting**. An invalid jump (e.g., `vfd → VFD`,
+`START → ACCEPTED`) returns `Status.FAILURE` with a descriptive
+`feedback_message` and no `ParticipantStatus` record is written.
 
-This is a known fragility (GitHub concern #1896): if a guard node is too weak
-(see "Guard Name Must Match the State-Machine Transition Precondition" above),
-an invalid status snapshot can be persisted silently. The same issue applies
-to RM and PXA dimension writes through this node.
+Validation rules (per `_validate_transitions()`):
 
-**When writing or reviewing guard nodes that precede `CreateParticipantStatusNode`**:
-treat the guard as the *only* line of defence for transition validity and verify
-it against the state-machine transitions defined in `vultron/core/states/`.
+- `target == current` → proceed (status confirmation; valid protocol
+  observation).
+- `is_valid_*_transition(current, target)` is `True` → proceed.
+- Otherwise → `Status.FAILURE` with an informational log line.
+- `None` target → skip validation (caller is preserving current state).
 
-<!-- Source: ISSUE-1825; GitHub concern #1896 -->
+**When writing guard nodes that precede `CreateParticipantStatusNode`**:
+guards remain the *first* line of defence — they prevent the BT from
+reaching an invalid write in the first place. The node-level validation
+is a safety net, not a replacement for correct guards. Verify guard
+preconditions against the state-machine transitions in
+`vultron/core/states/`.
+
+<!-- Source: ISSUE-1825; GitHub concern #1896; fixed PR #2095 -->

plan/history/2608/implementation/ISSUE-2081.md

@@ -0,0 +1,12 @@
+---
+source: ISSUE-2081
+timestamp: '2026-08-07T20:10:05.806120+00:00'
+title: validate VFD/RM/PXA transitions in CreateParticipantStatusNode
+type: implementation
+---
+
+## Issue #2081 — fix(status-write): validate VFD/RM/PXA transitions in CreateParticipantStatusNode
+
+Implements fail-closed transition validation in `CreateParticipantStatusNode.update()`. VFD, RM, and PXA state jumps are now validated before any DataLayer write; illegal transitions return `Status.FAILURE`. Closes #2081 and #1903.
+
+PR: <https://github.com/CERTCC/Vultron/pull/2095>

plan/incoming/learnings/20260807-invalid-rm-pre-states-in-test-fixtures.md

@@ -0,0 +1,26 @@
+---
+title: Test fixtures with invalid RM pre-states pass vacuously after fail-closed validation
+type: learning
+timestamp: 2026-08-07
+source: ISSUE-2081
+signal: concern
+---
+
+Three pre-existing test files (`test_develop_fix_tree.py`, `test_announce_tree.py`,
+`test_close_case_role_semantics.py`) seeded participants at `RM.START` and then
+attempted to write `RM.CLOSED`. These tests had been passing because
+`CreateParticipantStatusNode` was fail-open — it accepted any target state.
+
+After the fail-closed validation landed, those tests failed because `START→CLOSED`
+is not a valid RM transition. The tests were fixed by seeding participants at
+`RM.ACCEPTED` first (a valid pre-state for CLOSED).
+
+The underlying risk: test fixtures can silently produce unrealistic protocol
+scenarios (impossible state sequences) and the test suite won't catch this until
+something validates the transitions. We have no BTTestScenario or fixture-level
+guard that asserts "this participant's RM history is a valid sequence of transitions."
+
+A future concern issue could track: add a fixture-level invariant check that
+validates the RM/VFD/PXA transition sequence of seeded `ParticipantStatus` records,
+or add a warning when `BTTestScenario.seed()` detects an impossible transition in
+the seeded history.

test/architecture/test_vfd_rm_pxa_write_sites.py

@@ -0,0 +1,140 @@
+#!/usr/bin/env python
+
+#  Copyright (c) 2026 Carnegie Mellon University and Contributors.
+#  - see Contributors.md for a full list of Contributors
+#  - see ContributionInstructions.md for information on how you can Contribute to this project
+#  Vultron Multiparty Coordinated Vulnerability Disclosure Protocol Prototype is
+#  licensed under a MIT (SEI)-style license, please see LICENSE.md distributed
+#  with this Software or contact permission@sei.cmu.edu for full terms.
+#  Created, in part, with funding and support from the United States Government
+#  (see Acknowledgments file). This program may include and/or can make use of
+#  certain third party source code, object code, documentation and other files
+#  ("Third Party Software"). See LICENSE.md for more details.
+#  Carnegie Mellon®, CERT® and CERT Coordination Center® are registered in the
+#  U.S. Patent and Trademark Office by Carnegie Mellon University
+"""Architecture ratchet: direct VFD/RM/PXA dimension write sites in behaviors.
+
+AC-7 (issue #2081): audit confirms that all ``VfdDimension(state=…)``,
+``RmDimension(state=…)``, and ``PxaDimension(state=…)`` call sites within
+``vultron/core/behaviors/`` fall into one of the following exempt categories:
+
+- **validated-write** — ``status.py`` delegates to ``CreateParticipantStatusNode``
+  which now validates transitions (the fix in this PR).
+- **bootstrap-seeding** — ``owner.py`` and ``common.py``: deliberate known-state
+  writes for newly-joining participants where no prior state exists.
+- **guard-predicate** — ``develop_fix.py`` / ``deploy_fix.py``: ``VfdDimension``
+  used as a read-only state helper (``is_fix_ready()``, ``is_fix_deployed()``),
+  not as a DataLayer write.
+- **rm-transitions** — ``rm_transitions.py``: tracked separately under the RM
+  state machine; each write is guarded by its own upstream BT condition.
+- **case-proposal** — ``case_proposal_received_tree.py``: bootstrap-seeding on
+  first-time receipt of a case proposal, no prior state available.
+
+This test records the exact set of write sites found during the AC-7 audit.
+New sites not in ``AUDITED_WRITE_SITES`` fail immediately (regression guard).
+Sites in ``AUDITED_WRITE_SITES`` that no longer appear also fail (stale entry).
+
+Spec: SDO-02-004, BTND-10-001 (``specs/behavior-tree-node-design.yaml``).
+Issue: #2081, #1896.
+"""
+
+import ast
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).parents[2]
+_BEHAVIORS_ROOT = REPO_ROOT / "vultron" / "core" / "behaviors"
+
+# Dimension constructors that constitute a state write.
+_WRITE_CONSTRUCTORS: frozenset[str] = frozenset(
+    {"VfdDimension", "RmDimension", "PxaDimension"}
+)
+
+# Each entry is (relative_path_from_behaviors_root, constructor_name).
+AUDITED_WRITE_SITES: frozenset[tuple[str, str]] = frozenset(
+    {
+        # validated-write: inside CreateParticipantStatusNode, after transition check
+        ("case/nodes/participant/status.py", "PxaDimension"),
+        ("case/nodes/participant/status.py", "RmDimension"),
+        ("case/nodes/participant/status.py", "VfdDimension"),
+        # bootstrap-seeding: new participant, no prior state
+        ("case/nodes/participant/owner.py", "RmDimension"),
+        ("case/nodes/participant/common.py", "RmDimension"),
+        ("case/nodes/participant/common.py", "VfdDimension"),
+        # guard-predicate: VfdDimension used as helper, not persisted
+        ("report/nodes/develop_fix.py", "VfdDimension"),
+        ("report/nodes/deploy_fix.py", "VfdDimension"),
+        # rm-transitions: guarded by upstream BT conditions
+        ("report/nodes/rm_transitions.py", "RmDimension"),
+        # case-proposal: bootstrap on first receipt
+        ("case/case_proposal_received_tree.py", "RmDimension"),
+    }
+)
+
+
+def _constructor_name(call: ast.Call) -> str | None:
+    """Return the constructor name if *call* is a tracked dimension write, else None."""
+    func = call.func
+    name = None
+    if isinstance(func, ast.Name):
+        name = func.id
+    elif isinstance(func, ast.Attribute):
+        name = func.attr
+    if name not in _WRITE_CONSTRUCTORS:
+        return None
+    if not any(kw.arg == "state" for kw in call.keywords):
+        return None
+    return name
+
+
+def _write_sites_in_file(py_file: Path, root: Path) -> set[tuple[str, str]]:
+    """Return (rel_path, constructor_name) pairs for each write site in *py_file*."""
+    try:
+        tree = ast.parse(py_file.read_text(encoding="utf-8"))
+    except (OSError, SyntaxError):
+        return set()
+    rel = str(py_file.relative_to(root))
+    return {
+        (rel, name)
+        for node in ast.walk(tree)
+        if isinstance(node, ast.Call)
+        for name in (_constructor_name(node),)
+        if name is not None
+    }
+
+
+def _collect_write_sites(root: Path) -> set[tuple[str, str]]:
+    """Return (relative_path, constructor_name) pairs for each write site found."""
+    found: set[tuple[str, str]] = set()
+    for py_file in root.rglob("*.py"):
+        if "__pycache__" in py_file.parts:
+            continue
+        found |= _write_sites_in_file(py_file, root)
+    return found
+
+
+def test_vfd_rm_pxa_write_sites_match_audit():
+    """AC-7: dimension write sites in behaviors/ match the audited set exactly."""
+    found = _collect_write_sites(_BEHAVIORS_ROOT)
+
+    new_sites = found - AUDITED_WRITE_SITES
+    removed_sites = AUDITED_WRITE_SITES - found
+
+    messages: list[str] = []
+    if new_sites:
+        lines = "\n".join(
+            f"  {rel} — {ctor}" for rel, ctor in sorted(new_sites)
+        )
+        messages.append(
+            f"NEW unaudited write sites found (add to AUDITED_WRITE_SITES"
+            f" after classifying):\n{lines}"
+        )
+    if removed_sites:
+        lines = "\n".join(
+            f"  {rel} — {ctor}" for rel, ctor in sorted(removed_sites)
+        )
+        messages.append(
+            f"Previously-audited write sites no longer present"
+            f" (remove from AUDITED_WRITE_SITES):\n{lines}"
+        )
+
+    assert not messages, "\n\n".join(messages)

test/core/behaviors/report/test_develop_fix_tree.py

@@ -474,6 +474,7 @@ def test_success_and_creates_vfd_status(
         case_with_vendor: VultronCase,
     ) -> None:
         _seed_rm_state(bt_scenario, CASE_ID, VENDOR_ACTOR_ID, RM.ACCEPTED)
+        _seed_vfd_state(bt_scenario, CASE_ID, VENDOR_ACTOR_ID, CS_vfd.Vfd)
         result_out: dict = {}
         node = TransitionCStoFixReady(
             case_id=CASE_ID, actor_id=VENDOR_ACTOR_ID, result_out=result_out
@@ -507,6 +508,7 @@ def test_fix_ready_logged_in_narrative_form(
         import logging
 
         _seed_rm_state(bt_scenario, CASE_ID, VENDOR_ACTOR_ID, RM.ACCEPTED)
+        _seed_vfd_state(bt_scenario, CASE_ID, VENDOR_ACTOR_ID, CS_vfd.Vfd)
         node = TransitionCStoFixReady(
             case_id=CASE_ID, actor_id=VENDOR_ACTOR_ID, result_out={}
         )
@@ -524,9 +526,9 @@ def test_fix_ready_logged_in_narrative_form(
         ]
         assert narrative, "Expected a CS narrative line at INFO"
         message = narrative[0].getMessage()
-        # The fixture participant starts at `vfd`, so this single write
-        # advances two sub-dimensions and the label names both.
-        assert f"Actor '{VENDOR_ACTOR_ID}' CS: vfd → VFd" in message
+        # The fixture participant starts at `Vfd` (seeded above), so this write
+        # advances the fix-ready sub-dimension only.
+        assert f"Actor '{VENDOR_ACTOR_ID}' CS: Vfd → VFd" in message
         assert "fix ready" in message
 
         detail = [
@@ -584,6 +586,7 @@ def test_success_emits_cf_activity(
     ) -> None:
         """SUCCESS when status_id present and CASE_MANAGER participant exists."""
         _seed_rm_state(bt_scenario, CASE_ID, VENDOR_ACTOR_ID, RM.ACCEPTED)
+        _seed_vfd_state(bt_scenario, CASE_ID, VENDOR_ACTOR_ID, CS_vfd.Vfd)
         result_out: dict = {}
         transition_node = TransitionCStoFixReady(
             case_id=CASE_ID, actor_id=VENDOR_ACTOR_ID, result_out=result_out

test/core/behaviors/sync/test_announce_tree.py

@@ -17,8 +17,11 @@
 from vultron.core.models.case_ledger import HashChainLedgerRecord
 from vultron.core.models.case_ledger_entry import VultronCaseLedgerEntry
 from vultron.core.models.case_participant import CaseParticipant
+from vultron.core.models.dimensions import RmDimension, VfdDimension
 from vultron.core.models.events.sync import AnnounceLogEntryReceivedEvent
+from vultron.core.models.participant_status import ParticipantStatus
 from vultron.core.ports.sync_activity import SyncActivityPort
+from vultron.core.states.cs import CS_vfd
 from vultron.core.states.em import EM
 from vultron.core.states.rm import RM
 from vultron.core.behaviors.sync.nodes.chain import _to_persistable_entry
@@ -730,14 +733,24 @@ def _make_close_case_entry(
 def _make_case_with_departing_participant(
     datalayer: SqliteDataLayer,
 ) -> as_VulnerabilityCase:
-    """Seed CASE_ID with a departing participant so the apply node can find them."""
+    """Seed CASE_ID with a departing participant at RM.ACCEPTED so CLOSE is valid."""
     case = as_VulnerabilityCase(id_=CASE_ID, attributed_to=OWNER_ACTOR_ID)
     participant = CaseParticipant(
         id_=DEPARTING_PARTICIPANT_ID,
         attributed_to=DEPARTING_ACTOR_ID,
         context=CASE_ID,
     )
     datalayer.create(participant)
+    # Seed at RM.ACCEPTED so ACCEPTED→CLOSED is a valid RM transition.
+    seed_status = ParticipantStatus(
+        context=CASE_ID,
+        attributed_to=DEPARTING_ACTOR_ID,
+        rm=RmDimension(state=RM.ACCEPTED),
+        vfd=VfdDimension(state=CS_vfd.vfd),
+    )
+    datalayer.create(seed_status)
+    participant.participant_statuses.append(seed_status)
+    datalayer.save(participant)
     case.actor_participant_index[DEPARTING_ACTOR_ID] = DEPARTING_PARTICIPANT_ID
     datalayer.save(case)
     return case

test/core/use_cases/received/test_close_case_role_semantics.py

@@ -43,6 +43,9 @@
 from vultron.core.models.events.case import CloseCaseReceivedEvent
 from vultron.core.models.events.sync import AnnounceLogEntryReceivedEvent
 from vultron.core.ports.sync_activity import SyncActivityPort
+from vultron.core.models.dimensions import RmDimension, VfdDimension
+from vultron.core.models.participant_status import ParticipantStatus
+from vultron.core.states.cs import CS_vfd
 from vultron.core.states.rm import RM
 from vultron.core.use_cases.received.case.lifecycle import (
     CloseCaseReceivedUseCase,
@@ -88,6 +91,28 @@ def _make_dl() -> SqliteDataLayer:
     return SqliteDataLayer("sqlite:///:memory:")
 
 
+def _seed_rm(dl: SqliteDataLayer, case_id: str, actor_id: str, rm: RM) -> None:
+    """Append a ParticipantStatus at *rm* so CLOSED writes have a valid source."""
+    case = dl.read(case_id)
+    if not isinstance(case, VulnerabilityCase):
+        return
+    participant_id = case.actor_participant_index.get(actor_id)
+    if not participant_id:
+        return
+    participant = dl.read(participant_id)
+    if not isinstance(participant, CaseParticipant):
+        return
+    status = ParticipantStatus(
+        context=case_id,
+        attributed_to=actor_id,
+        rm=RmDimension(state=rm),
+        vfd=VfdDimension(state=CS_vfd.vfd),
+    )
+    dl.create(status)
+    participant.participant_statuses.append(status)
+    dl.save(participant)
+
+
 def _make_full_dl(
     owner_id: str = OWNER_ID,
     extra_participant_id: str | None = VENDOR_ID,
@@ -141,6 +166,11 @@ def _make_full_dl(
         )
 
     dl.save(case)
+
+    # Seed all participants at RM.ACCEPTED so ACCEPTED→CLOSED is a valid transition.
+    for actor_id in list(case.actor_participant_index.keys()):
+        _seed_rm(dl, CASE_ID, actor_id, RM.ACCEPTED)
+
     return dl