From 39b0a4aa6b32aacdb69e527008b5c64d2d9f65cc Mon Sep 17 00:00:00 2001
From: "Allen D. Householder" <adh@cert.org>
Date: Sat, 8 Aug 2026 01:40:58 +0000
Subject: [PATCH 1/3] =?UTF-8?q?feat:=20migrate=20core/behaviors/=20Ports?=
 =?UTF-8?q?=20(1/5)=20=E2=80=94=20trivial=20base-only=20reparent?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Migrate all Type A BT nodes (no domain-specific register_key() calls)
from DataLayerCondition/DataLayerAction to DataLayerConditionWithPorts/
DataLayerActionWithPorts across note/, status/, and case/ domains.

AC-1: 30 nodes migrated across 14 source files
AC-2: Remove trivial no-op setup() overrides from EmitOfferCaseOwnershipTransferNode
      and EmitAcceptCaseOwnershipTransferNode (ownership_transfer.py)
AC-3: All 1677 existing tests pass
AC-4: 21 new typed-ports tests (NoDataAvailable isolation + BTTestScenario
      happy-path) in 3 new test files across note/, status/, case/ domains

Intentionally NOT migrated (have domain register_key() calls):
CheckIsCaseManagerNode, CheckIsCaseOwnerNode, EmitCloseCaseNode,
BroadcastCaseUpdateNode, CaptureCaseUpdateBroadcastExclusionsNode,
EmitOfferCaseParticipantToOwnerNode

Misc: condense suggest_actor/emit.py docstring to stay within BTND-07-004
500-line limit after black reformatted multi-line imports (498 lines).

Closes #1883

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
---
 .../behaviors/case/nodes/test_typed_ports.py  | 135 +++++++++++++++
 .../behaviors/note/nodes/test_typed_ports.py  | 135 +++++++++++++++
 .../status/nodes/test_typed_ports.py          | 162 ++++++++++++++++++
 vultron/core/behaviors/case/nodes/announce.py |   4 +-
 .../case/nodes/case_participant_received.py   |   6 +-
 .../core/behaviors/case/nodes/conditions.py   |  15 +-
 .../behaviors/case/nodes/invite_response.py   |   6 +-
 vultron/core/behaviors/case/nodes/leave.py    |   6 +-
 .../case/nodes/ownership_transfer.py          |  14 +-
 .../case/nodes/participant/_bootstrap.py      |   4 +-
 .../case/nodes/participant/status.py          |   4 +-
 vultron/core/behaviors/case/nodes/proposal.py |   4 +-
 .../case/nodes/suggest_actor/accept_offer.py  |   4 +-
 .../case/nodes/suggest_actor/conditions.py    |   8 +-
 .../case/nodes/suggest_actor/emit.py          |  18 +-
 vultron/core/behaviors/case/nodes/update.py   |   3 +-
 .../behaviors/case/nodes/vfd_role_guards.py   |   9 +-
 vultron/core/behaviors/note/nodes/creation.py |   6 +-
 vultron/core/behaviors/note/nodes/storage.py  |   6 +-
 .../behaviors/status/nodes/case_status.py     |  11 +-
 .../core/behaviors/status/nodes/conditions.py |   6 +-
 .../core/behaviors/status/nodes/lifecycle.py  |  10 +-
 .../status/nodes/threat_termination.py        |   4 +-
 23 files changed, 511 insertions(+), 69 deletions(-)
 create mode 100644 test/core/behaviors/case/nodes/test_typed_ports.py
 create mode 100644 test/core/behaviors/note/nodes/test_typed_ports.py
 create mode 100644 test/core/behaviors/status/nodes/test_typed_ports.py

