XRebalancePredictor: the live persistence forecaster (off by default)

Phase 2 of the history+prediction design: the module that closes the
loop from observation to synthetic re-assertion.  After each hourly
askrene-age pass over the clboss-xrebalance layer, XMoveFunds now
raises Msg::XRebalanceLayerAged carrying the aging cutoff (on the
failure path too -- a skipped trim only leaves stale entries, which
is safe).  The new Boss::Mod::XRebalancePredictor subscribes, reads
the XRebalanceHistory observation store, runs the pure regime-walk
algorithm per channel direction, and re-asserts the surviving
walls/floors into the routed layer via askrene-inform-channel --
but only for directions whose newest real observation predates the
cutoff: directions with live evidence need no synthesis.  Synthetic
assertions are never recorded back into the observation store (no
self-confirmation).

OFF BY DEFAULT.  The master switch is the dynamic option
clboss-xrebalance-predict-horizon-max-secs (0 = disabled, the
default; 86400 is the intended enabled value -- and since an
asserted wall is never contradicted by routing, this cap IS the
wall re-test schedule).  The other constants are dynamic options
mirroring the read-only spot-check parameters: -horizon-frac (2.0),
-min-samples (2), -wall-margin (1.0), and -floor-factor, which
defaults to 0 = walls-only operation (floors are the riskier half:
a too-high floor attracts flow and costs a failed part to
self-correct).  Note the live floor default deliberately differs
from the spot-check commands' 0.9.  Also dormant unless
clboss-rebalance-mode is xrebalance.

The per-cycle decision is a pure static XRebalancePredictor::plan
(group directions, candidacy gate, predict, collect asserting
sides, skip amount-0 degenerates), unit-tested directly; the module
shell only reads the table, executes the plan, logs one Info
summary per asserting cycle, and reports an xrebalance_predictor
section (params + last-cycle counts) in clboss-status.
kind_is_bound (stored TEXT kind to bound side) is promoted into
XRebalancePredict and shared with XRebalanceHistory.

Boss/Mod/XMoveFunds/Main.cpp

@@ -12,6 +12,7 @@
 #include"Boss/Msg/Option.hpp"
 #include"Boss/Msg/TimerRandomHourly.hpp"
 #include"Boss/Msg/XRebalanceAttribution.hpp"
+#include"Boss/Msg/XRebalanceLayerAged.hpp"
 #include"Boss/Msg/XRebalanceObservation.hpp"
 #include"Boss/concurrent.hpp"
 #include"Boss/log.hpp"
