Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 34 additions & 13 deletions notes/py-trees-ports-adoption.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand Down
12 changes: 12 additions & 0 deletions plan/history/2608/implementation/ISSUE-1883.md
Original file line number Diff line number Diff line change
@@ -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>
187 changes: 187 additions & 0 deletions test/core/behaviors/case/nodes/test_typed_ports.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,187 @@
#!/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.update import (
CheckCaseUpdateOwnerNode,
)
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)


# ---------------------------------------------------------------------------
# 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)
Loading
Loading