diff --git a/test/core/behaviors/case/nodes/test_typed_ports.py b/test/core/behaviors/case/nodes/test_typed_ports.py
new file mode 100644
index 000000000..aaa6a40b4
--- /dev/null
+++ b/test/core/behaviors/case/nodes/test_typed_ports.py
@@ -0,0 +1,135 @@
+#!/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
+
+"""Typed-Ports isolation tests for case domain nodes (AC-4, issue #1883).
+
+Covers BTND-03-011 (NoDataAvailable on missing required port) and happy-path
+execution via BTTestScenario for one representative node per case sub-module.
+"""
+
+import pytest
+from py_trees.ports import NoDataAvailable
+
+from vultron.core.behaviors.case.nodes.conditions import (
+    CheckCaseAlreadyExists,
+)
+from vultron.core.behaviors.case.nodes.suggest_actor.conditions import (
+    ActorAlreadyParticipantNode,
+)
+from vultron.core.behaviors.case.nodes.vfd_role_guards import (
+    CheckVendorRoleNode,
+)
+from vultron.core.models.case import VulnerabilityCase
+from test.core.behaviors.bt_harness import BTTestScenario
+
+ACTOR_ID = "https://example.org/actors/vendor"
+CASE_ID = "https://example.org/cases/case-001"
+PARTICIPANT_ID = "https://example.org/participants/p-001"
+
+
+# ---------------------------------------------------------------------------
+# conditions.py — CheckCaseAlreadyExists
+# ---------------------------------------------------------------------------
+
+
+class TestCheckCaseAlreadyExistsPorts:
+    def test_missing_datalayer_raises_no_data_available(self) -> None:
+        node = CheckCaseAlreadyExists(case_id=CASE_ID)
+        node.setup_ports()
+        with pytest.raises(NoDataAvailable):
+            node.get_input("datalayer")
+
+    def test_failure_when_case_not_present(
+        self, bt_scenario: BTTestScenario
+    ) -> None:
+        result = bt_scenario.run(
+            CheckCaseAlreadyExists(case_id=CASE_ID), actor_id=ACTOR_ID
+        )
+        bt_scenario.assert_failure(result)
+
+    def test_success_when_case_has_participants(
+        self, bt_scenario: BTTestScenario
+    ) -> None:
+        from vultron.core.models.case_participant import CaseParticipant
+
+        participant = CaseParticipant(
+            id_=PARTICIPANT_ID,
+            attributed_to=ACTOR_ID,
+        )
+        case = VulnerabilityCase(
+            id_=CASE_ID,
+            name="Test Case",
+            attributed_to=ACTOR_ID,
+        )
+        case.case_participants.append(participant)
+        bt_scenario.seed(case, participant)
+        result = bt_scenario.run(
+            CheckCaseAlreadyExists(case_id=CASE_ID), actor_id=ACTOR_ID
+        )
+        bt_scenario.assert_success(result)
+
+
+# ---------------------------------------------------------------------------
+# vfd_role_guards.py — CheckVendorRoleNode
+# ---------------------------------------------------------------------------
+
+
+class TestCheckVendorRoleNodePorts:
+    def test_missing_datalayer_raises_no_data_available(self) -> None:
+        node = CheckVendorRoleNode(case_id=CASE_ID, actor_id=ACTOR_ID)
+        node.setup_ports()
+        with pytest.raises(NoDataAvailable):
+            node.get_input("datalayer")
+
+    def test_failure_when_case_not_found(
+        self, bt_scenario: BTTestScenario
+    ) -> None:
+        result = bt_scenario.run(
+            CheckVendorRoleNode(case_id=CASE_ID, actor_id=ACTOR_ID),
+            actor_id=ACTOR_ID,
+        )
+        bt_scenario.assert_failure(result)
+
+
+# ---------------------------------------------------------------------------
+# suggest_actor/conditions.py — ActorAlreadyParticipantNode
+# ---------------------------------------------------------------------------
+
+
+class TestActorAlreadyParticipantNodePorts:
+    def test_missing_datalayer_raises_no_data_available(self) -> None:
+        node = ActorAlreadyParticipantNode(
+            recommended_id=ACTOR_ID, case_id=CASE_ID
+        )
+        node.setup_ports()
+        with pytest.raises(NoDataAvailable):
+            node.get_input("datalayer")
+
+    def test_failure_when_actor_not_participant(
+        self, bt_scenario: BTTestScenario
+    ) -> None:
+        case = VulnerabilityCase(
+            id_=CASE_ID,
+            name="Test Case",
+            attributed_to=ACTOR_ID,
+        )
+        bt_scenario.seed(case)
+        result = bt_scenario.run(
+            ActorAlreadyParticipantNode(
+                recommended_id=ACTOR_ID, case_id=CASE_ID
+            ),
+            actor_id=ACTOR_ID,
+        )
+        bt_scenario.assert_failure(result)
diff --git a/test/core/behaviors/note/nodes/test_typed_ports.py b/test/core/behaviors/note/nodes/test_typed_ports.py
new file mode 100644
index 000000000..e431193c2
--- /dev/null
+++ b/test/core/behaviors/note/nodes/test_typed_ports.py
@@ -0,0 +1,135 @@
+#!/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
+
+"""Typed-Ports isolation tests for note domain nodes (AC-4, issue #1883).
+
+Covers BTND-03-011 (NoDataAvailable on missing required port) and happy-path
+execution via BTTestScenario for one representative node per note sub-module.
+"""
+
+import pytest
+from py_trees.ports import NoDataAvailable
+
+from vultron.core.behaviors.note.nodes.creation import (
+    CreateNoteNode,
+)
+from vultron.core.behaviors.note.nodes.storage import (
+    AttachNoteToCaseNode,
+    SaveNoteNode,
+)
+from vultron.core.models.case import VulnerabilityCase
+from vultron.core.models.note import VultronNote
+from test.core.behaviors.bt_harness import BTTestScenario
+
+ACTOR_ID = "https://example.org/actors/vendor"
+CASE_ID = "https://example.org/cases/case-001"
+NOTE_ID = "https://example.org/notes/note-001"
+
+
+# ---------------------------------------------------------------------------
+# creation.py — SaveNoteNode (isolated-port: BTND-03-011)
+# ---------------------------------------------------------------------------
+
+
+class TestSaveNoteNodePorts:
+    def test_missing_datalayer_raises_no_data_available(self) -> None:
+        note = VultronNote(
+            id_=NOTE_ID,
+            name="Test Note",
+            content="test content",
+            context=CASE_ID,
+            attributed_to=ACTOR_ID,
+        )
+        node = SaveNoteNode(note_obj=note)
+        node.setup_ports()
+        with pytest.raises(NoDataAvailable):
+            node.get_input("datalayer")
+
+    def test_saves_note_via_bt_scenario(
+        self, bt_scenario: BTTestScenario
+    ) -> None:
+        note = VultronNote(
+            id_=NOTE_ID,
+            name="Test Note",
+            content="test content",
+            context=CASE_ID,
+            attributed_to=ACTOR_ID,
+        )
+        result = bt_scenario.run(
+            SaveNoteNode(note_obj=note), actor_id=ACTOR_ID
+        )
+        bt_scenario.assert_success(result)
+
+
+# ---------------------------------------------------------------------------
+# creation.py — CreateNoteNode (isolated-port: BTND-03-011)
+# ---------------------------------------------------------------------------
+
+
+class TestCreateNoteNodePorts:
+    def test_missing_datalayer_raises_no_data_available(self) -> None:
+        result_out: dict = {}
+        node = CreateNoteNode(
+            note_name="N",
+            note_content="c",
+            case_id=CASE_ID,
+            result_out=result_out,
+        )
+        node.setup_ports()
+        with pytest.raises(NoDataAvailable):
+            node.get_input("datalayer")
+
+
+# ---------------------------------------------------------------------------
+# storage.py — AttachNoteToCaseNode (isolated-port: BTND-03-011)
+# ---------------------------------------------------------------------------
+
+
+class TestAttachNoteToCaseNodePorts:
+    def test_missing_datalayer_raises_no_data_available(self) -> None:
+        node = AttachNoteToCaseNode(note_id=NOTE_ID, case_id=CASE_ID)
+        node.setup_ports()
+        with pytest.raises(NoDataAvailable):
+            node.get_input("datalayer")
+
+    def test_skips_when_case_id_none(
+        self, bt_scenario: BTTestScenario
+    ) -> None:
+        """Skips silently (SUCCESS) when case_id is None."""
+        result = bt_scenario.run(
+            AttachNoteToCaseNode(note_id=NOTE_ID, case_id=None),
+            actor_id=ACTOR_ID,
+        )
+        bt_scenario.assert_success(result)
+
+    def test_attaches_note_to_case(self, bt_scenario: BTTestScenario) -> None:
+        case = VulnerabilityCase(
+            id_=CASE_ID,
+            name="Test Case",
+            attributed_to=ACTOR_ID,
+        )
+        note = VultronNote(
+            id_=NOTE_ID,
+            name="Test Note",
+            content="test content",
+            context=CASE_ID,
+            attributed_to=ACTOR_ID,
+        )
+        bt_scenario.seed(case, note)
+        result = bt_scenario.run(
+            AttachNoteToCaseNode(note_id=NOTE_ID, case_id=CASE_ID),
+            actor_id=ACTOR_ID,
+        )
+        bt_scenario.assert_success(result)
diff --git a/test/core/behaviors/status/nodes/test_typed_ports.py b/test/core/behaviors/status/nodes/test_typed_ports.py
new file mode 100644
index 000000000..f2f7f7c96
--- /dev/null
+++ b/test/core/behaviors/status/nodes/test_typed_ports.py
@@ -0,0 +1,162 @@
+#!/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
+
+"""Typed-Ports isolation tests for status domain nodes (AC-4, issue #1883).
+
+Covers BTND-03-011 (NoDataAvailable on missing required port) and happy-path
+execution via BTTestScenario for one representative node per status sub-module.
+"""
+
+import pytest
+from py_trees.ports import NoDataAvailable
+
+from vultron.core.behaviors.status.nodes.case_status import (
+    CheckCaseStatusIdempotencyNode,
+)
+from vultron.core.behaviors.status.nodes.conditions import (
+    AllParticipantsRMClosedConditionNode,
+)
+from vultron.core.behaviors.status.nodes.lifecycle import (
+    _PublicDisclosureSkipConditionNode,
+)
+from vultron.core.behaviors.status.nodes.threat_termination import (
+    _ThreatTerminationSkipConditionNode,
+)
+from vultron.core.models.case import VulnerabilityCase
+from test.core.behaviors.bt_harness import BTTestScenario
+
+ACTOR_ID = "https://example.org/actors/vendor"
+CASE_ID = "https://example.org/cases/case-001"
+STATUS_ID = "https://example.org/statuses/status-001"
+
+
+# ---------------------------------------------------------------------------
+# case_status.py — CheckCaseStatusIdempotencyNode
+# ---------------------------------------------------------------------------
+
+
+class TestCheckCaseStatusIdempotencyNodePorts:
+    def test_missing_datalayer_raises_no_data_available(self) -> None:
+        node = CheckCaseStatusIdempotencyNode(
+            case_id=CASE_ID, status_id=STATUS_ID
+        )
+        node.setup_ports()
+        with pytest.raises(NoDataAvailable):
+            node.get_input("datalayer")
+
+    def test_success_when_status_not_yet_present(
+        self, bt_scenario: BTTestScenario
+    ) -> None:
+        case = VulnerabilityCase(
+            id_=CASE_ID,
+            name="Test Case",
+            attributed_to=ACTOR_ID,
+        )
+        bt_scenario.seed(case)
+        result = bt_scenario.run(
+            CheckCaseStatusIdempotencyNode(
+                case_id=CASE_ID, status_id=STATUS_ID
+            ),
+            actor_id=ACTOR_ID,
+        )
+        bt_scenario.assert_success(result)
+
+
+# ---------------------------------------------------------------------------
+# conditions.py — AllParticipantsRMClosedConditionNode
+# ---------------------------------------------------------------------------
+
+
+class TestAllParticipantsRMClosedConditionNodePorts:
+    def test_missing_datalayer_raises_no_data_available(self) -> None:
+        node = AllParticipantsRMClosedConditionNode(case_id=CASE_ID)
+        node.setup_ports()
+        with pytest.raises(NoDataAvailable):
+            node.get_input("datalayer")
+
+    def test_failure_when_no_participants(
+        self, bt_scenario: BTTestScenario
+    ) -> None:
+        case = VulnerabilityCase(
+            id_=CASE_ID,
+            name="Test Case",
+            attributed_to=ACTOR_ID,
+        )
+        bt_scenario.seed(case)
+        result = bt_scenario.run(
+            AllParticipantsRMClosedConditionNode(case_id=CASE_ID),
+            actor_id=ACTOR_ID,
+        )
+        bt_scenario.assert_failure(result)
+
+
+# ---------------------------------------------------------------------------
+# lifecycle.py — _PublicDisclosureSkipConditionNode
+# ---------------------------------------------------------------------------
+
+
+class TestPublicDisclosureSkipConditionNodePorts:
+    def test_missing_datalayer_raises_no_data_available(self) -> None:
+        node = _PublicDisclosureSkipConditionNode(
+            status_obj=None,
+            sender_actor_id=ACTOR_ID,
+            case_id=CASE_ID,
+        )
+        node.setup_ports()
+        with pytest.raises(NoDataAvailable):
+            node.get_input("datalayer")
+
+    def test_success_skip_when_no_public_aware_status(
+        self, bt_scenario: BTTestScenario
+    ) -> None:
+        """Non-public-aware status → skip condition returns SUCCESS."""
+        result = bt_scenario.run(
+            _PublicDisclosureSkipConditionNode(
+                status_obj=None,
+                sender_actor_id=ACTOR_ID,
+                case_id=CASE_ID,
+            ),
+            actor_id=ACTOR_ID,
+        )
+        bt_scenario.assert_success(result)
+
+
+# ---------------------------------------------------------------------------
+# threat_termination.py — _ThreatTerminationSkipConditionNode
+# ---------------------------------------------------------------------------
+
+
+class TestThreatTerminationSkipConditionNodePorts:
+    def test_missing_datalayer_raises_no_data_available(self) -> None:
+        node = _ThreatTerminationSkipConditionNode(
+            status_obj=None,
+            case_id=CASE_ID,
+        )
+        node.setup_ports()
+        with pytest.raises(NoDataAvailable):
+            node.get_input("datalayer")
+
+    def test_success_skip_when_no_threat(
+        self, bt_scenario: BTTestScenario
+    ) -> None:
+        """No threat in status → skip returns SUCCESS."""
+        result = bt_scenario.run(
+            _ThreatTerminationSkipConditionNode(
+                status_obj=None,
+                case_id=CASE_ID,
+            ),
+            actor_id=ACTOR_ID,
+        )
+        bt_scenario.assert_success(result)
diff --git a/vultron/core/behaviors/case/nodes/announce.py b/vultron/core/behaviors/case/nodes/announce.py
index c26e21e31..22b507dfb 100644
--- a/vultron/core/behaviors/case/nodes/announce.py
+++ b/vultron/core/behaviors/case/nodes/announce.py
@@ -28,14 +28,14 @@
 
 from py_trees.common import Status
 
