feat: add input port update subscriptions

docs/INPUT_PORT_CALLBACKS.md

@@ -0,0 +1,102 @@
+# Input port callbacks
+
+`TreeNode::subscribeToInputPort()` observes writes to the blackboard key backing
+an input port. It is an instance method, so a callback can use node-specific
+state. Port declarations in `providedPorts()` remain unchanged.
+
+For example, a node can react to a new topic name before its next tick:
+
+```cpp
+class TopicListener : public BT::SyncActionNode
+{
+public:
+  TopicListener(const std::string& name, const BT::NodeConfig& config)
+    : BT::SyncActionNode(name, config)
+  {
+    topic_subscription_ = subscribeToInputPort("topic", [this] {
+      if(auto topic = getInput<std::string>("topic"))
+      {
+        configureSubscription(topic.value());
+      }
+    });
+  }
+
+  static BT::PortsList providedPorts()
+  {
+    return { BT::InputPort<std::string>("topic") };
+  }
+
+  BT::NodeStatus tick() override;
+
+private:
+  void configureSubscription(const std::string& topic);
+  BT::Blackboard::EntryUpdatedSubscriber topic_subscription_;
+};
+```
+
+The application must stop writers and wait for callbacks to finish before
+destroying the node. Resetting the handle alone is not a synchronization barrier.
+The example assumes the application serializes writes and access to the node's
+subscription state.
+
+## Notification contract
+
+- Retain the returned handle to keep the callback registered. Resetting or
+  destroying it unsubscribes; a callback already selected for invocation may
+  still complete.
+- Registration does not invoke the callback. Initialize from `getInput()`
+  separately when an initial value is needed. With concurrent writers, coordinate
+  initialization with the application's synchronization.
+- Successful writes notify even if the new value equals the previous value.
+  Generic port values do not necessarily have an equality operation.
+- Callbacks run synchronously on the writer's thread, outside blackboard locks.
+  They may call `getInput()`, write other keys, and change subscriptions.
+  Writing the observed key can recursively invoke callbacks.
+- Concurrent writes may invoke the same callback concurrently. A callback reads
+  the latest value, not a snapshot of the triggering write; delivery order is
+  not guaranteed across writers.
+- Callback exceptions propagate after the write has committed. As with other
+  synchronous signals, a throwing callback prevents the remaining callbacks in
+  that notification from running.
+
+## Supported inputs and writes
+
+Explicit subtree remapping, automatic remapping, `{=}`, `{@root_key}`, and
+blackboard-backed port defaults are supported. A literal or constant default
+returns an empty handle because it cannot change through the blackboard.
+Missing inputs, output-only ports, invalid blackboards, and empty callbacks are
+reported with `LogicError`.
+
+Subscriptions bind to the blackboard entry the key resolves to at registration
+time. Reconfiguring port or subtree remapping requires a new subscription.
+A subscription lives in the entry: `unset()` (or `cloneInto()` removing the key)
+drops it, without notifying, and an entry re-created afterwards under the same
+name is not observed. Subscribe again after re-creating the key.
+
+If the entry doesn't exist yet, subscribing creates it, so the key is listed by
+`getKeyNames()` before its first assignment, with an empty value.
+`subscribeToInputPort()` creates it with the port's type, as tree creation does
+for remapped ports. `Blackboard::subscribeToEntry()` creates a generic
+(`AnyTypeAllowed`) entry, like `set<std::string>()` does: the first strongly typed
+write sets its type. Note that, as for any existing generic entry, a first write
+of a `std::string` locks such an entry to `std::string`, whereas without the
+subscription `set<std::string>()` would have created a generic entry.
+
+Writes through `Blackboard::set()`, `TreeNode::setOutput()`, scripting assignments,
+JSON import, and `Blackboard::cloneInto()` notify. Clone notifications run after
+the entire snapshot has been copied, allowing a callback to read other restored
+entries. Raw writes through `Entry`, `AnyPtrLocked`, or shared objects are not
+observable automatically. Call `Blackboard::notifyEntryUpdated()` after releasing
+the entry lock to explicitly publish such a write.
+
+## Compatibility
+
+This adds an opt-in API; existing port declarations and unsubscribed nodes keep
+their behavior. It is ABI compatible with 4.x: the subscribers are stored in a
+member appended at the end of `Blackboard::Entry`, which is allocated only inside
+the library, and the layout of `Blackboard` is unchanged. Without subscribers, a
+write only checks that member for null under the entry lock it already holds.
+
+Every write path (`set()`, `setOutput()`, scripting assignments) is inline code
+in the headers. Writes performed by binaries compiled against older headers
+(4.10 and earlier) store the value but do not notify, until they are rebuilt.

include/behaviortree_cpp/blackboard.h

@@ -10,6 +10,7 @@
 #include "behaviortree_cpp/utils/locked_reference.hpp"
 #include "behaviortree_cpp/utils/polymorphic_cast_registry.hpp"
 #include "behaviortree_cpp/utils/safe_any.hpp"