@@ -611,9 +612,10 @@ private:
 	 * channel_update payload (gossmap_local_updatechan merges).
 	 */
 	Ev::Io<void> age_xrebalance_layer() {
-		return Ev::lift().then([this]() {
-			auto cutoff = std::uint64_t(std::time(nullptr))
-				    - aging_window_secs;
+		auto aged_time = std::make_shared<std::uint64_t>(0);
+		return Ev::lift().then([this, aged_time]() {
+			*aged_time = std::uint64_t(std::time(nullptr));
+			auto cutoff = *aged_time - aging_window_secs;
 			auto parms = Json::Out()
 				.start_object()
 					.field( "layer"
@@ -660,6 +662,18 @@ private:
 						  "until next successful "
 						  "aging pass)."
 					);
+		}).then([this, aged_time]() {
+			/* End-of-expiration-cycle hook: the persistence
+			 * forecaster (XRebalancePredictor) re-asserts
+			 * durable knowledge for directions the trim just
+			 * left without live evidence.  Raised on the
+			 * failure path too -- a skipped trim only means
+			 * stale entries remain, which is safe for
+			 * subscribers.  */
+			auto m = Msg::XRebalanceLayerAged{
+				*aged_time,
+				*aged_time - aging_window_secs};
+			return bus.raise(std::move(m));
 		});
 	}
 

Boss/Mod/XRebalanceHistory.cpp

@@ -391,21 +391,6 @@ private:
 		po.end_object();
 	}
 
-	/* Map a stored observation kind to its bound side; returns
-	 * false for kinds that are not per-channel liquidity bounds
-	 * (node_fail).  */
-	static bool kind_is_bound(std::string const& kind, bool& is_fail) {
-		if (kind == "success") {
-			is_fail = false;
-			return true;
-		}
-		if (kind == "liquidity_fail" || kind == "policy_fail") {
-			is_fail = true;
-			return true;
-		}
-		return false;
-	}
-
 	Ev::Io<void> report(Msg::CommandRequest const& req) {
 		auto id = req.id;
 		auto paramfail = [this, id]() {
@@ -553,7 +538,8 @@ private:
 
 				if (want_predictions) {
 					auto is_fail = false;
-					if (kind_is_bound(kind, is_fail))
+					if (XRebalancePredict::kind_is_bound(
+						kind, is_fail))
 						bounds_by_dir[dir].push_back(
 						    {time, is_fail, amount});
 				}
@@ -664,7 +650,8 @@ private:
 				 * (with decline reasons).  */
 				auto& bounds = groups[{row_scid, dir}];
 				auto is_fail = false;
-				if (kind_is_bound(kind, is_fail))
+				if (XRebalancePredict::kind_is_bound(
+					kind, is_fail))
 					bounds.push_back(
 					    {time, is_fail, amount});
 			}

Boss/Mod/XRebalancePredict.cpp

@@ -39,6 +39,18 @@ std::string samples_reason(std::size_t have, std::size_t need) {
 
 namespace Boss { namespace Mod { namespace XRebalancePredict {
 
+bool kind_is_bound(std::string const& kind, bool& is_fail) {
+	if (kind == "success") {
+		is_fail = false;
+		return true;
+	}
+	if (kind == "liquidity_fail" || kind == "policy_fail") {
+		is_fail = true;
+		return true;
+	}
+	return false;
+}
+
 Result predict( std::vector<Bound> bounds
 	      , std::uint64_t now
 	      , Params const& params

Boss/Mod/XRebalancePredict.hpp

@@ -107,6 +107,12 @@ Result predict( std::vector<Bound> bounds
 	      , Params const& params
 	      );
 
+/* Map a stored observation kind (the TEXT values of the
+ * XRebalanceHistory table: success / liquidity_fail / policy_fail /
+ * node_fail) to its bound side.  Returns false for kinds that are
+ * not per-channel liquidity bounds (node_fail).  */
+bool kind_is_bound(std::string const& kind, bool& is_fail);
+
 }}}
 
 #endif /* !defined(BOSS_MOD_XREBALANCEPREDICT_HPP) */

Boss/Mod/XRebalancePredictor.cpp

@@ -0,0 +1,486 @@
+#include"Boss/Mod/AskreneLayer.hpp"
+#include"Boss/Mod/Rpc.hpp"
+#include"Boss/Mod/XRebalancePredictor.hpp"
+#include"Boss/ModG/RebalanceModeProxy.hpp"
+#include"Boss/Msg/DbResource.hpp"
+#include"Boss/Msg/Init.hpp"
+#include"Boss/Msg/ManifestOption.hpp"
+#include"Boss/Msg/Manifestation.hpp"
+#include"Boss/Msg/Option.hpp"
+#include"Boss/Msg/OptionType.hpp"
+#include"Boss/Msg/ProvideStatus.hpp"
+#include"Boss/Msg/SolicitStatus.hpp"
+#include"Boss/Msg/XRebalanceLayerAged.hpp"
+#include"Boss/RebalanceMode.hpp"
+#include"Boss/concurrent.hpp"
+#include"Boss/log.hpp"
+#include"Ev/Io.hpp"
+#include"Jsmn/Object.hpp"
+#include"Json/Out.hpp"
+#include"Ln/Amount.hpp"
+#include"Ln/Scid.hpp"
+#include"S/Bus.hpp"
+#include"Sqlite3.hpp"
+#include"Util/make_unique.hpp"
+#include<cinttypes>
+#include<cstdint>
+#include<map>
+#include<string>
+#include<utility>
+#include<vector>
+
+namespace {
+
+auto const opt_horizon_max =
+	std::string("clboss-xrebalance-predict-horizon-max-secs");
+auto const opt_horizon_frac =
+	std::string("clboss-xrebalance-predict-horizon-frac");
+auto const opt_min_samples =
+	std::string("clboss-xrebalance-predict-min-samples");
+auto const opt_wall_margin =
+	std::string("clboss-xrebalance-predict-wall-margin");
+auto const opt_floor_factor =
+	std::string("clboss-xrebalance-predict-floor-factor");
+
+/* The live defaults deliberately differ from the spot-check
+ * defaults (XRebalancePredict::default_params) in two places:
+ * horizon-max 0 keeps the predictor OFF until the operator opts in,
+ * and floor-factor 0 makes the first enablement walls-only (floors
+ * are the riskier half: a too-high floor attracts flow and costs a
+ * failed part to self-correct).  */
+auto constexpr default_horizon_max = std::uint64_t(0);
+auto constexpr default_horizon_frac = double(2.0);
+auto constexpr default_min_samples = std::uint64_t(2);
+auto constexpr default_wall_margin = double(1.0);
+auto constexpr default_floor_factor = double(0.0);
+
+/* Tolerate both a JSON number (startup primitive) and a JSON
+ * string (runtime setconfig encoding).  */
+bool parse_double(Jsmn::Object const& v, double& out) {
+	try {
+		if (v.is_number()) {
+			out = double(v);
+			return true;
+		}
+		if (v.is_string()) {
+			out = std::stod(std::string(v));
+			return true;
+		}
+	} catch (std::exception const&) { }
+	return false;
+}
+bool parse_u64(Jsmn::Object const& v, std::uint64_t& out) {
+	try {
+		if (v.is_number()) {
+			out = std::uint64_t(double(v));
+			return true;
+		}
+		if (v.is_string()) {
+			out = std::stoull(std::string(v));
+			return true;
+		}
+	} catch (std::exception const&) { }
+	return false;
+}
+
+}
+
+namespace Boss { namespace Mod {
+
+XRebalancePredictor::Plan
+XRebalancePredictor::plan( std::vector<Row> const& rows
+			 , std::uint64_t cutoff
+			 , std::uint64_t now
+			 , XRebalancePredict::Params const& params
+			 ) {
+	struct Group {
+		std::vector<XRebalancePredict::Bound> bounds;
+		/* Newest observation of ANY kind: a node_fail is not
+		 * a liquidity bound, but it IS fresh real data, and
+		 * candidacy is about whether the routed layer still
+		 * carries live evidence for this direction.  */
+		std::uint64_t newest = 0;
+	};
+	auto groups = std::map< std::pair<std::string, std::uint32_t>
+			      , Group>();
+	for (auto const& row : rows) {
+		auto& g = groups[{row.scid, row.dir}];
+		if (row.time > g.newest)
+			g.newest = row.time;
+		auto is_fail = false;
+		if (XRebalancePredict::kind_is_bound(row.kind, is_fail))
+			g.bounds.push_back(
+			    {row.time, is_fail, row.amount_msat});
+	}
+
+	auto result = Plan();
+	result.directions = groups.size();
+	result.candidates = 0;
+	for (auto const& e : groups) {
+		/* Fresh real data: the routed layer already carries
+		 * live evidence; nothing to synthesize.  */
+		if (e.second.newest >= cutoff)
+			continue;
+		++result.candidates;
+		auto res = XRebalancePredict::predict(
+		    e.second.bounds, now, params);
+		/* amount 0 would be a degenerate inform (a wall at 0
+		 * is a full exclusion we did not observe; a floor at
+		 * 0 is a no-op) -- can arise from margins/factors
+		 * scaling a tiny bound down.  Skip.  */
+		if (res.wall.would_assert && res.wall.amount_msat > 0)
+			result.assertions.push_back(
+			    { e.first.first, e.first.second
+			    , true, res.wall.amount_msat});
+		if (res.floor.would_assert && res.floor.amount_msat > 0)
+			result.assertions.push_back(
+			    { e.first.first, e.first.second
+			    , false, res.floor.amount_msat});
+	}
+	return result;
+}
+
+class XRebalancePredictor::Impl {
+private:
+	S::Bus& bus;
+	std::function<double()> get_now;
+	Sqlite3::Db db;
+	Boss::Mod::Rpc* rpc;
+	ModG::RebalanceModeProxy mode_proxy;
+
+	/* Live (dynamic-option) parameter values.  */
+	std::uint64_t horizon_max_secs;
+	double horizon_frac;
+	std::uint64_t min_samples;
+	double wall_margin;
+	double floor_factor;
+
+	/* Last-cycle summary, for clboss-status.  */
+	std::uint64_t last_run_time;
+	std::size_t last_directions;
+	std::size_t last_candidates;
+	std::size_t last_walls;
+	std::size_t last_floors;
+
+	void start() {
+		bus.subscribe<Msg::DbResource
+			     >([this](Msg::DbResource const& r) {
+			db = r.db;
+			return Ev::lift();
+		});
+		bus.subscribe<Msg::Init
+			     >([this](Msg::Init const& init) {
+			rpc = &init.rpc;
+			return Ev::lift();
+		});
+		bus.subscribe<Msg::Manifestation
+			     >([this](Msg::Manifestation const&) {
+			return manifest_int_option( opt_horizon_max
+						  , default_horizon_max,
+				"Maximum forecast horizon (seconds) of the "
+				"xrebalance persistence forecaster, AND its "
+				"master switch: 0 (the default) disables "
+				"synthetic re-assertion entirely.  When "
+				"enabled, after each hourly aging pass the "
+				"forecaster re-asserts walls/floors for "
+				"channel directions with no live evidence, "
+				"for up to min(this, horizon-frac * the "
+				"regime's evidence span) past the last "
+				"observation.  Since an asserted wall is "
+				"never contradicted by routing (the router "
+				"will not attempt amounts above it), this "
+				"cap IS the wall re-test schedule.  86400 "
+				"(24h) is the intended enabled value.")
+			     + manifest_double_option( opt_horizon_frac
+						     , default_horizon_frac,
+				"Forecast horizon as a multiple of the "
+				"regime's evidence span (newest - oldest "
+				"consistent observation).  2.0: two "
+				"observations an hour apart are asserted "
+				"for two hours past the newest.")
+			     + manifest_int_option( opt_min_samples
+						  , default_min_samples,
+				"Minimum observations of a side's kind "
+				"(failures for walls, successes for "
+				"floors) in the current regime before "
+				"that side is asserted.")
+			     + manifest_double_option( opt_wall_margin
+						     , default_wall_margin,
+				"Multiplier on asserted wall amounts.  "
+				">= 1.0 biases errors high, which "
+				"self-corrects (a too-high wall costs a "
+				"failed part that writes a fresh real "
+				"bound; a too-low wall is sticky until "
+				"the horizon).")
+			     + manifest_double_option( opt_floor_factor
+						     , default_floor_factor,
+				"Multiplier on asserted floor amounts; "
+				"<= 1.0 is conservative.  0 (the default) "
+				"disables floor assertion entirely "
+				"(walls-only operation; floors are the "
+				"riskier half).");
+		});
+		bus.subscribe<Msg::Option
+			     >([this](Msg::Option const& o) {
+			return handle_option(o);
+		});
+		bus.subscribe<Msg::XRebalanceLayerAged
+			     >([this](Msg::XRebalanceLayerAged const& m) {
+			if (horizon_max_secs == 0)
+				return Ev::lift();
+			if (!db || !rpc)
+				return Ev::lift();
+			auto cutoff = m.cutoff;
+			return Boss::concurrent(run(cutoff));
+		});
+		bus.subscribe<Msg::SolicitStatus
+			     >([this](Msg::SolicitStatus const&) {
+			return status();
+		});
+	}
+
+	Ev::Io<void> manifest_int_option( std::string const& name
+					, std::uint64_t dflt
+					, std::string desc
+					) {
+		return bus.raise(Msg::ManifestOption{
+			name, Msg::OptionType_Int,
+			Json::Out::direct(dflt), std::move(desc),
+			true /* dynamic */
+		});
+	}
+
+	Ev::Io<void> manifest_double_option( std::string const& name
+					   , double dflt
+					   , std::string desc
+					   ) {
+		return bus.raise(Msg::ManifestOption{
+			name, Msg::OptionType_String,
+			Json::Out::direct(dflt), std::move(desc),
+			true /* dynamic */
+		});
+	}
+
+	Ev::Io<void> handle_option(Msg::Option const& o) {
+		if (o.name == opt_horizon_max) {
+			auto v = std::uint64_t(0);
+			if (!parse_u64(o.value, v))
+				return bad_option(o.name);
+			horizon_max_secs = v;
+			if (horizon_max_secs == 0)
+				return Boss::log( bus, Info
+						, "XRebalancePredictor: "
+						  "disabled (%s = 0)."
+						, o.name.c_str());
+			return Boss::log( bus, Info
+					, "XRebalancePredictor: horizon "
+					  "cap = %" PRIu64 " seconds."
+					, horizon_max_secs);
+		}
+		if (o.name == opt_horizon_frac) {
+			auto v = double(0.0);
+			if (!parse_double(o.value, v) || v <= 0)
+				return bad_option(o.name);
+			horizon_frac = v;
+			return Boss::log( bus, Info
+					, "XRebalancePredictor: horizon "
+					  "frac = %.3f."
+					, horizon_frac);
+		}
+		if (o.name == opt_min_samples) {
+			auto v = std::uint64_t(0);
+			if (!parse_u64(o.value, v) || v < 1)
+				return bad_option(o.name);
+			min_samples = v;
+			return Boss::log( bus, Info
+					, "XRebalancePredictor: min "
+					  "samples = %" PRIu64 "."
+					, min_samples);
+		}
+		if (o.name == opt_wall_margin) {
+			auto v = double(0.0);
+			if (!parse_double(o.value, v) || v <= 0)
+				return bad_option(o.name);
+			wall_margin = v;
+			return Boss::log( bus, Info
+					, "XRebalancePredictor: wall "
+					  "margin = %.3f."
+					, wall_margin);
+		}
+		if (o.name == opt_floor_factor) {
+			auto v = double(0.0);
+			if (!parse_double(o.value, v) || v < 0)
+				return bad_option(o.name);
+			floor_factor = v;
+			if (floor_factor == 0)
+				return Boss::log( bus, Info
+						, "XRebalancePredictor: "
+						  "floors disabled "
+						  "(walls-only).");
+			return Boss::log( bus, Info
+					, "XRebalancePredictor: floor "
+					  "factor = %.3f."
+					, floor_factor);
+		}
+		return Ev::lift();
+	}
+
+	Ev::Io<void> bad_option(std::string const& name) {
+		return Boss::log( bus, Warn
+				, "XRebalancePredictor: %s: could not "
+				  "parse value; keeping current setting."
+				, name.c_str());
+	}
+
+	Ev::Io<void> run(std::uint64_t cutoff) {
+		return mode_proxy.get_mode().then([this, cutoff
+						  ](RebalanceMode m) {
+			if (m != RebalanceMode::xrebalance)
+				return Boss::log( bus, Debug
+						, "XRebalancePredictor: "
+						  "mode is not xrebalance; "
+						  "skipping cycle.");
+			return evaluate(cutoff);
+		});
+	}
+
+	Ev::Io<void> evaluate(std::uint64_t cutoff) {
+		auto rows = std::make_shared<std::vector<Row>>();
+		return db.transact().then([rows](Sqlite3::Tx tx) {
+			auto fetch = tx.query(R"QRY(
+			SELECT time, scid, dir, kind, amount_msat
+			  FROM "XRebalanceHistory"
+			 ORDER BY scid, dir, time;
+			)QRY").execute();
+			for (auto& r : fetch) {
+				auto ndx = std::size_t(0);
+				auto row = Row();
+				row.time = r.get<std::uint64_t>(ndx++);
+				row.scid = r.get<std::string>(ndx++);
+				row.dir = r.get<std::uint32_t>(ndx++);
+				row.kind = r.get<std::string>(ndx++);
+				row.amount_msat =
+					r.get<std::uint64_t>(ndx++);
+				rows->push_back(std::move(row));
+			}
+			tx.commit();
+			return Ev::lift();
+		}).then([this, rows, cutoff]() {
+			auto params = XRebalancePredict::Params{
+				horizon_frac, horizon_max_secs,
+				std::size_t(min_samples), wall_margin,
+				floor_factor};
+			auto now = std::uint64_t(get_now());
+			auto result = plan(*rows, cutoff, now, params);
+
+			last_run_time = now;
+			last_directions = result.directions;
+			last_candidates = result.candidates;
+			last_walls = 0;
+			last_floors = 0;
+
+			auto act = Ev::lift();
+			for (auto const& a : result.assertions) {
+				if (a.is_wall) {
+					++last_walls;
+					act = std::move(act)
+					    + Boss::Mod::AskreneLayer::
+						inform_channel_constrained(
+						*rpc,
+						Boss::Mod::AskreneLayer::
+						    xrebalance_layer_name,
+						Ln::Scid(a.scid), a.dir,
+						Ln::Amount::msat(
+						    a.amount_msat));
+				} else {
+					++last_floors;
+					act = std::move(act)
+					    + Boss::Mod::AskreneLayer::
+						inform_channel_unconstrained(
+						*rpc,
+						Boss::Mod::AskreneLayer::
+						    xrebalance_layer_name,
+						Ln::Scid(a.scid), a.dir,
+						Ln::Amount::msat(
+						    a.amount_msat));
+				}
+			}
+			if (result.assertions.empty())
+				act = std::move(act)
+				    + Boss::log( bus, Debug
+					, "XRebalancePredictor: nothing "
+					  "to assert (%zu directions, "
+					  "%zu candidates)."
+					, result.directions
+					, result.candidates);
+			else
+				act = std::move(act)
+				    + Boss::log( bus, Info
+					, "XRebalancePredictor: asserted "
+					  "%zu wall(s), %zu floor(s) "
+					  "(%zu directions, %zu "
+					  "candidates)."
+					, last_walls, last_floors
+					, result.directions
+					, result.candidates);
+			return act;
+		});
+	}
+
+	Ev::Io<void> status() {
+		auto out = Json::Out();
+		auto obj = out.start_object();
+		obj
+			.field("enabled", horizon_max_secs != 0)
+			.field("horizon_max_secs", horizon_max_secs)
+			.field("horizon_frac", horizon_frac)
+			.field("min_samples", min_samples)
+			.field("wall_margin", wall_margin)
+			.field("floor_factor", floor_factor)
+			.field("last_run_time", last_run_time)
+			.field( "last_directions"
+			      , std::uint64_t(last_directions))
+			.field( "last_candidates"
+			      , std::uint64_t(last_candidates))
+			.field("last_walls", std::uint64_t(last_walls))
+			.field("last_floors", std::uint64_t(last_floors))
+			;
+		obj.end_object();
+		return bus.raise(Msg::ProvideStatus{
+			"xrebalance_predictor",
+			std::move(out)
+		});
+	}
+
+public:
+	Impl() =delete;
+	Impl(Impl&&) =delete;
+	Impl(Impl const&) =delete;
+
+	explicit
+	Impl(S::Bus& bus_, std::function<double()> get_now_)
+		: bus(bus_)
+		, get_now(std::move(get_now_))
+		, rpc(nullptr)
+		, mode_proxy(bus_)
+		, horizon_max_secs(default_horizon_max)
+		, horizon_frac(default_horizon_frac)
+		, min_samples(default_min_samples)
+		, wall_margin(default_wall_margin)
+		, floor_factor(default_floor_factor)
+		, last_run_time(0)
+		, last_directions(0)
+		, last_candidates(0)
+		, last_walls(0)
+		, last_floors(0) { start(); }
+};
+
+XRebalancePredictor::XRebalancePredictor(XRebalancePredictor&&) =default;
+XRebalancePredictor::~XRebalancePredictor() =default;
+
+XRebalancePredictor::XRebalancePredictor( S::Bus& bus
+					, std::function<double()> get_now_
+					)
+	: pimpl(Util::make_unique<Impl>(bus, get_now_)) { }
+
+}}

Boss/Mod/XRebalancePredictor.hpp

@@ -0,0 +1,99 @@
+#ifndef BOSS_MOD_XREBALANCEPREDICTOR_HPP
+#define BOSS_MOD_XREBALANCEPREDICTOR_HPP
+
+#include"Boss/Mod/XRebalancePredict.hpp"
+#include"Ev/now.hpp"
+#include<cstddef>
+#include<cstdint>
+#include<functional>
+#include<memory>
+#include<string>
+#include<vector>
+
+namespace S { class Bus; }
+
+namespace Boss { namespace Mod {
+
+/** class Boss::Mod::XRebalancePredictor
+ *
+ * @brief The live persistence forecaster (phase 2 of the
+ * history+prediction design).
+ *
+ * @desc After each hourly aging pass over the clboss-xrebalance
+ * layer (Boss::Msg::XRebalanceLayerAged), reads the
+ * XRebalanceHistory observation store, runs the pure
+ * Boss::Mod::XRebalancePredict algorithm per channel direction,
+ * and re-asserts the surviving walls/floors into the routed layer
+ * via askrene-inform-channel -- but ONLY for directions whose
+ * newest real observation predates the aging cutoff (directions
+ * with live evidence need no synthesis).  Synthetic assertions are
+ * NEVER recorded back into the observation store (history purity;
+ * no self-confirmation).
+ *
+ * OFF BY DEFAULT: the master switch is the dynamic option
+ * clboss-xrebalance-predict-horizon-max-secs (0 = disabled), so a
+ * build carrying this module changes nothing until the operator
+ * flips it via setconfig.  Also dormant unless
+ * clboss-rebalance-mode is xrebalance.  All constants are dynamic
+ * options, mirroring the per-query override parameters of the
+ * read-only clboss-xrebalance-history / -predictions spot-check
+ * commands.  See DEVSTATE/XREBALANCE-HISTORY-PREDICT-2026-06-10.org.
+ */
+class XRebalancePredictor {
+public:
+	/* One observation-store row; input to plan().  */
+	struct Row {
+		std::string scid;
+		std::uint32_t dir;
+		std::uint64_t time;
+		std::string kind;
+		std::uint64_t amount_msat;
+	};
+	/* One synthetic inform the live predictor would issue.  */
+	struct Assertion {
+		std::string scid;
+		std::uint32_t dir;
+		/* true: inform constrained (wall);
+		 * false: inform unconstrained (floor).  */
+		bool is_wall;
+		std::uint64_t amount_msat;
+	};
+	struct Plan {
+		std::vector<Assertion> assertions;
+		/* Channel directions present in the store.  */
+		std::size_t directions;
+		/* Directions without live evidence (newest real
+		 * observation older than the aging cutoff) -- the
+		 * candidate set.  */
+		std::size_t candidates;
+	};
+
+	/* The full per-cycle decision, as a pure function (exposed
+	 * for unit testing): group rows per direction, gate on
+	 * candidacy, run XRebalancePredict::predict, collect the
+	 * asserting sides.  `rows` in any order.  */
+	static Plan plan( std::vector<Row> const& rows
+			, std::uint64_t cutoff
+			, std::uint64_t now
+			, XRebalancePredict::Params const& params
+			);
+
+private:
+	class Impl;
+	std::unique_ptr<Impl> pimpl;
+
+public:
+	XRebalancePredictor() =delete;
+
+	XRebalancePredictor(XRebalancePredictor&&);
+	~XRebalancePredictor();
+
+	explicit
+	XRebalancePredictor( S::Bus& bus
+			   , std::function<double()> get_now_ = &Ev::now
+			   );
+};
+
+}}
+
+#endif /* !defined(BOSS_MOD_XREBALANCEPREDICTOR_HPP) */

Boss/Mod/all.cpp

@@ -77,6 +77,7 @@
 #include"Boss/Mod/Waiter.hpp"
 #include"Boss/Mod/XMoveFunds/Main.hpp"
 #include"Boss/Mod/XRebalanceHistory.hpp"
+#include"Boss/Mod/XRebalancePredictor.hpp"
 #include"Boss/Mod/XRebalancer.hpp"
 #include"Boss/Mod/all.hpp"
 #include<vector>
@@ -215,6 +216,7 @@ std::shared_ptr<void> all( std::ostream& cout
 	all->install<FundsMover::Main>(bus);
 	all->install<XMoveFunds::Main>(bus);
 	all->install<XRebalanceHistory>(bus);
+	all->install<XRebalancePredictor>(bus);
 	all->install<XRebalancer>(bus, *waiter);
 	all->install<MoveFundsCommand>(bus);
 	all->install<EarningsTracker>(bus);

Boss/Msg/XRebalanceLayerAged.hpp

@@ -0,0 +1,32 @@
+#ifndef BOSS_MSG_XREBALANCELAYERAGED_HPP
+#define BOSS_MSG_XREBALANCELAYERAGED_HPP
+
+#include<cstdint>
+
+namespace Boss { namespace Msg {
+
+/** struct Boss::Msg::XRebalanceLayerAged
+ *
+ * @brief Raised by Boss::Mod::XMoveFunds after its hourly
+ * `askrene-age` pass over the clboss-xrebalance layer (whether or
+ * not the RPC succeeded; on failure the layer merely retains stale
+ * entries, which is safe for subscribers).
+ *
+ * @desc This is the "end of expiration cycle" hook: constraints
+ * older than `cutoff` have just been trimmed from the routed layer,
+ * so any channel direction whose newest real observation is older
+ * than `cutoff` now has NO live evidence -- exactly the set the
+ * persistence forecaster (Boss::Mod::XRebalancePredictor) considers
+ * for synthetic re-assertion.
+ */
+struct XRebalanceLayerAged {
+	/* When the aging pass ran (unix seconds).  */
+	std::uint64_t time;
+	/* The cutoff passed to askrene-age: constraints with
+	 * timestamp < cutoff were removed.  */
+	std::uint64_t cutoff;
+};
+
+}}
+
+#endif /* !defined(BOSS_MSG_XREBALANCELAYERAGED_HPP) */

Makefile.am

@@ -284,6 +284,8 @@ libclboss_la_SOURCES = \
 	Boss/Mod/XRebalanceHistory.hpp \
 	Boss/Mod/XRebalancePredict.cpp \
 	Boss/Mod/XRebalancePredict.hpp \
+	Boss/Mod/XRebalancePredictor.cpp \
+	Boss/Mod/XRebalancePredictor.hpp \
 	Boss/Mod/XRebalancer.cpp \
 	Boss/Mod/XRebalancer.hpp \
 	Boss/Mod/all.cpp \
@@ -408,6 +410,7 @@ libclboss_la_SOURCES = \
 	Boss/Msg/TimerRandomHourly.hpp \
 	Boss/Msg/TimerTwiceDaily.hpp \
 	Boss/Msg/XRebalanceAttribution.hpp \
+	Boss/Msg/XRebalanceLayerAged.hpp \
 	Boss/Msg/XRebalanceObservation.hpp \
 	Boss/RebalanceMode.hpp \
 	Boss/Shutdown.hpp \
@@ -644,6 +647,7 @@ TESTS = \
 	tests/boss/test_earningshistory \
 	tests/boss/test_xrebalancehistory \
 	tests/boss/test_xrebalancepredict \
+	tests/boss/test_xrebalancepredictor \
 	tests/boss/test_peerjudge_algo \
 	tests/boss/test_peerjudge_datagatherer \
 	tests/boss/test_peerstatistician \

tests/boss/test_xrebalancepredictor.cpp

@@ -0,0 +1,97 @@
+#undef NDEBUG
+
+#include"Boss/Mod/XRebalancePredictor.hpp"
+
+#include<cassert>
+
+using Boss::Mod::XRebalancePredictor;
+using Row = Boss::Mod::XRebalancePredictor::Row;
+
+namespace {
+
+/* Walls-only live defaults (the planned first enablement), with the
+ * master switch open.  */
+auto const walls_only = Boss::Mod::XRebalancePredict::Params{
+	2.0, 86400, 2, 1.0, 0.0};
+
+}
+
+int main() {
+	auto rows = std::vector<Row>{
+		/* Stale wall regime: 2 failures spanning 1800s.  */
+		{"100x1x0", 0, 1000, "liquidity_fail", 7000},
+		{"100x1x0", 0, 2800, "liquidity_fail", 6000},
+		/* Fresh data: not a candidate regardless of regime.  */
+		{"200x2x0", 1, 4500, "liquidity_fail", 9000},
+		{"200x2x0", 1, 3000, "liquidity_fail", 9500},
+		/* Stale floor regime: 2 successes spanning 1800s.  */
+		{"300x3x0", 0, 1000, "success", 10000},
+		{"300x3x0", 0, 2800, "success", 12000},
+		/* node_fail only: a candidate, but not a liquidity
+		 * bound -- nothing to assert.  */
+		{"400x4x0", 0, 500, "node_fail", 1},
+	};
+	auto const cutoff = std::uint64_t(4000);
+	/* data age for the stale regimes: 3000s <= horizon 3600s.  */
+	auto const now = std::uint64_t(2800 + 3000);
+
+	/* Walls-only: exactly the 100x1x0 wall at the tightest
+	 * failure bound.  */
+	{
+		auto plan = XRebalancePredictor::plan(
+			rows, cutoff, now, walls_only);
+		assert(plan.directions == 4);
+		assert(plan.candidates == 3);
+		assert(plan.assertions.size() == 1);
+		auto const& a = plan.assertions[0];
+		assert(a.scid == "100x1x0");
+		assert(a.dir == 0);
+		assert(a.is_wall);
+		assert(a.amount_msat == 6000);
+	}
+
+	/* Enabling floors adds the 300x3x0 floor (12000 * 0.9).  */
+	{
+		auto p = walls_only;
+		p.floor_factor = 0.9;
+		auto plan = XRebalancePredictor::plan(
+			rows, cutoff, now, p);
+		assert(plan.assertions.size() == 2);
+		assert(!plan.assertions[1].is_wall);
+		assert(plan.assertions[1].scid == "300x3x0");
+		assert(plan.assertions[1].amount_msat == 10800);
+	}
+
+	/* Wall margin scales the asserted amount.  */
+	{
+		auto p = walls_only;
+		p.wall_margin = 1.5;
+		auto plan = XRebalancePredictor::plan(
+			rows, cutoff, now, p);
+		assert(plan.assertions.size() == 1);
+		assert(plan.assertions[0].amount_msat == 9000);
+	}
+
+	/* Master switch: horizon cap 0 means every regime is past
+	 * its horizon; nothing asserts (the live module also skips
+	 * the cycle entirely).  */
+	{
+		auto p = walls_only;
+		p.horizon_max_secs = 0;
+		auto plan = XRebalancePredictor::plan(
+			rows, cutoff, now, p);
+		assert(plan.assertions.size() == 0);
+		assert(plan.candidates == 3);
+	}
+
+	/* Later `now`: the same regimes go stale past their horizon
+	 * (3600s) and assert nothing.  */
+	{
+		auto plan = XRebalancePredictor::plan(
+			rows, cutoff, std::uint64_t(2800 + 3601),
+			walls_only);
+		assert(plan.assertions.size() == 0);
+	}
+
+	return 0;
+}