-from vultron.core.behaviors.helpers import DataLayerAction
+from vultron.core.behaviors.helpers import DataLayerActionWithPorts
 from vultron.core.models.events.actor import (
     AnnounceVulnerabilityCaseReceivedEvent,
 )
 from vultron.core.models.case import VulnerabilityCase
 
 
-class SeedAnnouncedCaseNode(DataLayerAction):
+class SeedAnnouncedCaseNode(DataLayerActionWithPorts):
     """Persist a received ``VulnerabilityCase`` announcement in the DataLayer.
 
     On first receipt the node saves ``case_obj`` and stores any embedded
diff --git a/vultron/core/behaviors/case/nodes/case_participant_received.py b/vultron/core/behaviors/case/nodes/case_participant_received.py
index 2b540ff86..d35c2176f 100644
--- a/vultron/core/behaviors/case/nodes/case_participant_received.py
+++ b/vultron/core/behaviors/case/nodes/case_participant_received.py
@@ -25,13 +25,13 @@
 
 from py_trees.common import Status
 
-from vultron.core.behaviors.helpers import DataLayerAction
+from vultron.core.behaviors.helpers import DataLayerActionWithPorts
 from vultron.core.models._helpers import _as_id
 from vultron.core.models.case import VulnerabilityCase
 from vultron.core.models.case_participant import CaseParticipant
 
 
-class AddCaseParticipantToCaseReceivedNode(DataLayerAction):
+class AddCaseParticipantToCaseReceivedNode(DataLayerActionWithPorts):
     """Add a participant to a case and persist the updated case.
 
     Reads both the participant and the case from the DataLayer, calls
@@ -89,7 +89,7 @@ def update(self) -> Status:
         return Status.SUCCESS
 
 
-class RemoveCaseParticipantFromCaseReceivedNode(DataLayerAction):
+class RemoveCaseParticipantFromCaseReceivedNode(DataLayerActionWithPorts):
     """Remove a participant from a case and persist the updated case.
 
     Reads the case from the DataLayer and calls
diff --git a/vultron/core/behaviors/case/nodes/conditions.py b/vultron/core/behaviors/case/nodes/conditions.py
index 6e638d9ba..9081673f2 100644
--- a/vultron/core/behaviors/case/nodes/conditions.py
+++ b/vultron/core/behaviors/case/nodes/conditions.py
@@ -35,7 +35,12 @@
 
 from vultron.config import get_config
 from vultron.core.behaviors.case.nodes.case_setup import _derive_case_slug
-from vultron.core.behaviors.helpers import DataLayerAction, DataLayerCondition
+from vultron.core.behaviors.helpers import (
+    DataLayerActionWithPorts,
+    DataLayerCondition,
+    DataLayerConditionWithPorts,
+)
+
 from vultron.config.actor import ActorConfig
 from vultron.core.models.case import VulnerabilityCase
 from vultron.core.models.case_actor import CaseActor as VultronCaseActor
@@ -98,7 +103,7 @@ def update(self) -> Status:
         return Status.FAILURE
 
 
-class CheckCaseAlreadyExists(DataLayerCondition):
+class CheckCaseAlreadyExists(DataLayerConditionWithPorts):
     """
     Check if a VulnerabilityCase already exists in DataLayer.
 
@@ -150,7 +155,7 @@ def update(self) -> Status:
             return Status.FAILURE
 
 
-class CheckCaseExistsForReport(DataLayerCondition):
+class CheckCaseExistsForReport(DataLayerConditionWithPorts):
     """
     Check if a VulnerabilityCase already exists for the given report.
 
@@ -279,7 +284,7 @@ def update(self) -> Status:
         return Status.FAILURE
 
 
-class CheckPendingProposalExistsForReport(DataLayerCondition):
+class CheckPendingProposalExistsForReport(DataLayerConditionWithPorts):
     """Return SUCCESS when a pending ``VultronReportCaseLink`` exists for the report.
 
     Used as the idempotency guard in the slimmed vendor tree (ADR-0041).
@@ -334,7 +339,7 @@ def update(self) -> Status:
             return Status.FAILURE
 
 