+#include "behaviortree_cpp/utils/thread_safe_signal.h"
 
 #include <memory>
 #include <mutex>
@@ -39,6 +40,9 @@ class Blackboard
 {
 public:
   using Ptr = std::shared_ptr<Blackboard>;
+  using EntryUpdatedSignal = ThreadSafeSignal<>;
+  using EntryUpdatedSubscriber = EntryUpdatedSignal::Subscriber;
+  using EntryUpdatedCallback = EntryUpdatedSignal::CallableFunction;
 
 protected:
   // This is intentionally protected. Use Blackboard::create instead
@@ -62,6 +66,12 @@ class Blackboard
     // timestamp since epoch
     std::chrono::nanoseconds stamp = std::chrono::nanoseconds{ 0 };
 
+    // Subscribers of subscribeToEntry(), created on the first subscription.
+    // Protected by entry_mutex. It must remain the last member: Entry is
+    // allocated only inside the library, so appending it keeps the offsets of
+    // the members above (accessed by inline code) unchanged.
+    std::shared_ptr<EntryUpdatedSignal> update_signal;
+
     Entry(const TypeInfo& _info) : info(_info)
     {}
 
@@ -119,6 +129,30 @@ class Blackboard
   template <typename T>
   void set(const std::string& key, const T& value);
 
+  /** Subscribe to successful writes to the entry of a key.
+   * Subtree remapping and root keys are resolved at subscription time. If the
+   * entry doesn't exist, a generic (AnyTypeAllowed) one is created: the first
+   * strongly typed write sets its type, as usual. The subscription is bound to
+   * that entry: unset() drops it and a re-created entry is not observed.
+   * No initial notification is sent, and assigning the same value still notifies.
+   * Writes compiled against headers older than this API do not notify.
+   *
+   * Callbacks run synchronously on the writer's thread, outside blackboard
+   * locks, and may read the latest value. Concurrent writers may invoke a
+   * callback concurrently; intermediate values are not retained. Exceptions
+   * propagate to the writer after the value has been stored.
+   * Resetting the handle prevents future calls but does not wait for a call
+   * already in progress. Stop writers before destroying callback targets.
+   */
+  [[nodiscard]] EntryUpdatedSubscriber subscribeToEntry(const std::string& key,
+                                                        EntryUpdatedCallback callback);
+
+  /** Notify subscribers after directly modifying an entry. Release its lock
+   * before calling this function. Raw Entry/AnyPtrLocked mutations do not
+   * notify automatically; set(), scripts, JSON import and cloneInto() do.
+   */
+  void notifyEntryUpdated(const std::string& key);
+
   void unset(const std::string& key);
 
   [[nodiscard]] const TypeInfo* entryInfo(const std::string& key);
@@ -296,6 +330,8 @@ inline void Blackboard::set(const std::string& key, const T& value)
   // follow the same remapping and return the pre-existing entry, and writing
   // to it blindly bypasses the type lock and the string conversion.
   auto entry_ptr = getEntry(key);
+  // copied under entry_mutex, notified after releasing it
+  std::shared_ptr<EntryUpdatedSignal> update_signal;
   if(!entry_ptr)
   {
     // create a new entry
@@ -319,14 +355,15 @@ inline void Blackboard::set(const std::string& key, const T& value)
     entry->value = new_value;
     entry->sequence_id++;
     entry->stamp = std::chrono::steady_clock::now().time_since_epoch();
+    update_signal = entry->update_signal;
   }
   else
   {
     // this is not the first time we set this entry, we need to check
     // if the type is the same or not.
     Entry& entry = *entry_ptr;
 
-    std::scoped_lock scoped_lock(entry.entry_mutex);
+    std::unique_lock scoped_lock(entry.entry_mutex);
 
     Any& previous_any = entry.value;
     Any new_value(value);
@@ -339,6 +376,12 @@ inline void Blackboard::set(const std::string& key, const T& value)
       entry.sequence_id++;
       entry.stamp = std::chrono::steady_clock::now().time_since_epoch();
       previous_any = std::move(new_value);
+      update_signal = entry.update_signal;
+      scoped_lock.unlock();
+      if(update_signal)
+      {
+        update_signal->notify();
+      }
       return;
     }
 
@@ -393,6 +436,11 @@ inline void Blackboard::set(const std::string& key, const T& value)
     }
     entry.sequence_id++;
     entry.stamp = std::chrono::steady_clock::now().time_since_epoch();
+    update_signal = entry.update_signal;
+  }
+  if(update_signal)
+  {
+    update_signal->notify();
   }
 }
 

include/behaviortree_cpp/scripting/operators.hpp

@@ -562,7 +562,7 @@ struct ExprAssignment : ExprBase
     }
     auto value = rhs->evaluate(env);
 
