diff --git a/docs/adr/0044-py-trees-typed-ports-adoption.md b/docs/adr/0044-py-trees-typed-ports-adoption.md new file mode 100644 index 000000000..7406562bf --- /dev/null +++ b/docs/adr/0044-py-trees-typed-ports-adoption.md @@ -0,0 +1,120 @@ +--- +status: accepted +date: 2026-07-29 +deciders: + - vultron maintainers +consulted: + - project stakeholders +informed: + - contributors +--- + +# Adopt py_trees Typed Ports for BT Node Blackboard Contracts + +## Context and Problem Statement + +ADR-0008 established `py_trees` as the BT library for prototype handler +execution in `vultron/core/behaviors/`. The current convention for declaring +a node's blackboard dependencies is imperative: each node calls +`register_key()` in `setup()` and reads keys via direct attribute access on +`self.blackboard` in `initialise()`. This approach works, but has two +weaknesses: + +1. **Hidden data contracts**: the keys a node reads or writes are scattered + across `setup()`, `initialise()`, and `update()` — there is no single, + machine-readable declaration. +2. **Late error detection**: a misspelled key or a missing required input + surfaces as a mid-tick `KeyError` or a silent `None`, not at setup time. + +`py_trees >= 2.5.0` (already pinned in `pyproject.toml`) provides a typed +Ports API (`BehaviourWithPorts`, `PortInformation`, `setup_ports()`, +`get_input()`) that addresses both weaknesses. + +## Decision Drivers + +- Explicit, readable data-flow contracts at the class level (one method per + direction: `input_ports()` / `output_ports()`). +- Structured early error detection: `NoDataAvailable` on a missing required + input surfaces in `initialise()` (at the start of the first tick) rather than + silently mid-tick as a `KeyError`. +- Easier isolated unit tests: a node can be exercised by wiring its ports to + known blackboard keys without scaffolding a full tree. +- No dependency upgrade: `py_trees >= 2.5.0` is already installed. +- Compatibility with the existing `BTBridge` flat-key convention: port + remappings (`{"datalayer": "/datalayer", "actor_id": "/actor_id"}`) bridge + the Ports namespace to the keys `BTBridge.setup_tree()` writes. + +## Considered Options + +1. Keep the imperative `register_key` / direct-attribute convention. +2. Adopt `BehaviourWithPorts` as base for all `vultron/core/behaviors/` nodes. +3. Adopt `BehaviourWithPorts` for new nodes only; keep legacy nodes unchanged. + +## Decision Outcome + +**Option 2 — adopt `BehaviourWithPorts` as the standard base for all nodes +in `vultron/core/behaviors/`**, piloted on the `validate_report` subtree. + +The pilot validates the migration recipe before the full sweep (Issue #1809). + +### Consequences + +**Good:** + +- `input_ports()` / `output_ports()` are the single source of truth for + each node's blackboard contract — readable without tracing `update()`. +- `NoDataAvailable` on a missing required input surfaces when `get_input()` is + called in `initialise()` — at the start of the first tick, not during `setup()`. +- Isolated node tests can call `setup_ports()` + `get_input()` directly + without running a full tree. +- `DataLayerConditionWithPorts` and `DataLayerActionWithPorts` in + `helpers.py` encapsulate the standard port remappings and `_require_*` + guard helpers, so subclasses need only declare domain-specific ports. + +**Known consequence (not a blocker):** + +- The py_trees XML parser instantiates nodes via port remapping only and + has no mechanism for passing constructor arguments (factories, case IDs). + Vultron nodes are **heavily constructor-parameterized** (BTND-01-001), so + XML-based tree authoring is not yet feasible. This tension is documented + in `notes/py-trees-ports-adoption.md` and is the subject of the XML + feasibility spike (Issue #1810). + +### Implementation pattern + +```python +class CheckRMStateValid(DataLayerConditionWithPorts): + @classmethod + def input_ports(cls) -> dict[str, PortInformation]: + # Inherit datalayer + actor_id from base class defaults. + # Override only to add domain-specific ports. + return super().input_ports() + + @classmethod + def output_ports(cls) -> dict[str, PortInformation]: + return {} + + def update(self) -> Status: + if (f := self._require_datalayer()) is not None: + return f + # ... domain logic using self.datalayer and self.actor_id +``` + +`DataLayerConditionWithPorts.setup()` calls `setup_ports()` with the +standard BTBridge remappings; `initialise()` calls `get_input()` to +populate `self.datalayer` and `self.actor_id`. + +## More Information + +- `notes/py-trees-ports-adoption.md` — full migration recipe and issue + sequence. +- `specs/behavior-tree-node-design.yaml` BTND-03-009 through BTND-03-011 — + normative requirements for typed port declarations. +- `vultron/core/behaviors/helpers.py` — `DataLayerConditionWithPorts` and + `DataLayerActionWithPorts` base classes. +- Pilot subtree: `vultron/core/behaviors/report/nodes/` (`CheckRMStateValid`, + `CheckRMStateReceivedOrInvalid`, `EnsureEmbargoExists`, + `TransitionRMtoValid`). +- Related ADRs: ADR-0008 (py_trees adoption), ADR-0025 (call-out point + abstraction). +- Follow-on: Issue #1809 (full migration), Issue #1810 (XML spike). diff --git a/docs/adr/index.md b/docs/adr/index.md index f5668a986..9fea6520c 100644 --- a/docs/adr/index.md +++ b/docs/adr/index.md @@ -89,6 +89,7 @@ General information about architectural decision records is available at Planning analysis for adopting py_trees 2.5.0 typed Ports in vultron/core/behaviors/: the concrete wins (typed data contracts, early error detection, isolated node @@ -118,6 +118,104 @@ pilot subtree, not pre-committed here: - **`specs/behavior-tree-integration.yaml`**: update where it describes the blackboard/bridge contract if Ports change the bridge's setup path. +## Migration recipe (established by #1808) + +Follow these steps when migrating any `DataLayerCondition` / `DataLayerAction` +node to typed Ports (required for all new nodes; follow-on sweep in #1809). + +### 1. Change the base class + +```python +# Before +class MyNode(DataLayerCondition): + ... + +# After +class MyNode(DataLayerConditionWithPorts): + ... +``` + +Use `DataLayerConditionWithPorts` for condition (read-only) nodes and +`DataLayerActionWithPorts` for action (mutating) nodes. + +### 2. Add `input_ports()` and `output_ports()` + +Declare every blackboard key the node uses. The base classes already declare +`datalayer`, `actor_id`, and (for actions) `trigger_activity_factory`. +Override `input_ports()` to **add** domain-specific ports; call +`super().input_ports()` first if you want to extend rather than replace: + +```python +@classmethod +def input_ports(cls) -> dict[str, PortInformation]: + ports = super().input_ports() # inherit base ports + ports["report_id"] = PortInformation(data_type=str, required=True) + return ports + +@classmethod +def output_ports(cls) -> dict[str, PortInformation]: + return {} # or declare output ports here +``` + +If the node has **no** domain-specific ports beyond the base class defaults, +omit `input_ports()` and `output_ports()` entirely — the base class +definitions are abstract but already implemented. + +### 3. Remove `setup()` overrides that only call `register_key()` + +The base class `setup()` calls `setup_ports()` with the standard BTBridge +remappings. Only override `setup()` if you need to register **additional** +domain-specific ports: + +```python +def setup(self, **kwargs: Any) -> None: + super().setup(**kwargs) # registers datalayer, actor_id + # register additional domain ports here if needed +``` + +### 4. Update `initialise()` to use `get_input()` + +```python +# Before +def initialise(self) -> None: + self.datalayer = self.blackboard.datalayer + self.actor_id = self.blackboard.actor_id + +# After — base class initialise() already does this; only override if +# you need to read additional domain ports: +def initialise(self) -> None: + super().initialise() # sets self.datalayer, self.actor_id + self.report_id = self.get_input("report_id") +``` + +### 5. Leave `update()` unchanged + +The `update()` logic itself does not change — it still uses `self.datalayer`, +`self.actor_id`, and the `_require_*` guard helpers. + +### 6. Add isolated-node tests + +Add at least two tests per migrated node: + +- **Happy path**: wire required ports, tick once, assert SUCCESS. +- **Missing required port**: call `setup_ports()` without the required port + written to the blackboard; assert `NoDataAvailable` is raised at setup time. + +Use the `BTTestScenario` harness from `test/core/behaviors/bt_harness.py` for +the happy-path test. For the isolated-port test, call `setup_ports()` and +`get_input()` directly: + +```python +from py_trees.ports import NoDataAvailable +import pytest + +def test_missing_required_port(): + node = CheckRMStateValid(report_id="https://example.org/reports/1") + node.setup_ports() # no remappings → default keys; blackboard is empty + with pytest.raises(NoDataAvailable): + node.get_input("datalayer") +``` + ## Issue sequence Derived from the #1558 grill-me interview. All Tasks are children of Epic #427. diff --git a/plan/history/2607/implementation/ISSUE-1808.md b/plan/history/2607/implementation/ISSUE-1808.md new file mode 100644 index 000000000..9ada9217f --- /dev/null +++ b/plan/history/2607/implementation/ISSUE-1808.md @@ -0,0 +1,27 @@ +--- +source: ISSUE-1808 +timestamp: '2026-07-29T20:40:49.748458+00:00' +title: Typed Ports base classes, ADR-0044, BTND-03-009–011, validate_report pilot +type: implementation +--- + +## Issue #1808 — Adopt py_trees typed Ports: base class, convention, ADR/spec, pilot one subtree + +Implementation complete. PR: + +### Deliverables + +- `DataLayerConditionWithPorts` and `DataLayerActionWithPorts` in `vultron/core/behaviors/helpers.py` +- ADR-0044 (`docs/adr/0044-py-trees-typed-ports-adoption.md`) + ADR index + mkdocs nav +- BTND-03-009 through BTND-03-011 in `specs/behavior-tree-node-design.yaml` +- 4 pilot nodes migrated: `CheckRMStateValid`, `CheckRMStateReceivedOrInvalid`, `EnsureEmbargoExists`, `TransitionRMtoValid` +- 25-test `test/core/behaviors/report/nodes/test_typed_ports.py` +- 6-step migration recipe in `notes/py-trees-ports-adoption.md` + +### Key fixes during implementation + +- YAML spec entries with inline `{key: value}` maps required `>` folded scalars +- py_trees ports registry collision in tests: inline subclasses across two tests with the same name trigger UserWarning → narrowed with unique class names +- mkdocs nav required explicit ADR entry (enforced by `adr-index-sync` pre-commit hook) +- `except Exception` narrowed to `except NoDataAvailable` for optional trigger_activity_factory port +- `NoDataAvailable` timing corrected: surfaces in `initialise()` (first tick), not `setup()` diff --git a/plan/incoming/learnings/20260729-no-data-available-timing.md b/plan/incoming/learnings/20260729-no-data-available-timing.md new file mode 100644 index 000000000..e5cbe68a2 --- /dev/null +++ b/plan/incoming/learnings/20260729-no-data-available-timing.md @@ -0,0 +1,26 @@ +--- +title: NoDataAvailable surfaces in initialise(), not setup() +type: learning +timestamp: 2026-07-29 +source: ISSUE-1808 +signal: spec-ambiguity +--- + +The py_trees `BehaviourWithPorts` API raises `NoDataAvailable` when `get_input()` +is called on a port whose key has no value on the blackboard — which happens in +`initialise()` at the start of the first tick, not during `setup_ports()` or +`setup()`. + +The initial spec entries (BTND-03-011) and ADR-0044 both stated that the error +surfaces "at setup time" / "at `setup_ports()` time". This is incorrect: +`setup_ports()` only registers key access; the read and the potential raise happen +in `initialise()`. + +The spec and ADR were corrected before the PR was opened, but this is a recurring +source of confusion: "early error detection" means "at the start of the first tick" +(earlier than mid-tick `KeyError` from direct attribute access), not "before any tick." + +**Implication for tests**: A missing-required-port test should call `setup_ports()` +(with no remappings → default namespace keys; blackboard empty), then call +`get_input("port_name")` and assert `NoDataAvailable`. Calling only `setup_ports()` +and asserting an error there will produce a false-passing test. diff --git a/specs/behavior-tree-node-design.yaml b/specs/behavior-tree-node-design.yaml index e0bb93fc1..a45046e49 100644 --- a/specs/behavior-tree-node-design.yaml +++ b/specs/behavior-tree-node-design.yaml @@ -156,6 +156,50 @@ groups: relationships: - rel_type: refines spec_id: BTND-03-001 + - id: BTND-03-009 + priority: MUST + kind: project + statement: New BT nodes in `vultron/core/behaviors/` MUST subclass `DataLayerConditionWithPorts` or + `DataLayerActionWithPorts` (from `helpers.py`) rather than `DataLayerCondition` / `DataLayerAction`. The typed + `input_ports()` / `output_ports()` class methods constitute the node's formal blackboard contract and MUST declare + every port the node reads or writes. + rationale: Typed port declarations are machine-readable, appear at the class level, and are the single source of truth + for a node's data-flow contract. The imperative `register_key()` approach scatters declarations across `setup()`, + `initialise()`, and `update()`. See ADR-0044. + relationships: + - rel_type: extends + spec_id: BTND-03-001 + - rel_type: refines + spec_id: BTND-03-003 + - id: BTND-03-010 + priority: MUST + kind: project + statement: > + A `DataLayerConditionWithPorts` or `DataLayerActionWithPorts` subclass MUST call `setup_ports()` in its + `setup()` method with port remappings `{"datalayer": "/datalayer", "actor_id": "/actor_id"}` (plus + `"trigger_activity_factory": "/trigger_activity_factory"` for action nodes). These absolute-path remappings wire + the node's declared ports to the flat keys written by `BTBridge.setup_tree()`. + rationale: > + The `BehaviourWithPorts` default namespace scopes port keys under the node's subtree path, which does not + match the flat `/datalayer` and `/actor_id` keys that `BTBridge.setup_tree()` writes. Explicit absolute-path + remappings are required for compatibility. The base classes `DataLayerConditionWithPorts` and + `DataLayerActionWithPorts` implement this in their own `setup()` so subclasses that do not add extra ports need + not override `setup()`. + relationships: + - rel_type: refines + spec_id: BTND-03-009 + - id: BTND-03-011 + priority: MUST + kind: project + statement: A `DataLayerConditionWithPorts` or `DataLayerActionWithPorts` subclass MUST read injected values via + `get_input(port_name)` in `initialise()` (not via direct `self.blackboard.` attribute access). A missing + required port raises `NoDataAvailable` when `get_input()` is first called in `initialise()` — at the start of + the first tick, not during `setup()`. + rationale: Direct blackboard attribute access bypasses the typed port layer and its early error detection. Using + `get_input()` preserves type checking and `NoDataAvailable` detection for required ports. + relationships: + - rel_type: refines + spec_id: BTND-03-009 - id: BTND-04 title: Module Ownership specs: diff --git a/test/core/behaviors/report/nodes/test_typed_ports.py b/test/core/behaviors/report/nodes/test_typed_ports.py new file mode 100644 index 000000000..a4f53c1b8 --- /dev/null +++ b/test/core/behaviors/report/nodes/test_typed_ports.py @@ -0,0 +1,389 @@ +#!/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 + +"""Tests for typed-Ports base classes and pilot node migrations. + +Covers: +- BTND-03-009: typed port declarations replace register_key(). +- BTND-03-010: setup_ports() with remappings wires BTBridge flat keys. +- BTND-03-011: get_input() reads injected values correctly. +- AC-4 (issue #1808): isolated-node port tests + early-error-detection tests. +""" + +import pytest +import py_trees +from py_trees.ports import NoDataAvailable + +from vultron.adapters.driven.datalayer_sqlite import SqliteDataLayer +from vultron.core.behaviors.helpers import ( + DataLayerConditionWithPorts, + DataLayerActionWithPorts, +) +from vultron.core.behaviors.report.nodes.conditions import ( + CheckRMStateValid, + CheckRMStateReceivedOrInvalid, + EnsureEmbargoExists, +) +from vultron.core.behaviors.report.nodes.rm_transitions import ( + TransitionRMtoValid, +) +from vultron.core.models.case import VulnerabilityCase +from vultron.core.models.case_actor import VultronCaseActor +from vultron.core.models.dimensions import RmDimension +from vultron.core.models.participant_status import ParticipantStatus +from vultron.core.models.report import VultronReport +from vultron.core.states.rm import RM +from vultron.core.models._helpers import _report_phase_status_id +from test.core.behaviors.bt_harness import BTTestScenario + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +ACTOR_ID = "https://example.org/actors/vendor" +REPORT_ID = "https://example.org/reports/CVE-2024-001" +OFFER_ID = "https://example.org/activities/offer-123" + + +def _fresh_dl() -> SqliteDataLayer: + return SqliteDataLayer("sqlite:///:memory:") + + +def _write_btbridge_keys( + dl: SqliteDataLayer, actor_id: str = ACTOR_ID +) -> None: + """Simulate what BTBridge.setup_tree() writes to the flat blackboard.""" + bb = py_trees.blackboard.Client(name="test_setup") + bb.register_key(key="datalayer", access=py_trees.common.Access.WRITE) + bb.register_key(key="actor_id", access=py_trees.common.Access.WRITE) + bb.datalayer = dl + bb.actor_id = actor_id + + +# --------------------------------------------------------------------------- +# Base-class contract: DataLayerConditionWithPorts +# --------------------------------------------------------------------------- + + +class TestDataLayerConditionWithPortsContract: + """Verify the base class declares required ports and reads via get_input.""" + + def test_input_ports_declares_datalayer(self) -> None: + ports = DataLayerConditionWithPorts.input_ports() + assert "datalayer" in ports + assert ports["datalayer"].required + + def test_input_ports_declares_actor_id(self) -> None: + ports = DataLayerConditionWithPorts.input_ports() + assert "actor_id" in ports + assert ports["actor_id"].required + + def test_output_ports_empty(self) -> None: + assert DataLayerConditionWithPorts.output_ports() == {} + + def test_missing_required_port_raises_at_get_input(self) -> None: + """NoDataAvailable raised when required port not on blackboard (BTND-03-011).""" + py_trees.blackboard.Blackboard.storage.clear() + + class _MinimalNode(DataLayerConditionWithPorts): + def update(self): + return py_trees.common.Status.SUCCESS + + node = _MinimalNode(name="MinimalTest") + node.setup_ports() + with pytest.raises(NoDataAvailable): + node.get_input("datalayer") + + def test_get_input_reads_btbridge_flat_key(self) -> None: + """get_input('datalayer') reads the /datalayer key written by BTBridge.""" + py_trees.blackboard.Blackboard.storage.clear() + dl = _fresh_dl() + _write_btbridge_keys(dl) + + class _MinimalNodeWithRemapping(DataLayerConditionWithPorts): + def update(self): + return py_trees.common.Status.SUCCESS + + node = _MinimalNodeWithRemapping(name="MinimalTest") + node.setup_ports( + port_remappings={ + "datalayer": "/datalayer", + "actor_id": "/actor_id", + } + ) + assert node.get_input("datalayer") is dl + assert node.get_input("actor_id") == ACTOR_ID + + +# --------------------------------------------------------------------------- +# Base-class contract: DataLayerActionWithPorts +# --------------------------------------------------------------------------- + + +class TestDataLayerActionWithPortsContract: + def test_input_ports_declares_trigger_factory(self) -> None: + ports = DataLayerActionWithPorts.input_ports() + assert "trigger_activity_factory" in ports + assert not ports["trigger_activity_factory"].required + + def test_output_ports_empty(self) -> None: + assert DataLayerActionWithPorts.output_ports() == {} + + +# --------------------------------------------------------------------------- +# Pilot: CheckRMStateValid +# --------------------------------------------------------------------------- + + +class TestCheckRMStateValidPorts: + def test_input_ports_declared(self) -> None: + ports = CheckRMStateValid.input_ports() + assert "datalayer" in ports + assert "actor_id" in ports + + def test_output_ports_empty(self) -> None: + assert CheckRMStateValid.output_ports() == {} + + def test_missing_datalayer_raises_no_data_available(self) -> None: + py_trees.blackboard.Blackboard.storage.clear() + node = CheckRMStateValid(report_id=REPORT_ID) + node.setup_ports() + with pytest.raises(NoDataAvailable): + node.get_input("datalayer") + + def test_via_bt_scenario_when_valid( + self, bt_scenario: BTTestScenario + ) -> None: + actor = VultronCaseActor(id_=ACTOR_ID, name="Vendor") + report = VultronReport(id_=REPORT_ID, name="R1", content="c") + status = ParticipantStatus( + id_=_report_phase_status_id(ACTOR_ID, REPORT_ID, RM.VALID.value), + context=REPORT_ID, + attributed_to=ACTOR_ID, + rm=RmDimension(state=RM.VALID), + ) + bt_scenario.seed(actor, report, status) + result = bt_scenario.run( + CheckRMStateValid(report_id=REPORT_ID), actor_id=ACTOR_ID + ) + bt_scenario.assert_success(result) + + def test_via_bt_scenario_when_not_valid( + self, bt_scenario: BTTestScenario + ) -> None: + actor = VultronCaseActor(id_=ACTOR_ID, name="Vendor") + report = VultronReport(id_=REPORT_ID, name="R1", content="c") + bt_scenario.seed(actor, report) + result = bt_scenario.run( + CheckRMStateValid(report_id=REPORT_ID), actor_id=ACTOR_ID + ) + bt_scenario.assert_failure(result) + + def test_sender_actor_id_checked_instead_of_blackboard_actor( + self, bt_scenario: BTTestScenario + ) -> None: + """sender_actor_id overrides blackboard actor_id (ADR-0022 single-BT).""" + SENDER_ID = "https://example.org/actors/reporter" + actor = VultronCaseActor(id_=ACTOR_ID, name="Vendor") + sender = VultronCaseActor(id_=SENDER_ID, name="Reporter") + report = VultronReport(id_=REPORT_ID, name="R1", content="c") + # Only the sender has a VALID status record, not the blackboard actor. + status = ParticipantStatus( + id_=_report_phase_status_id(SENDER_ID, REPORT_ID, RM.VALID.value), + context=REPORT_ID, + attributed_to=SENDER_ID, + rm=RmDimension(state=RM.VALID), + ) + bt_scenario.seed(actor, sender, report, status) + # Tree runs under ACTOR_ID (blackboard actor_id = ACTOR_ID); + # node must check SENDER_ID's RM state and return SUCCESS. + result = bt_scenario.run( + CheckRMStateValid(report_id=REPORT_ID, sender_actor_id=SENDER_ID), + actor_id=ACTOR_ID, + ) + bt_scenario.assert_success(result) + + +# --------------------------------------------------------------------------- +# Pilot: CheckRMStateReceivedOrInvalid +# --------------------------------------------------------------------------- + + +class TestCheckRMStateReceivedOrInvalidPorts: + def test_input_ports_declared(self) -> None: + ports = CheckRMStateReceivedOrInvalid.input_ports() + assert "datalayer" in ports + assert "actor_id" in ports + + def test_missing_datalayer_raises_no_data_available(self) -> None: + py_trees.blackboard.Blackboard.storage.clear() + node = CheckRMStateReceivedOrInvalid(report_id=REPORT_ID) + node.setup_ports() + with pytest.raises(NoDataAvailable): + node.get_input("datalayer") + + def test_success_when_no_valid_status( + self, bt_scenario: BTTestScenario + ) -> None: + actor = VultronCaseActor(id_=ACTOR_ID, name="Vendor") + report = VultronReport(id_=REPORT_ID, name="R1", content="c") + bt_scenario.seed(actor, report) + result = bt_scenario.run( + CheckRMStateReceivedOrInvalid(report_id=REPORT_ID), + actor_id=ACTOR_ID, + ) + bt_scenario.assert_success(result) + + def test_failure_when_already_valid( + self, bt_scenario: BTTestScenario + ) -> None: + actor = VultronCaseActor(id_=ACTOR_ID, name="Vendor") + report = VultronReport(id_=REPORT_ID, name="R1", content="c") + status = ParticipantStatus( + id_=_report_phase_status_id(ACTOR_ID, REPORT_ID, RM.VALID.value), + context=REPORT_ID, + attributed_to=ACTOR_ID, + rm=RmDimension(state=RM.VALID), + ) + bt_scenario.seed(actor, report, status) + result = bt_scenario.run( + CheckRMStateReceivedOrInvalid(report_id=REPORT_ID), + actor_id=ACTOR_ID, + ) + bt_scenario.assert_failure(result) + + +# --------------------------------------------------------------------------- +# Pilot: EnsureEmbargoExists +# --------------------------------------------------------------------------- + + +class TestEnsureEmbargoExistsPorts: + def test_input_ports_declared(self) -> None: + ports = EnsureEmbargoExists.input_ports() + assert "datalayer" in ports + assert "actor_id" in ports + + def test_missing_datalayer_raises_no_data_available(self) -> None: + py_trees.blackboard.Blackboard.storage.clear() + node = EnsureEmbargoExists(report_id=REPORT_ID) + node.setup_ports() + with pytest.raises(NoDataAvailable): + node.get_input("datalayer") + + def test_failure_when_no_case(self, bt_scenario: BTTestScenario) -> None: + actor = VultronCaseActor(id_=ACTOR_ID, name="Vendor") + report = VultronReport(id_=REPORT_ID, name="R1", content="c") + bt_scenario.seed(actor, report) + result = bt_scenario.run( + EnsureEmbargoExists(report_id=REPORT_ID), actor_id=ACTOR_ID + ) + bt_scenario.assert_failure(result) + + def test_success_when_case_has_active_embargo( + self, bt_scenario: BTTestScenario + ) -> None: + actor = VultronCaseActor(id_=ACTOR_ID, name="Vendor") + report = VultronReport(id_=REPORT_ID, name="R1", content="c") + # active_embargo is a str | None URI field on VulnerabilityCase + case = VulnerabilityCase( + name="Test Case", + vulnerability_reports=[REPORT_ID], + attributed_to=ACTOR_ID, + ) + case.active_embargo = "https://example.org/embargos/em-001" + bt_scenario.seed(actor, report, case) + result = bt_scenario.run( + EnsureEmbargoExists(report_id=REPORT_ID), actor_id=ACTOR_ID + ) + bt_scenario.assert_success(result) + + def test_failure_when_case_has_no_active_embargo( + self, bt_scenario: BTTestScenario + ) -> None: + actor = VultronCaseActor(id_=ACTOR_ID, name="Vendor") + report = VultronReport(id_=REPORT_ID, name="R1", content="c") + case = VulnerabilityCase( + name="Test Case", + vulnerability_reports=[REPORT_ID], + attributed_to=ACTOR_ID, + ) + # active_embargo defaults to None — no explicit assignment needed + bt_scenario.seed(actor, report, case) + result = bt_scenario.run( + EnsureEmbargoExists(report_id=REPORT_ID), actor_id=ACTOR_ID + ) + bt_scenario.assert_failure(result) + + +# --------------------------------------------------------------------------- +# Pilot: TransitionRMtoValid +# --------------------------------------------------------------------------- + + +class TestTransitionRMtoValidPorts: + def test_input_ports_declared(self) -> None: + ports = TransitionRMtoValid.input_ports() + assert "datalayer" in ports + assert "actor_id" in ports + assert "trigger_activity_factory" in ports + + def test_output_ports_empty(self) -> None: + assert TransitionRMtoValid.output_ports() == {} + + def test_missing_datalayer_raises_no_data_available(self) -> None: + py_trees.blackboard.Blackboard.storage.clear() + node = TransitionRMtoValid(report_id=REPORT_ID, offer_id=OFFER_ID) + node.setup_ports() + with pytest.raises(NoDataAvailable): + node.get_input("datalayer") + + def test_creates_rm_valid_status_record( + self, bt_scenario: BTTestScenario + ) -> None: + from vultron.core.models.activity import VultronOffer + + actor = VultronCaseActor(id_=ACTOR_ID, name="Vendor") + report = VultronReport(id_=REPORT_ID, name="R1", content="c") + offer = VultronOffer( + id_=OFFER_ID, actor=ACTOR_ID, object_=REPORT_ID, target=ACTOR_ID + ) + case = VulnerabilityCase( + name="Test Case", + vulnerability_reports=[REPORT_ID], + attributed_to=ACTOR_ID, + ) + bt_scenario.seed(actor, report, offer, case) + result = bt_scenario.run( + TransitionRMtoValid(report_id=REPORT_ID, offer_id=OFFER_ID), + actor_id=ACTOR_ID, + ) + bt_scenario.assert_success(result) + bt_scenario.assert_rm_state(REPORT_ID, RM.VALID, actor_id=ACTOR_ID) + + def test_failure_when_datalayer_not_available(self) -> None: + """_require_datalayer() guard returns FAILURE when datalayer is None.""" + py_trees.blackboard.Blackboard.storage.clear() + node = TransitionRMtoValid(report_id=REPORT_ID, offer_id=OFFER_ID) + # setup_ports() with no remappings → ports namespace keys; blackboard empty + node.setup_ports() + # initialise() would raise NoDataAvailable; set datalayer=None manually + # to test the _require_datalayer() guard path directly. + node.datalayer = None + node.actor_id = ACTOR_ID + from py_trees.common import Status + + result = node.update() + assert result == Status.FAILURE diff --git a/vultron/core/behaviors/helpers.py b/vultron/core/behaviors/helpers.py index 7e911504c..3be94bd6a 100644 --- a/vultron/core/behaviors/helpers.py +++ b/vultron/core/behaviors/helpers.py @@ -23,6 +23,15 @@ - BT-07-001: BT nodes interact with DataLayer via Protocol interface - BT-07-002: BT nodes use type-safe DataLayer wrappers - BT-07-003: State transitions logged via DataLayer integration helpers + +Per specs/behavior-tree-node-design.yaml BTND-03-009 through BTND-03-011: +- BTND-03-009: Typed-Ports nodes declare blackboard contracts via input_ports() + and output_ports() rather than imperative register_key() calls. +- BTND-03-010: Typed-Ports nodes call setup_ports() in setup() with remappings + {"datalayer": "/datalayer", "actor_id": "/actor_id"} to wire the BTBridge + flat keys. +- BTND-03-011: Typed-Ports nodes read injected values via get_input() in + initialise() rather than direct blackboard attribute access. """ import logging @@ -31,6 +40,7 @@ import py_trees from pydantic import BaseModel from py_trees.common import Status +from py_trees.ports import BehaviourWithPorts, NoDataAvailable, PortInformation from vultron.core.models.case import VulnerabilityCase from vultron.core.models.case_participant import CaseParticipant @@ -686,3 +696,156 @@ def update(self) -> Status: except Exception as e: self.logger.error(f"{self.name}: Error updating actor outbox: {e}") return Status.FAILURE + + +# --------------------------------------------------------------------------- +# Typed-Ports base classes (BTND-03-009 through BTND-03-011) +# --------------------------------------------------------------------------- + +# BTBridge writes datalayer and actor_id as flat root-level blackboard keys. +# Ports nodes remap their port names to these absolute paths so that +# get_input("datalayer") reads the same value that was written by +# BTBridge.setup_tree(). +_DL_KEY = "/datalayer" +_ACTOR_KEY = "/actor_id" + + +class DataLayerConditionWithPorts(BehaviourWithPorts): + """Base class for typed-Ports BT condition nodes with DataLayer access. + + Declares ``datalayer`` and ``actor_id`` as required input ports, remapped + to the flat BTBridge blackboard keys ``/datalayer`` and ``/actor_id``. + Subclasses must implement ``input_ports()``, ``output_ports()``, and + ``update()``. They read injected values via ``get_input()`` in + ``initialise()``. + + Per BTND-03-009: typed port declarations replace imperative register_key(). + Per BTND-03-010: setup_ports() with remappings wires BTBridge flat keys. + Per BTND-03-011: get_input() replaces direct blackboard attribute access. + """ + + logger: logging.Logger # type: ignore[assignment] + + def __init__(self, name: str): + super().__init__(name=name) + self.logger = logging.getLogger( + f"{self.__class__.__module__}.{self.__class__.__name__}" + ) + self.datalayer: CasePersistence | None = None + self.actor_id: str | None = None + + @classmethod + def input_ports(cls) -> dict[str, PortInformation]: + return { + "datalayer": PortInformation(data_type=object, required=True), + "actor_id": PortInformation(data_type=str, required=True), + } + + @classmethod + def output_ports(cls) -> dict[str, PortInformation]: + return {} + + def setup(self, **kwargs: Any) -> None: + self.setup_ports( + port_remappings={"datalayer": _DL_KEY, "actor_id": _ACTOR_KEY} + ) + + def initialise(self) -> None: + self.datalayer = self.get_input("datalayer") + self.actor_id = self.get_input("actor_id") + + def _require_datalayer(self) -> Status | None: + if self.datalayer is None: + self.feedback_message = "DataLayer not available" + return Status.FAILURE + return None + + def _require_datalayer_and_actor(self) -> Status | None: + if self.datalayer is None or self.actor_id is None: + self.feedback_message = "DataLayer or actor_id not available" + return Status.FAILURE + return None + + def update(self) -> Status: + raise NotImplementedError( + f"{self.__class__.__name__}.update() must be implemented" + ) + + +class DataLayerActionWithPorts(BehaviourWithPorts): + """Base class for typed-Ports BT action nodes with DataLayer access. + + Declares ``datalayer``, ``actor_id``, and optionally + ``trigger_activity_factory`` as input ports, remapped to BTBridge flat + keys. Subclasses must implement ``input_ports()``, ``output_ports()``, + and ``update()``. + + Per BTND-03-009 through BTND-03-011. + """ + + logger: logging.Logger # type: ignore[assignment] + + def __init__(self, name: str): + super().__init__(name=name) + self.logger = logging.getLogger( + f"{self.__class__.__module__}.{self.__class__.__name__}" + ) + self.datalayer: CasePersistence | None = None + self.actor_id: str | None = None + self.trigger_activity_factory: "TriggerActivityPort | None" = None + + @classmethod + def input_ports(cls) -> dict[str, PortInformation]: + return { + "datalayer": PortInformation(data_type=object, required=True), + "actor_id": PortInformation(data_type=str, required=True), + "trigger_activity_factory": PortInformation( + data_type=object, required=False + ), + } + + @classmethod + def output_ports(cls) -> dict[str, PortInformation]: + return {} + + def setup(self, **kwargs: Any) -> None: + self.setup_ports( + port_remappings={ + "datalayer": _DL_KEY, + "actor_id": _ACTOR_KEY, + "trigger_activity_factory": "/trigger_activity_factory", + } + ) + + def initialise(self) -> None: + self.datalayer = self.get_input("datalayer") + self.actor_id = self.get_input("actor_id") + try: + self.trigger_activity_factory = self.get_input( + "trigger_activity_factory" + ) + except NoDataAvailable: + self.trigger_activity_factory = None + + def _require_datalayer(self) -> Status | None: + if self.datalayer is None: + self.feedback_message = "DataLayer not available" + return Status.FAILURE + return None + + def _require_datalayer_and_actor(self) -> Status | None: + if self.datalayer is None or self.actor_id is None: + self.feedback_message = "DataLayer or actor_id not available" + return Status.FAILURE + return None + + def _require_factory(self) -> Status | None: + if self.trigger_activity_factory is None: + self.feedback_message = "trigger_activity_factory not available" + return Status.FAILURE + return None + + def update(self) -> Status: + raise NotImplementedError( + f"{self.__class__.__name__}.update() must be implemented" + ) diff --git a/vultron/core/behaviors/report/nodes/conditions.py b/vultron/core/behaviors/report/nodes/conditions.py index e91d7639c..e8cc9e95c 100644 --- a/vultron/core/behaviors/report/nodes/conditions.py +++ b/vultron/core/behaviors/report/nodes/conditions.py @@ -19,6 +19,7 @@ from vultron.core.behaviors.helpers import ( DataLayerCondition, + DataLayerConditionWithPorts, FindParticipantByActorIdNode, ) from vultron.core.states.rm import RM @@ -26,14 +27,17 @@ from vultron.errors import VultronInvalidStateTransitionError -class CheckRMStateValid(DataLayerCondition): - """ - Check if report is already in RM.VALID state. +class CheckRMStateValid(DataLayerConditionWithPorts): + """Check if report is already in RM.VALID state. Returns SUCCESS if report status is RM.VALID (early exit optimization). Returns FAILURE if report is in any other state. - This node implements the early-exit check from the simulation BT. + Input ports (inherited + declared): + datalayer (object, required): CasePersistence, remapped to /datalayer. + actor_id (str, required): Executing actor ID, remapped to /actor_id. + + Per BTND-03-009: typed port declarations replace register_key(). """ def __init__( @@ -42,28 +46,21 @@ def __init__( sender_actor_id: str | None = None, name: str | None = None, ): - """ - Initialize CheckRMStateValid node. + """Initialize CheckRMStateValid node. Args: - report_id: ID of VulnerabilityReport to check + report_id: ID of VulnerabilityReport to check. sender_actor_id: Explicit actor ID to use instead of the blackboard ``actor_id``. Thread this in when the tree runs under ``receiving_actor_id`` but the RM check must target the message sender (ADR-0022 single-BT pattern). - name: Optional custom node name (defaults to class name) + name: Optional custom node name (defaults to class name). """ super().__init__(name=name or self.__class__.__name__) self.report_id = report_id self.sender_actor_id = sender_actor_id def update(self) -> Status: - """ - Check if report is in RM.VALID state. - - Returns: - SUCCESS if report is RM.VALID, FAILURE otherwise - """ if (f := self._require_datalayer()) is not None: return f assert self.datalayer is not None @@ -88,14 +85,17 @@ def update(self) -> Status: return Status.FAILURE -class CheckRMStateReceivedOrInvalid(DataLayerCondition): - """ - Check if report is in RM.RECEIVED or RM.INVALID state. +class CheckRMStateReceivedOrInvalid(DataLayerConditionWithPorts): + """Check if report is in RM.RECEIVED or RM.INVALID state. Returns SUCCESS if report is in acceptable precondition state. Returns FAILURE if report is in any other state (e.g., CLOSED, VALID). - This node implements the precondition check from the simulation BT. + Input ports (inherited + declared): + datalayer (object, required): CasePersistence, remapped to /datalayer. + actor_id (str, required): Executing actor ID, remapped to /actor_id. + + Per BTND-03-009: typed port declarations replace register_key(). """ def __init__( @@ -104,28 +104,21 @@ def __init__( sender_actor_id: str | None = None, name: str | None = None, ): - """ - Initialize CheckRMStateReceivedOrInvalid node. + """Initialize CheckRMStateReceivedOrInvalid node. Args: - report_id: ID of VulnerabilityReport to check + report_id: ID of VulnerabilityReport to check. sender_actor_id: Explicit actor ID to use instead of the blackboard ``actor_id``. Thread this in when the tree runs under ``receiving_actor_id`` but the RM check must target the message sender (ADR-0022 single-BT pattern). - name: Optional custom node name (defaults to class name) + name: Optional custom node name (defaults to class name). """ super().__init__(name=name or self.__class__.__name__) self.report_id = report_id self.sender_actor_id = sender_actor_id def update(self) -> Status: - """ - Check if report is in RM.RECEIVED or RM.INVALID state. - - Returns: - SUCCESS if report is in acceptable state, FAILURE otherwise - """ if (f := self._require_datalayer()) is not None: return f assert self.datalayer is not None @@ -151,38 +144,33 @@ def update(self) -> Status: return Status.SUCCESS -class EnsureEmbargoExists(DataLayerCondition): - """ - Check that the case linked to this report has an active embargo. +class EnsureEmbargoExists(DataLayerConditionWithPorts): + """Check that the case linked to this report has an active embargo. Returns SUCCESS if the case exists and has a non-None ``active_embargo``. Returns FAILURE if the case is not found or its ``active_embargo`` is None. - This node implements DUR-07-004: an embargo end time MUST be established - before the case reaches RM.VALID. It runs after ``TransitionRMtoValid`` - in ``ValidationActions`` to confirm that the default embargo seeded at - RM.RECEIVED (DUR-07-002, via ``InitializeDefaultEmbargoNode``) is - present before validation completes. + Implements DUR-07-004: an embargo end time MUST be established before the + case reaches RM.VALID. + + Input ports (inherited + declared): + datalayer (object, required): CasePersistence, remapped to /datalayer. + actor_id (str, required): Executing actor ID, remapped to /actor_id. + + Per BTND-03-009: typed port declarations replace register_key(). """ def __init__(self, report_id: str, name: str | None = None): - """ - Initialize EnsureEmbargoExists node. + """Initialize EnsureEmbargoExists node. Args: - report_id: ID of VulnerabilityReport whose linked case to check - name: Optional custom node name (defaults to class name) + report_id: ID of VulnerabilityReport whose linked case to check. + name: Optional custom node name (defaults to class name). """ super().__init__(name=name or self.__class__.__name__) self.report_id = report_id def update(self) -> Status: - """ - Verify the case linked to this report has an active embargo. - - Returns: - SUCCESS if case has active_embargo, FAILURE otherwise - """ if (f := self._require_datalayer()) is not None: return f assert self.datalayer is not None diff --git a/vultron/core/behaviors/report/nodes/rm_transitions.py b/vultron/core/behaviors/report/nodes/rm_transitions.py index cb5322cdd..b8e4b91a4 100644 --- a/vultron/core/behaviors/report/nodes/rm_transitions.py +++ b/vultron/core/behaviors/report/nodes/rm_transitions.py @@ -17,7 +17,10 @@ from py_trees.common import Status -from vultron.core.behaviors.helpers import DataLayerAction +from vultron.core.behaviors.helpers import ( + DataLayerAction, + DataLayerActionWithPorts, +) from vultron.core.models.dimensions import PecDimension, RmDimension from vultron.core.models.participant_status import ParticipantStatus from vultron.core.models.case import VulnerabilityCase @@ -96,14 +99,19 @@ def _transition_case_participant_rm( return Status.SUCCESS -class TransitionRMtoValid(DataLayerAction): - """ - Transition report to RM.VALID and offer to ACCEPTED. +class TransitionRMtoValid(DataLayerActionWithPorts): + """Transition report to RM.VALID and offer to ACCEPTED. + + Updates both report status (RM.VALID) and the CaseParticipant RM state. + Logs state transitions at INFO level. - Updates both report status (RM.VALID) and offer status (ACCEPTED) in the - status layer. Logs state transitions at INFO level. + Input ports (inherited + declared): + datalayer (object, required): CasePersistence, remapped to /datalayer. + actor_id (str, required): Executing actor ID, remapped to /actor_id. + trigger_activity_factory (object, optional): remapped to + /trigger_activity_factory. - This node implements the core state transition from the validate_report handler. + Per BTND-03-009: typed port declarations replace register_key(). """ def __init__( @@ -113,17 +121,16 @@ def __init__( sender_actor_id: str | None = None, name: str | None = None, ): - """ - Initialize TransitionRMtoValid node. + """Initialize TransitionRMtoValid node. Args: - report_id: ID of VulnerabilityReport to update - offer_id: ID of Offer to update + report_id: ID of VulnerabilityReport to update. + offer_id: ID of Offer to update. sender_actor_id: Explicit actor ID to use instead of the blackboard ``actor_id``. Thread this in when the tree runs under ``receiving_actor_id`` but the RM transition must target the message sender (ADR-0022 single-BT pattern). - name: Optional custom node name (defaults to class name) + name: Optional custom node name (defaults to class name). """ super().__init__(name=name or self.__class__.__name__) self.report_id = report_id