-class WritePendingReportCaseLinkNode(DataLayerAction):
+class WritePendingReportCaseLinkNode(DataLayerActionWithPorts):
     """Write a pending ``VultronReportCaseLink`` for the given report (ADR-0041).
 
     Creates or updates the link with ``case_id=None`` and sets
diff --git a/vultron/core/behaviors/case/nodes/invite_response.py b/vultron/core/behaviors/case/nodes/invite_response.py
index ba8f7646f..c33348848 100644
--- a/vultron/core/behaviors/case/nodes/invite_response.py
+++ b/vultron/core/behaviors/case/nodes/invite_response.py
@@ -31,13 +31,13 @@
 
 from py_trees.common import Status
 
-from vultron.core.behaviors.helpers import DataLayerAction
+from vultron.core.behaviors.helpers import DataLayerActionWithPorts
 from vultron.core.ports.case_persistence import CaseOutboxPersistence
 
 logger = logging.getLogger(__name__)
 
 
-class EmitAcceptCaseInviteNode(DataLayerAction):
+class EmitAcceptCaseInviteNode(DataLayerActionWithPorts):
     """Create Accept(Invite) and queue in the invitee's outbox.
 
     Uses ``trigger_activity_factory.accept_case_invite()`` — the factory
@@ -88,7 +88,7 @@ def update(self) -> Status:
             return Status.FAILURE
 
 
-class EmitRejectCaseInviteNode(DataLayerAction):
+class EmitRejectCaseInviteNode(DataLayerActionWithPorts):
     """Create Reject(Invite) and queue in the invitee's outbox.
 
     Uses ``trigger_activity_factory.reject_case_invite()`` — the factory
diff --git a/vultron/core/behaviors/case/nodes/leave.py b/vultron/core/behaviors/case/nodes/leave.py
index 08d20650c..7b5129350 100644
--- a/vultron/core/behaviors/case/nodes/leave.py
+++ b/vultron/core/behaviors/case/nodes/leave.py
@@ -37,7 +37,7 @@
 from vultron.core.behaviors.case.nodes.participant.status import (
     CreateParticipantStatusNode,
 )
-from vultron.core.behaviors.helpers import DataLayerAction
+from vultron.core.behaviors.helpers import DataLayerActionWithPorts
 from vultron.core.models.case import VulnerabilityCase
 from vultron.core.models.case_participant import CaseParticipant
 from vultron.core.states.rm import RM
@@ -45,7 +45,7 @@
 logger = logging.getLogger(__name__)
 
 
-class AdvanceParticipantToRMClosedNode(DataLayerAction):
+class AdvanceParticipantToRMClosedNode(DataLayerActionWithPorts):
     """Advance the leaving actor's RM state to ``RM.CLOSED`` in the DataLayer.
 
     Reads the leaving actor's :class:`~vultron.core.models.case_participant
@@ -152,7 +152,7 @@ def update(self) -> Status:
         return Status.SUCCESS
 
 
-class AdvanceCaseActorToRMClosedNode(DataLayerAction):
+class AdvanceCaseActorToRMClosedNode(DataLayerActionWithPorts):
     """Advance the Case Actor's own RM state to ``RM.CLOSED``.
 
     Reads the Case Actor's :class:`~vultron.core.models.case_participant
diff --git a/vultron/core/behaviors/case/nodes/ownership_transfer.py b/vultron/core/behaviors/case/nodes/ownership_transfer.py
index 33b12f87c..6d29c13f4 100644
--- a/vultron/core/behaviors/case/nodes/ownership_transfer.py
+++ b/vultron/core/behaviors/case/nodes/ownership_transfer.py
@@ -38,7 +38,7 @@
 
 from py_trees.common import Status
 
-from vultron.core.behaviors.helpers import DataLayerAction
+from vultron.core.behaviors.helpers import DataLayerActionWithPorts
 from vultron.core.models.case import VulnerabilityCase
 from vultron.core.models.case_participant import CaseParticipant
 from vultron.core.models._helpers import _as_id
@@ -49,7 +49,7 @@
 logger = logging.getLogger(__name__)
 
 
-class EmitOfferCaseOwnershipTransferNode(DataLayerAction):
+class EmitOfferCaseOwnershipTransferNode(DataLayerActionWithPorts):
     """Emit ``Offer(VulnerabilityCase)`` (ownership transfer) to ``transferee_id``.
 
     Calls ``trigger_activity_factory.offer_case_ownership_transfer()`` with