-    std::scoped_lock lock(entry->entry_mutex);
+    std::unique_lock lock(entry->entry_mutex);
     auto* dst_ptr = &entry->value;
 
     auto errorPrefix = [dst_ptr, &key]() {
@@ -624,7 +624,14 @@ struct ExprAssignment : ExprBase
       }
       entry->sequence_id++;
       entry->stamp = std::chrono::steady_clock::now().time_since_epoch();
-      return *dst_ptr;
+      const Any result = *dst_ptr;
+      auto update_signal = entry->update_signal;
+      lock.unlock();
+      if(update_signal)
+      {
+        update_signal->notify();
+      }
+      return result;
     }
 
     if(dst_ptr->empty())
@@ -680,7 +687,14 @@ struct ExprAssignment : ExprBase
     temp_variable.copyInto(*dst_ptr);
     entry->sequence_id++;
     entry->stamp = std::chrono::steady_clock::now().time_since_epoch();
-    return *dst_ptr;
+    const Any result = *dst_ptr;
+    auto update_signal = entry->update_signal;
+    lock.unlock();
+    if(update_signal)
+    {
+      update_signal->notify();
+    }
+    return result;
   }
 };
 }  // namespace BT::Ast

include/behaviortree_cpp/tree_node.h

@@ -214,6 +214,21 @@ class TreeNode
   [[nodiscard]] StatusChangeSubscriber
   subscribeToStatusChange(StatusChangeCallback callback);
 
+  /** Subscribe to writes to the blackboard key backing an input port.
+   * Unlike providedPorts(), this is an instance method: a callback may use
+   * node-specific state and getInput(). Retain the returned handle for as long
+   * as the callback should remain active. See Blackboard::subscribeToEntry()
+   * for threading and lifetime requirements.
+   *
+   * Constant inputs return an empty handle. Missing inputs, output-only ports
+   * and empty callbacks throw LogicError. Defaults and {=} remapping are
+   * supported. Changing the port mapping later does not rebind subscriptions.
+   * The subscription is bound to the blackboard entry, which is created with
+   * the port's type if it doesn't exist yet; unset() of the key drops it.
+   */
+  [[nodiscard]] Blackboard::EntryUpdatedSubscriber
+  subscribeToInputPort(const std::string& key, Blackboard::EntryUpdatedCallback callback);
+
   /** This method attaches to the TreeNode a callback with signature:
      *
      *     NodeStatus callback(TreeNode& node)

include/behaviortree_cpp/utils/thread_safe_signal.h

@@ -0,0 +1,86 @@
+#pragma once
+
+#include <functional>
+#include <memory>
+#include <mutex>
+#include <vector>
+
+namespace BT
+{
+/** A signal supporting concurrent notification and subscription.
+ * Callbacks run outside the subscription lock and may be invoked concurrently.
+ * Dropping a subscriber prevents future calls, but does not wait for callbacks
+ * already selected for invocation. The owner must synchronize their lifetime
+ * with callback targets.
+ */
+template <typename... CallableArgs>
+class ThreadSafeSignal
+{
+public:
+  using CallableFunction = std::function<void(CallableArgs...)>;
+
+  class Subscription
+  {
+    friend class ThreadSafeSignal;
+
+  public:
+    explicit Subscription(CallableFunction func)
+      : callback_(std::make_shared<CallableFunction>(std::move(func)))
+    {}
+
+  private:
+    std::shared_ptr<CallableFunction> callback_;
+  };
+
+  using Subscriber = std::shared_ptr<Subscription>;
+
+  [[nodiscard]] Subscriber subscribe(CallableFunction func)
+  {
+    auto subscriber = std::make_shared<Subscription>(std::move(func));
+    const std::scoped_lock lock(mutex_);
+    removeExpired();
+    subscribers_.emplace_back(subscriber);
+    return subscriber;
+  }
+
+  void notify(CallableArgs... args)
+  {
+    std::vector<std::weak_ptr<Subscription>> subscribers;
+    {
+      const std::scoped_lock lock(mutex_);
+      removeExpired();
+      subscribers = subscribers_;
+    }
+    for(const auto& weak_subscriber : subscribers)
+    {
+      if(auto subscriber = weak_subscriber.lock())
+      {
+        // Keep the callable alive without keeping its subscription alive.
+        // A callback can reset its own handle and recursively notify safely.
+        auto callback = subscriber->callback_;
+        subscriber.reset();
+        (*callback)(args...);
+      }
+    }
+  }
+
+private:
+  void removeExpired()
+  {
+    for(auto it = subscribers_.begin(); it != subscribers_.end();)
+    {
+      if(it->expired())
+      {
+        it = subscribers_.erase(it);
+      }
+      else
+      {
+        ++it;
+      }
+    }
+  }
+
+  std::mutex mutex_;
+  std::vector<std::weak_ptr<Subscription>> subscribers_;
+};
+}  // namespace BT