@@ -71,9 +71,6 @@ def __init__(
         self.content = content
         self._captured = captured
 
-    def setup(self, **kwargs: Any) -> None:
-        super().setup(**kwargs)
-
     def _emit(self) -> tuple[str, dict]:
         assert self.trigger_activity_factory is not None
         assert self.actor_id is not None
@@ -121,7 +118,7 @@ def update(self) -> Status:
             return Status.FAILURE
 
 
-class EmitAcceptCaseOwnershipTransferNode(DataLayerAction):
+class EmitAcceptCaseOwnershipTransferNode(DataLayerActionWithPorts):
     """Emit ``Accept(Offer(VulnerabilityCase))`` (ownership transfer) to offerer.
 
     Calls ``trigger_activity_factory.accept_case_ownership_transfer()``
@@ -141,9 +138,6 @@ def __init__(
         self.case_id = case_id
         self._captured = captured
 
-    def setup(self, **kwargs: Any) -> None:
-        super().setup(**kwargs)
-
     def _emit(self) -> tuple[str, dict]:
         assert self.trigger_activity_factory is not None
         assert self.actor_id is not None
@@ -188,7 +182,7 @@ def update(self) -> Status:
             return Status.FAILURE
 
 
-class AcceptCaseOwnershipTransferNode(DataLayerAction):
+class AcceptCaseOwnershipTransferNode(DataLayerActionWithPorts):
     """Apply an ownership-transfer acceptance to the case record.
 
     Enforces the at-most-one CASE_OWNER invariant atomically (CM-21-001,
diff --git a/vultron/core/behaviors/case/nodes/participant/_bootstrap.py b/vultron/core/behaviors/case/nodes/participant/_bootstrap.py
index a6683665c..cbe41eca3 100644
--- a/vultron/core/behaviors/case/nodes/participant/_bootstrap.py
+++ b/vultron/core/behaviors/case/nodes/participant/_bootstrap.py
@@ -25,12 +25,12 @@
 from vultron.core.behaviors.case.nodes.participant.common import (
     _ensure_reporter_participant,
 )
-from vultron.core.behaviors.helpers import DataLayerAction
+from vultron.core.behaviors.helpers import DataLayerActionWithPorts
 from vultron.core.models.case import VulnerabilityCase
 from vultron.core.models.report_case_link import VultronReportCaseLink
 
 
-class EnsureReporterParticipantAtAcceptedNode(DataLayerAction):
+class EnsureReporterParticipantAtAcceptedNode(DataLayerActionWithPorts):
     """BT leaf node that seeds or upgrades the reporter participant to RM.ACCEPTED.
 
     Called from ``CreateCaseReceivedUseCase._handle_bootstrap`` via BTBridge
diff --git a/vultron/core/behaviors/case/nodes/participant/status.py b/vultron/core/behaviors/case/nodes/participant/status.py
index e294e0ecd..79d91fe68 100644
--- a/vultron/core/behaviors/case/nodes/participant/status.py
+++ b/vultron/core/behaviors/case/nodes/participant/status.py
@@ -20,7 +20,7 @@
 from vultron.core.behaviors.case.nodes.participant.common import (
     resolve_participant_state_from_dl,
 )
-from vultron.core.behaviors.helpers import DataLayerAction
+from vultron.core.behaviors.helpers import DataLayerActionWithPorts
 from vultron.core.behaviors.narrative_log import (
     log_cs_transition,
     log_rm_transition,
@@ -91,7 +91,7 @@ def _resolve_pxa_state(case: object, participant: object) -> CS_pxa:
     return _pxa_from_case(case) or CS_pxa.pxa
 
 
-class CreateParticipantStatusNode(DataLayerAction):
+class CreateParticipantStatusNode(DataLayerActionWithPorts):
     """Create a ParticipantStatus snapshot and append it to the participant."""
 
     def __init__(
diff --git a/vultron/core/behaviors/case/nodes/proposal.py b/vultron/core/behaviors/case/nodes/proposal.py
index c86e1f481..0f48be38c 100644
--- a/vultron/core/behaviors/case/nodes/proposal.py
+++ b/vultron/core/behaviors/case/nodes/proposal.py
@@ -33,11 +33,11 @@
 
 from vultron.config import get_config
 from vultron.core.behaviors.case.nodes.case_setup import _derive_case_slug
-from vultron.core.behaviors.helpers import DataLayerAction
+from vultron.core.behaviors.helpers import DataLayerActionWithPorts
 from vultron.core.ports.case_persistence import CaseOutboxPersistence
 
 
-class ProposeReportCaseToActorNode(DataLayerAction):
+class ProposeReportCaseToActorNode(DataLayerActionWithPorts):
     """Send ``Create(as_CaseProposal)`` from ``report_id`` without a prior case.
 
     Used by the slimmed vendor ``receive_report_case_tree`` (ADR-0041).  Unlike
diff --git a/vultron/core/behaviors/case/nodes/suggest_actor/accept_offer.py b/vultron/core/behaviors/case/nodes/suggest_actor/accept_offer.py
index d6661b33c..9eed94115 100644
--- a/vultron/core/behaviors/case/nodes/suggest_actor/accept_offer.py
+++ b/vultron/core/behaviors/case/nodes/suggest_actor/accept_offer.py
@@ -25,11 +25,11 @@
 
 from py_trees.common import Status
 
-from vultron.core.behaviors.helpers import DataLayerAction
+from vultron.core.behaviors.helpers import DataLayerActionWithPorts
 from vultron.core.ports.case_persistence import CaseOutboxPersistence
 
 
-class EmitAcceptCaseParticipantOfferNode(DataLayerAction):
+class EmitAcceptCaseParticipantOfferNode(DataLayerActionWithPorts):
     """Case Owner sends Accept(Offer(CaseParticipant)) back to the CaseActor.
 
     Triggered by the Case Owner after reviewing the Offer(CaseParticipant)
diff --git a/vultron/core/behaviors/case/nodes/suggest_actor/conditions.py b/vultron/core/behaviors/case/nodes/suggest_actor/conditions.py
index 306f54d21..812c86273 100644
--- a/vultron/core/behaviors/case/nodes/suggest_actor/conditions.py
+++ b/vultron/core/behaviors/case/nodes/suggest_actor/conditions.py
@@ -28,14 +28,14 @@
 
 from py_trees.common import Status
 
-from vultron.core.behaviors.helpers import DataLayerAction
+from vultron.core.behaviors.helpers import DataLayerActionWithPorts
 from vultron.core.models.protocol_pair import (
     INVITE_ACTOR_TO_CASE_REPLY_TYPES,
     OFFER_CASE_PARTICIPANT_REPLY_TYPES,
 )
 
 
-class ActorAlreadyParticipantNode(DataLayerAction):
+class ActorAlreadyParticipantNode(DataLayerActionWithPorts):
     """Return SUCCESS if the recommended actor is already a case participant.
 
     Reads ``VulnerabilityCase.actor_participant_index`` from the DataLayer.
@@ -73,7 +73,7 @@ def update(self) -> Status:
         return Status.FAILURE
 
 
-class InviteInFlightNode(DataLayerAction):
+class InviteInFlightNode(DataLayerActionWithPorts):
     """Return SUCCESS if an Invite to the recommended actor is in-flight.
 
     Queries the case ledger via ``find_protocol_pair`` with
@@ -116,7 +116,7 @@ def update(self) -> Status:
         return Status.FAILURE
 
 
-class PendingOfferCaseParticipantNode(DataLayerAction):
+class PendingOfferCaseParticipantNode(DataLayerActionWithPorts):
     """Return SUCCESS if an Offer(CaseParticipant) to the Case Owner is pending.
 
     Queries the case ledger via ``find_protocol_pair`` with
diff --git a/vultron/core/behaviors/case/nodes/suggest_actor/emit.py b/vultron/core/behaviors/case/nodes/suggest_actor/emit.py
index 928620f89..e14ba5725 100644
--- a/vultron/core/behaviors/case/nodes/suggest_actor/emit.py
+++ b/vultron/core/behaviors/case/nodes/suggest_actor/emit.py
@@ -31,10 +31,7 @@
   ``Create(Note)`` + ``Add(Note, Case)`` to the Case Owner when a duplicate
   recommendation arrives (CM-16-008).
 
-The Case Owner owner-side Accept response
-(:class:`~vultron.core.behaviors.case.nodes.suggest_actor.accept_offer.EmitAcceptCaseParticipantOfferNode`)
-lives in the ``accept_offer`` submodule to keep this module under the
-BTND-07-004 line limit.
+The Case Owner Accept response lives in ``accept_offer`` (BTND-07-004 limit).
 """
 
 from typing import cast
@@ -43,7 +40,10 @@
 from py_trees.common import Status
 
 from vultron.core.behaviors.bridge import BTBridge
-from vultron.core.behaviors.helpers import DataLayerAction
+from vultron.core.behaviors.helpers import (
+    DataLayerAction,
+    DataLayerActionWithPorts,
+)
 from vultron.core.behaviors.sync.commit_tree import (
     create_commit_log_entry_tree,
 )
@@ -55,7 +55,7 @@
 from vultron.enums.roles import CVDRole
 
 
-class RecordRecommendationRecommenderNode(DataLayerAction):
+class RecordRecommendationRecommenderNode(DataLayerActionWithPorts):
     """Write recommendation_id → recommender_id into core case state.
 
     Runs as the first effect node in ``RecommendActorToCaseBT`` so downstream
@@ -232,7 +232,7 @@ def update(self) -> Status:
             return Status.FAILURE
 
 
-class EmitAcceptActorRecommendationNode(DataLayerAction):
+class EmitAcceptActorRecommendationNode(DataLayerActionWithPorts):
     """Queue AcceptActorRecommendation to the original recommender.
 
     Used after the Case Owner accepts Offer(CaseParticipant) (CM-16-006 step 3).
@@ -313,7 +313,7 @@ def update(self) -> Status:
             return Status.FAILURE
 
 
-class EmitRejectActorRecommendationNode(DataLayerAction):
+class EmitRejectActorRecommendationNode(DataLayerActionWithPorts):
     """Queue RejectActorRecommendation to the original recommender.
 
     Used after the Case Owner rejects Offer(CaseParticipant) (CM-16-007 step 3).
@@ -391,7 +391,7 @@ def update(self) -> Status:
             return Status.FAILURE
 
 
-class EmitNoteDuplicateRecommendationToOwnerNode(DataLayerAction):
+class EmitNoteDuplicateRecommendationToOwnerNode(DataLayerActionWithPorts):
     """Send a Note DM to the Case Owner noting reinforcing demand.
 
     Used when a second ``Offer(Actor, Case)`` arrives while a first
diff --git a/vultron/core/behaviors/case/nodes/update.py b/vultron/core/behaviors/case/nodes/update.py
index b6d7c1a41..f8d94f013 100644
--- a/vultron/core/behaviors/case/nodes/update.py
+++ b/vultron/core/behaviors/case/nodes/update.py
@@ -29,6 +29,7 @@
 )
 from vultron.core.behaviors.helpers import (
     DataLayerAction,
+    DataLayerActionWithPorts,
     DataLayerCondition,
 )
 from vultron.core.models.events.case import UpdateCaseReceivedEvent
@@ -107,7 +108,7 @@ def update(self) -> Status:
         return Status.SUCCESS
 
 
-class ApplyCaseUpdateNode(DataLayerAction):
+class ApplyCaseUpdateNode(DataLayerActionWithPorts):
     """Apply mutable fields from the inbound update payload to the case."""
 
     def __init__(
diff --git a/vultron/core/behaviors/case/nodes/vfd_role_guards.py b/vultron/core/behaviors/case/nodes/vfd_role_guards.py
index d12b378b1..c14fa550b 100644
--- a/vultron/core/behaviors/case/nodes/vfd_role_guards.py
+++ b/vultron/core/behaviors/case/nodes/vfd_role_guards.py
@@ -32,7 +32,10 @@
 import py_trees
 from py_trees.common import Status
 
-from vultron.core.behaviors.helpers import DataLayerCondition
+from vultron.core.behaviors.helpers import (
+    DataLayerCondition,
+    DataLayerConditionWithPorts,
+)
 from vultron.core.models.case import VulnerabilityCase
 from vultron.core.models.case_participant import CaseParticipant
 from vultron.core.ports.case_persistence import CasePersistence
@@ -78,7 +81,7 @@ def _resolve_actor_roles(
     return list(participant.roles) if participant.roles else []
 
 
-class CheckVendorRoleNode(DataLayerCondition):
+class CheckVendorRoleNode(DataLayerConditionWithPorts):
     """Gate f→F (vfd_state=VFd): actor MUST hold CVDRole.VENDOR.
 
     Returns ``SUCCESS`` when the executing actor holds ``CVDRole.VENDOR`` in
@@ -131,7 +134,7 @@ def update(self) -> Status:
         return Status.SUCCESS
 
 
-class CheckDeployerRoleNode(DataLayerCondition):
+class CheckDeployerRoleNode(DataLayerConditionWithPorts):
     """Gate d→D (vfd_state=VFD): actor MUST hold CVDRole.DEPLOYER.
 
     Returns ``SUCCESS`` when the executing actor holds ``CVDRole.DEPLOYER`` in
diff --git a/vultron/core/behaviors/note/nodes/creation.py b/vultron/core/behaviors/note/nodes/creation.py
index c4945eaa5..2d85a3022 100644
--- a/vultron/core/behaviors/note/nodes/creation.py
+++ b/vultron/core/behaviors/note/nodes/creation.py
@@ -19,12 +19,12 @@
 
 from py_trees.common import Status
 
-from vultron.core.behaviors.helpers import DataLayerAction
+from vultron.core.behaviors.helpers import DataLayerActionWithPorts
 from vultron.core.models._helpers import _as_id
 from vultron.core.models.case import VulnerabilityCase
 
 
-class CreateNoteNode(DataLayerAction):
+class CreateNoteNode(DataLayerActionWithPorts):
     """Create and persist a Note via the TriggerActivityPort."""
 
     def __init__(
@@ -71,7 +71,7 @@ def update(self) -> Status:
             return Status.FAILURE
 
 
-class AttachNoteFromResultNode(DataLayerAction):
+class AttachNoteFromResultNode(DataLayerActionWithPorts):
     """Attach a note to a case, reading ``note_id`` from ``result_out``."""
 
     def __init__(
diff --git a/vultron/core/behaviors/note/nodes/storage.py b/vultron/core/behaviors/note/nodes/storage.py
index faa8ddaa3..a20ba948b 100644
--- a/vultron/core/behaviors/note/nodes/storage.py
+++ b/vultron/core/behaviors/note/nodes/storage.py
@@ -19,13 +19,13 @@
 
 from py_trees.common import Status
 
-from vultron.core.behaviors.helpers import DataLayerAction
+from vultron.core.behaviors.helpers import DataLayerActionWithPorts
 from vultron.core.models._helpers import _as_id
 from vultron.core.models.case import VulnerabilityCase
 from vultron.core.models.note import VultronNote
 
 
-class SaveNoteNode(DataLayerAction):
+class SaveNoteNode(DataLayerActionWithPorts):
     """Persist a Note to the DataLayer using upsert semantics."""
 
     def __init__(self, note_obj: VultronNote, name: str | None = None):
@@ -48,7 +48,7 @@ def update(self) -> Status:
             return Status.FAILURE
 
 
-class AttachNoteToCaseNode(DataLayerAction):
+class AttachNoteToCaseNode(DataLayerActionWithPorts):
     """Attach a note to a VulnerabilityCase in the DataLayer."""
 
     def __init__(
diff --git a/vultron/core/behaviors/status/nodes/case_status.py b/vultron/core/behaviors/status/nodes/case_status.py
index c475e890d..31c8e3f8c 100644
--- a/vultron/core/behaviors/status/nodes/case_status.py
+++ b/vultron/core/behaviors/status/nodes/case_status.py
@@ -24,7 +24,10 @@
 
 from py_trees.common import Status
 
-from vultron.core.behaviors.helpers import DataLayerAction, DataLayerCondition
+from vultron.core.behaviors.helpers import (
+    DataLayerActionWithPorts,
+    DataLayerConditionWithPorts,
+)
 from vultron.core.behaviors.idempotency import SilentIdempotencyGuardMixin
 from vultron.core.models.case import VulnerabilityCase
 from vultron.core.models.case_status import CaseStatus
@@ -42,7 +45,7 @@
 
 
 class CheckCaseStatusIdempotencyNode(
-    SilentIdempotencyGuardMixin, DataLayerCondition
+    SilentIdempotencyGuardMixin, DataLayerConditionWithPorts
 ):
     """AC-1: Verify the CaseStatus has not already been added to the case.
 
@@ -95,7 +98,7 @@ def update(self) -> Status:
         return Status.SUCCESS
 
 
-class ValidateCaseStatusTransitionNode(DataLayerCondition):
+class ValidateCaseStatusTransitionNode(DataLayerConditionWithPorts):
     """AC-2: Validate that the new CaseStatus represents a legal state transition.
 
     Uses ``case.current_status`` as the reference point.  When the case has no
@@ -196,7 +199,7 @@ def update(self) -> Status:
         return Status.SUCCESS
 
 
-class AppendCaseStatusToCaseNode(DataLayerAction):
+class AppendCaseStatusToCaseNode(DataLayerActionWithPorts):
     """Append the resolved CaseStatus to ``case.case_statuses`` and persist.
 
     Resolves the status object from the DataLayer first; if not found there,
diff --git a/vultron/core/behaviors/status/nodes/conditions.py b/vultron/core/behaviors/status/nodes/conditions.py
index 4adae1a34..d563e1b1b 100644
--- a/vultron/core/behaviors/status/nodes/conditions.py
+++ b/vultron/core/behaviors/status/nodes/conditions.py
@@ -27,7 +27,7 @@
 from py_trees.common import Status
 
 from vultron.core.behaviors.helpers import (
-    DataLayerCondition,
+    DataLayerConditionWithPorts,
     FindParticipantByActorIdNode,
 )
 from vultron.core.models.case import VulnerabilityCase
@@ -111,7 +111,7 @@ def update(self) -> Status:
         return Status.SUCCESS
 
 
-class AllParticipantsRMClosedConditionNode(DataLayerCondition):
+class AllParticipantsRMClosedConditionNode(DataLayerConditionWithPorts):
     """Precondition: all CVD participants in the case have RM.CLOSED.
 
     Iterates ``case.actor_participant_index`` and returns ``FAILURE`` if any
@@ -190,7 +190,7 @@ def update(self) -> Status:
         return Status.SUCCESS
 
 
-class CloseNotYetEmittedConditionNode(DataLayerCondition):
+class CloseNotYetEmittedConditionNode(DataLayerConditionWithPorts):
     """Idempotency guard: no ``Leave(VulnerabilityCase)`` in the outbox yet.
 
     Queries the actor's outbox for existing activities and checks whether any
diff --git a/vultron/core/behaviors/status/nodes/lifecycle.py b/vultron/core/behaviors/status/nodes/lifecycle.py
index 45db80f10..384730e9a 100644
--- a/vultron/core/behaviors/status/nodes/lifecycle.py
+++ b/vultron/core/behaviors/status/nodes/lifecycle.py
@@ -36,7 +36,11 @@
     reject_proposed_embargo_bt,
     terminate_embargo_bt,
 )
-from vultron.core.behaviors.helpers import DataLayerAction, DataLayerCondition
+from vultron.core.behaviors.helpers import (
+    DataLayerAction,
+    DataLayerActionWithPorts,
+    DataLayerConditionWithPorts,
+)
 from vultron.core.ports.case_persistence import CaseOutboxPersistence
 from vultron.core.models.case import VulnerabilityCase
 from vultron.core.models.case_participant import CaseParticipant
@@ -51,7 +55,7 @@
 logger = logging.getLogger(__name__)
 
 
-class _PublicDisclosureSkipConditionNode(DataLayerCondition):
+class _PublicDisclosureSkipConditionNode(DataLayerConditionWithPorts):
     """Inner guard for :class:`PublicDisclosureBranchNode`.
 
     Returns SUCCESS (skip teardown) when:
@@ -223,7 +227,7 @@ def __init__(
         )
 
 
-class EmitAddCaseStatusToSelfNode(DataLayerAction):
+class EmitAddCaseStatusToSelfNode(DataLayerActionWithPorts):
     """Emit a self-addressed ``Add(CaseStatus, VulnerabilityCase)`` to the CaseActor.
 
     When ``StatusUpdateGuard`` passes (RSH-01-003), this node:
diff --git a/vultron/core/behaviors/status/nodes/threat_termination.py b/vultron/core/behaviors/status/nodes/threat_termination.py
index 67c1210dc..9fb0c3a5b 100644
--- a/vultron/core/behaviors/status/nodes/threat_termination.py
+++ b/vultron/core/behaviors/status/nodes/threat_termination.py
@@ -27,7 +27,7 @@
 from py_trees.common import Status
 
 from vultron.core.behaviors.embargo.trigger_tree import terminate_embargo_bt
-from vultron.core.behaviors.helpers import DataLayerCondition
+from vultron.core.behaviors.helpers import DataLayerConditionWithPorts
 from vultron.core.models.case import VulnerabilityCase
 from vultron.core.models.protocols import PersistableModel
 from vultron.core.models._helpers import _as_id
@@ -35,7 +35,7 @@
 logger = logging.getLogger(__name__)
 
 
-class _ThreatTerminationSkipConditionNode(DataLayerCondition):
+class _ThreatTerminationSkipConditionNode(DataLayerConditionWithPorts):
     """Inner guard for :class:`ThreatTerminationBranchNode`.
 
     Returns SUCCESS (skip teardown) when:

From b6db3b75ebdfbad45268775ca48cfda4f928a452 Mon Sep 17 00:00:00 2001
From: "Allen D. Householder" <adh@cert.org>
Date: Sat, 8 Aug 2026 01:45:54 +0000
Subject: [PATCH 2/3] =?UTF-8?q?history:=20archive=20implementation=20ISSUE?=
 =?UTF-8?q?-1883=20=E2=80=94=20feat:=20migrate=20core/behaviors/=20Ports?=
 =?UTF-8?q?=20(1/5)=20=E2=80=94=20trivial=20base-only=20reparent?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
---
 plan/history/2608/implementation/ISSUE-1883.md | 12 ++++++++++++
 1 file changed, 12 insertions(+)
 create mode 100644 plan/history/2608/implementation/ISSUE-1883.md

diff --git a/plan/history/2608/implementation/ISSUE-1883.md b/plan/history/2608/implementation/ISSUE-1883.md
new file mode 100644
index 000000000..51586d3c4
--- /dev/null
+++ b/plan/history/2608/implementation/ISSUE-1883.md
@@ -0,0 +1,12 @@
+---
+source: ISSUE-1883
+timestamp: '2026-08-08T01:45:16.535612+00:00'
+title: 'feat: migrate core/behaviors/ Ports (1/5) — trivial base-only reparent'
+type: implementation
+---
+
+## Issue #1883 — Migrate core/behaviors/ Ports (1/5): trivial base-only reparent
+
+Migrated all Type A BT nodes from DataLayerCondition/DataLayerAction to DataLayerConditionWithPorts/DataLayerActionWithPorts across note/, status/, and case/ domains. 30 nodes across 14 source files migrated. Trivial no-op setup() overrides removed from ownership_transfer.py (AC-2). 21 new typed-ports tests added across 3 new test files (AC-4). All 1677 unit tests pass.
+
+PR: <https://github.com/CERTCC/Vultron/pull/2125>

From 31a289a6cf924d2487a936e0d412647596a5ba42 Mon Sep 17 00:00:00 2001
From: "Allen D. Householder" <adh@cert.org>
Date: Sat, 8 Aug 2026 14:28:11 +0000
Subject: [PATCH 3/3] fix(pr-execute): address 2 findings from triage
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Both findings from the PR #2125 triage pass (IMPROVE, fix-now):

1. phase5-missed-type-a-node-0 — CheckCaseUpdateOwnerNode in
   vultron/core/behaviors/case/nodes/update.py was a genuine Type-A node
   (no setup() override, no register_key() calls; only reads the inherited
   datalayer/actor_id) that AC-1 missed. Reparented it from
   DataLayerCondition to DataLayerConditionWithPorts and added four tests
   to test/core/behaviors/case/nodes/test_typed_ports.py: the
   NoDataAvailable isolated-port check (BTND-03-011) plus BTTestScenario
   coverage for case-not-found, actor-owns-case, and actor-is-not-owner.

2. phase9-stale-notes-0 — notes/py-trees-ports-adoption.md still claimed
   a repo-wide search finds "zero" references to BehaviourWithPorts in
   vultron/, which became false when the pilot (#1808) landed. Rewrote the
   "Current state" section with verified counts (44 nodes migrated, 168 on
   the legacy bases, 275 register_key() sites remaining) broken out by
   domain, and recorded the concrete five-part #1809 split
   (#1883 through #1887) in the issue sequence.

Verification: test/core, test/bt, test/metadata, test/adapters, test/wire,
test/architecture, test/ci and the root-level test modules all pass. black,
flake8, mypy and pyright clean on the changed files; markdownlint and the
notes-frontmatter validator clean on the changed note.

Refs #1883
Refs #1809

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 notes/py-trees-ports-adoption.md              | 47 ++++++++++++-----
 .../behaviors/case/nodes/test_typed_ports.py  | 52 +++++++++++++++++++
 vultron/core/behaviors/case/nodes/update.py   |  3 +-
 3 files changed, 88 insertions(+), 14 deletions(-)

diff --git a/notes/py-trees-ports-adoption.md b/notes/py-trees-ports-adoption.md
index d609109d0..aa4de9095 100644
--- a/notes/py-trees-ports-adoption.md
+++ b/notes/py-trees-ports-adoption.md
@@ -29,22 +29,32 @@ adopting, what is deferred, the known technical mismatch, and the issue
 sequence — so each implementing agent starts from evidence rather than the
 Idea's optimistic framing.
 
-## Current state (verified 2026-07-29)
+## Current state (migration in progress — verified 2026-08-08)
 
 - **Dependency**: `pyproject.toml` already pins `py-trees>=2.5.0`. The Idea's
   "evaluate the upgrade path from the current pin" step is already satisfied —
   no upgrade is required to reach the Ports API.
-- **Ports API present, unused**: `py_trees.ports` exposes `BehaviourWithPorts`
-  (a `PortsMixin` + `Behaviour` subclass), `PortInformation`,
-  `NoDataAvailable`, and a ports registry. A repo-wide search finds **zero**
-  references to `input_ports`, `output_ports`, `BehaviourWithPorts`, or
-  `PortInformation` in `vultron/`, `specs/`, `notes/`, `docs/`, or `test/`.
-- **Node population**: `vultron/core/behaviors/` contains roughly **60 node
-  classes** and about **249 `register_key()` call sites**. Every node currently
-  subclasses `py_trees.behaviour.Behaviour` (or the DataLayer-aware base classes
-  in `helpers.py`) and declares blackboard access imperatively in `setup()` via
-  `register_key()`, following the `{noun}_{id_segment}` naming convention
-  (BTND-03-005, BTND-03-008).
+- **Ports API present and in use**: `py_trees.ports` exposes
+  `BehaviourWithPorts` (a `PortsMixin` + `Behaviour` subclass),
+  `PortInformation`, `NoDataAvailable`, and a ports registry. The pilot (#1808)
+  landed `DataLayerConditionWithPorts` and `DataLayerActionWithPorts` in
+  `vultron/core/behaviors/helpers.py` and migrated a first tranche of
+  `report/nodes/`. The pattern is now the standard base for all new nodes
+  (ADR-0044, BTND-03-009 through BTND-03-011).
+- **Migration progress**: of the node classes deriving from the DataLayer-aware
+  base classes in `helpers.py`, **44 have been migrated** to the `*WithPorts`
+  bases and **168 remain on the legacy `DataLayerCondition` /
+  `DataLayerAction`** bases. Migrated counts by domain: `case` 29, `status` 7,
+  `note` 4, `report` 4. Remaining legacy counts by domain: `case` 67, `sync` 32,
+  `report` 27, `embargo` 26, `status` 7, `sender` 3, `inbox` 1, plus the five
+  generic helper nodes in `helpers.py` itself. Roughly **275 `register_key()`
+  call sites** remain in the unmigrated nodes, still following the
+  `{noun}_{id_segment}` naming convention (BTND-03-005, BTND-03-008).
+- **Remaining work** is tracked under the #1809 full-migration chain, split into
+  five parts. Part 1 (#1883) covered the trivial Type-A base-only reparents in
+  `case/`, `status/`, and `note/`. Later parts cover the Type-B nodes that carry
+  domain-specific `register_key()` calls and therefore need explicit
+  `input_ports()` / `output_ports()` declarations.
 - **XML parser**: `py_trees.parsers.behaviour_tree_xml` exists but is documented
   as **experimental** ("the parser is experimental and its API may change
   between releases"). It instantiates only classes registered in a
@@ -227,7 +237,18 @@ Derived from the #1558 grill-me interview. All Tasks are children of Epic #427.
    migration recipe. `size:M`.
 2. **#1809 — Full node migration** *(Task, blocked-by #1808)*
    Migrate the remaining `vultron/core/behaviors/` nodes to typed Ports,
-   following #1808's recipe. `size:L`.
+   following #1808's recipe. `size:L`. Split into five sequential sub-Tasks so
+   each part has a reviewable blast radius:
+   - **#1883 (1/5)** — trivial base-only reparent: `case`, `status`, `note`,
+     misc. Type-A nodes only (no domain `register_key()`), so the change is a
+     pure base-class swap plus isolated-port tests.
+   - **#1884 (2/5)** — trivial base-only reparent: `report`, `embargo`.
+   - **#1885 (3/5)** — read-only extra-input nodes: add `input_ports()` and
+     replace direct blackboard reads with `get_input()`.
+   - **#1886 (4/5)** — WRITE-handoff nodes: establish `output_ports()` and the
+     execution-scoped key convention (AC-3's BTND-03-004 property).
+   - **#1887 (5/5)** — non-DataLayer nodes, composite exemptions, and
+     finalization (including this note's AC-4 completion update).
 3. **#1810 — XML feasibility spike** *(Task, blocked-by #1809)*
    Assess whether protocol BTs can be authored/exported as BehaviorTree XML
    given the constructor-vs-remapping mismatch and the experimental parser.
diff --git a/test/core/behaviors/case/nodes/test_typed_ports.py b/test/core/behaviors/case/nodes/test_typed_ports.py
index aaa6a40b4..3299bb027 100644
--- a/test/core/behaviors/case/nodes/test_typed_ports.py
+++ b/test/core/behaviors/case/nodes/test_typed_ports.py
@@ -28,6 +28,9 @@
 from vultron.core.behaviors.case.nodes.suggest_actor.conditions import (
     ActorAlreadyParticipantNode,
 )
+from vultron.core.behaviors.case.nodes.update import (
+    CheckCaseUpdateOwnerNode,
+)
 from vultron.core.behaviors.case.nodes.vfd_role_guards import (
     CheckVendorRoleNode,
 )
@@ -133,3 +136,52 @@ def test_failure_when_actor_not_participant(
             actor_id=ACTOR_ID,
         )
         bt_scenario.assert_failure(result)
+
+
+# ---------------------------------------------------------------------------
+# update.py — CheckCaseUpdateOwnerNode
+# ---------------------------------------------------------------------------
+
+
+class TestCheckCaseUpdateOwnerNodePorts:
+    def test_missing_datalayer_raises_no_data_available(self) -> None:
+        node = CheckCaseUpdateOwnerNode(case_id=CASE_ID)
+        node.setup_ports()
+        with pytest.raises(NoDataAvailable):
+            node.get_input("datalayer")
+
+    def test_failure_when_case_not_found(
+        self, bt_scenario: BTTestScenario
+    ) -> None:
+        result = bt_scenario.run(
+            CheckCaseUpdateOwnerNode(case_id=CASE_ID), actor_id=ACTOR_ID
+        )
+        bt_scenario.assert_failure(result)
+
+    def test_success_when_actor_owns_case(
+        self, bt_scenario: BTTestScenario
+    ) -> None:
+        case = VulnerabilityCase(
+            id_=CASE_ID,
+            name="Test Case",
+            attributed_to=ACTOR_ID,
+        )
+        bt_scenario.seed(case)
+        result = bt_scenario.run(
+            CheckCaseUpdateOwnerNode(case_id=CASE_ID), actor_id=ACTOR_ID
+        )
+        bt_scenario.assert_success(result)
+
+    def test_failure_when_actor_is_not_owner(
+        self, bt_scenario: BTTestScenario
+    ) -> None:
+        case = VulnerabilityCase(
+            id_=CASE_ID,
+            name="Test Case",
+            attributed_to="https://example.org/actors/other",
+        )
+        bt_scenario.seed(case)
+        result = bt_scenario.run(
+            CheckCaseUpdateOwnerNode(case_id=CASE_ID), actor_id=ACTOR_ID
+        )
+        bt_scenario.assert_failure(result)
diff --git a/vultron/core/behaviors/case/nodes/update.py b/vultron/core/behaviors/case/nodes/update.py
index f8d94f013..a7c2294a9 100644
--- a/vultron/core/behaviors/case/nodes/update.py
+++ b/vultron/core/behaviors/case/nodes/update.py
@@ -31,13 +31,14 @@
     DataLayerAction,
     DataLayerActionWithPorts,
     DataLayerCondition,
+    DataLayerConditionWithPorts,
 )
 from vultron.core.models.events.case import UpdateCaseReceivedEvent
 from vultron.core.models._helpers import _as_id
 from vultron.core.models.case import VulnerabilityCase
 
 
-class CheckCaseUpdateOwnerNode(DataLayerCondition):
+class CheckCaseUpdateOwnerNode(DataLayerConditionWithPorts):
     """Return SUCCESS when the current actor owns the case."""
 
     def __init__(self, case_id: str, name: str | None = None) -> None: