loop.hpp

#pragma once
 
#include "./clock.hpp"
#include "./execution.hpp"
#include "./stop.hpp"
 
#include <algotradelib/core>
#include <algotradelib/strategy>
 
#include <algorithm>
#include <array>
#include <chrono>
#include <cstddef>
#include <optional>
#include <string>
#include <string_view>
#include <type_traits>
#include <utility>
#include <vector>
 
namespace algotradelib {
namespace runner {

enum class PollKind {
  BARS,
  QUOTES,
  HOLDINGS,
  WORKING_ORDERS,
  BUYING_POWER,
  CALENDAR,
  PRICE_LIMITS,
};

inline constexpr std::string_view poll_kind_name(PollKind kind) noexcept {
  switch (kind) {
    case PollKind::CALENDAR:
      return "calendar";
    case PollKind::BARS:
      return "bars";
    case PollKind::QUOTES:
      return "quotes";
    case PollKind::HOLDINGS:
      return "holdings";
    case PollKind::WORKING_ORDERS:
      return "working_orders";
    case PollKind::BUYING_POWER:
      return "buying_power";
    case PollKind::PRICE_LIMITS:
      return "price_limits";
  }
  return "";
}

inline constexpr PollKind all_poll_kinds[] = {
  // First: the phase derived from it decides whether anything else in this iteration may be sent.
  PollKind::CALENDAR,
  // Second, before every poll that can feed, so this iteration's orders use this iteration's bands.
  PollKind::PRICE_LIMITS,
  // The listing, then holdings, before anything that feeds. A fill between the two reads then
  // over-counts the position (harmless) rather than under-counting it (which would buy again).
  PollKind::WORKING_ORDERS,
  PollKind::HOLDINGS,
  // After holdings, so money and positions are equally fresh.
  PollKind::BUYING_POWER,
  PollKind::BARS,
  PollKind::QUOTES,
};
 
static_assert(
  lists_every_value(all_poll_kinds, [](PollKind kind) { return !poll_kind_name(kind).empty(); }),
  "every PollKind must appear in all_poll_kinds once: a kind left out is never polled, and the loop "
  "would silently answer with whatever the previous poll left behind"
);

inline constexpr std::size_t poll_kind_count = std::size(all_poll_kinds);

enum class PollOutcome {
  UPDATED,
  UNCHANGED,
  FAILED,
  RATE_LIMITED,
};

inline constexpr std::array<PollOutcome, 4> all_poll_outcomes{
  PollOutcome::UPDATED,
  PollOutcome::UNCHANGED,
  PollOutcome::FAILED,
  PollOutcome::RATE_LIMITED,
};

inline constexpr std::string_view poll_outcome_name(PollOutcome outcome) noexcept {
  switch (outcome) {
    case PollOutcome::UPDATED:
      return "updated";
    case PollOutcome::UNCHANGED:
      return "unchanged";
    case PollOutcome::FAILED:
      return "failed";
    case PollOutcome::RATE_LIMITED:
      return "rate_limited";
  }
  return "";
}
 
static_assert(
  lists_every_value(all_poll_outcomes, [](PollOutcome outcome) { return !poll_outcome_name(outcome).empty(); }),
  "all_poll_outcomes must list every PollOutcome"
);

struct PollSchedule {
  std::optional<std::chrono::nanoseconds> bars_;
  std::optional<std::chrono::nanoseconds> quotes_;
  std::optional<std::chrono::nanoseconds> holdings_;
  std::optional<std::chrono::nanoseconds> working_orders_;
  std::optional<std::chrono::nanoseconds> buying_power_;
  std::chrono::nanoseconds tick_{std::chrono::seconds(1)};
  std::optional<std::chrono::nanoseconds> calendar_{std::chrono::hours(6)};
  std::optional<std::chrono::nanoseconds> price_limits_;
  std::optional<std::chrono::nanoseconds> interval_for(PollKind kind) const noexcept {
    switch (kind) {
      case PollKind::CALENDAR:
        return calendar_;
      case PollKind::BARS:
        return bars_;
      case PollKind::QUOTES:
        return quotes_;
      case PollKind::HOLDINGS:
        return holdings_;
      case PollKind::WORKING_ORDERS:
        return working_orders_;
      case PollKind::BUYING_POWER:
        return buying_power_;
      case PollKind::PRICE_LIMITS:
        return price_limits_;
    }
    return std::nullopt;
  }
};

template <typename Source, typename = void>
struct source_provides_working_orders : std::false_type {};
 
template <typename Source>
struct source_provides_working_orders<
  Source, std::void_t<decltype(Source::provides_working_orders)>
> : std::bool_constant<Source::provides_working_orders> {};

template <typename Runner, typename = void>
struct runner_keeps_sent_orders : std::true_type {};
 
template <typename Runner>
struct runner_keeps_sent_orders<
  Runner, std::void_t<decltype(Runner::keeps_sent_orders_until_listed)>
> : std::bool_constant<Runner::keeps_sent_orders_until_listed> {};

template <typename Runner, typename = void>
struct runner_counts_kept_drops : std::false_type {};
 
template <typename Runner>
struct runner_counts_kept_drops<
  Runner, std::void_t<decltype(std::declval<const Runner&>().kept_intents_dropped())>
> : std::true_type {};

template <typename Runner, typename = void>
struct runner_counts_warmup : std::false_type {};
 
template <typename Runner>
struct runner_counts_warmup<
  Runner, std::void_t<decltype(std::declval<const Runner&>().warmup_instructions_discarded())>
> : std::true_type {};

struct ChargedCosts {
  struct StatedTotal {
    Money amount_;
    std::size_t statements_{0};
  };

  std::vector<StatedTotal> commission_;
  std::vector<StatedTotal> tax_;
  std::size_t orders_stating_cost_{0};
  std::size_t orders_stating_none_{0};
  std::size_t working_list_disappearances_{0};
  std::size_t conflicting_restatements_{0};
  std::size_t unrepresentable_sums_{0};

  std::optional<Decimal64> commission_in(const Currency& currency) const {
    return figure_in(commission_, currency);
  }

  std::optional<Decimal64> tax_in(const Currency& currency) const {
    return figure_in(tax_, currency);
  }

  std::size_t commission_statements_in(const Currency& currency) const {
    return statements_in(commission_, currency);
  }

  std::size_t tax_statements_in(const Currency& currency) const {
    return statements_in(tax_, currency);
  }

  bool add(std::vector<StatedTotal>& totals, const Money& amount) {
    for (auto& held : totals) {
      if (held.amount_.currency_ == amount.currency_) {
        const auto sum = decimal_add(held.amount_.amount_, amount.amount_);
        if (!sum) {
          ++unrepresentable_sums_;
          return false;
        }
        held.amount_.amount_ = *sum;
        ++held.statements_;
        return true;
      }
    }
    totals.push_back(StatedTotal{amount, 1});
    return true;
  }

  static std::optional<Decimal64> figure_in(
    const std::vector<StatedTotal>& totals,
    const Currency& currency
  ) {
    for (const auto& held : totals) {
      if (held.amount_.currency_ == currency) {
        return held.amount_.amount_;
      }
    }
    return std::nullopt;
  }

  static std::size_t statements_in(
    const std::vector<StatedTotal>& totals,
    const Currency& currency
  ) {
    for (const auto& held : totals) {
      if (held.amount_.currency_ == currency) {
        return held.statements_;
      }
    }
    return 0;
  }

  ChargedCosts since(const ChargedCosts& before) const {
    ChargedCosts delta;
    std::size_t refused = 0;
    const auto difference = [&refused](
      const std::vector<StatedTotal>& now,
      const std::vector<StatedTotal>& then
    ) {
      std::vector<StatedTotal> out;
      for (const auto& held : now) {
        auto earlier = Decimal64::parse("0");
        std::size_t before_statements = 0;
        for (const auto& was : then) {
          if (was.amount_.currency_ == held.amount_.currency_) {
            earlier = was.amount_.amount_;
            before_statements = was.statements_;
            break;
          }
        }
        // Whether anything was stated is the statement count, not the money.
        if (held.statements_ == before_statements) {
          continue;
        }
        const auto moved = decimal_subtract(held.amount_.amount_, earlier);
        if (!moved) {
          // Counted, not skipped: something was stated and its amount cannot be represented.
          ++refused;
          continue;
        }
        out.push_back(
          StatedTotal{Money{*moved, held.amount_.currency_}, held.statements_ - before_statements}
        );
      }
      return out;
    };
    delta.commission_ = difference(commission_, before.commission_);
    delta.tax_ = difference(tax_, before.tax_);
    delta.orders_stating_cost_ = orders_stating_cost_ - before.orders_stating_cost_;
    delta.orders_stating_none_ = orders_stating_none_ - before.orders_stating_none_;
    delta.working_list_disappearances_ =
      working_list_disappearances_ - before.working_list_disappearances_;
    delta.conflicting_restatements_ = conflicting_restatements_ - before.conflicting_restatements_;
    delta.unrepresentable_sums_ =
      (unrepresentable_sums_ - before.unrepresentable_sums_) + refused;
    return delta;
  }
};

template <typename Source, typename = void>
struct source_states_charged_costs : std::false_type {};
 
template <typename Source>
struct source_states_charged_costs<
  Source, std::void_t<decltype(std::declval<const Source&>().charged_costs())>
> : std::true_type {};

template <typename Source, typename = void>
struct source_names_buying_power_currencies : std::false_type {};
 
template <typename Source>
struct source_names_buying_power_currencies<
  Source, std::void_t<decltype(std::declval<const Source&>().buying_power_currencies())>
> : std::true_type {};

struct FailurePolicy {
  std::size_t consecutive_limit_{5};
};

struct IterationReport {
  std::chrono::system_clock::time_point at_{};
  std::vector<std::pair<PollKind, PollOutcome>> polls_;
  std::optional<strategy::HaltReason> gate_;
  bool fed_{false};
  strategy::MarketPhase phase_{strategy::MarketPhase::UNKNOWN};
  bool phase_changed_{false};

  bool updated() const noexcept {
    for (const auto& entry : polls_) {
      if (entry.second == PollOutcome::UPDATED) {
        return true;
      }
    }
    return false;
  }
};

struct LoopReport {
  std::size_t iterations_{0};
  ChargedCosts charged_costs_;
  std::size_t attempted_[poll_kind_count] = {};
  std::size_t failed_[poll_kind_count] = {};
  std::size_t blocked_{0};
  std::optional<StopIntent> stopped_by_;
  std::optional<strategy::HaltReason> final_gate_;
  strategy::MarketPhase final_phase_{strategy::MarketPhase::UNKNOWN};
  std::size_t sends_under_a_revised_phase_{0};
  std::size_t price_bands_withheld_{0};
  std::size_t unclamped_orders_rejected_{0};
  std::size_t unclamped_orders_refused_{0};
  std::size_t orders_unbuildable_{0};
  std::size_t orders_reconcile_withheld_{0};
  std::size_t orders_guard_refused_{0};
  std::size_t orders_binding_refused_{0};
  std::size_t orders_broker_refused_{0};
  std::size_t orders_indeterminate_{0};
  std::size_t orders_unconfirmed_timed_out_{0};
  std::size_t plans_withheld_{0};
  std::size_t instructions_unavailable_{0};
  std::size_t cancels_rejected_{0};
  std::size_t cancels_refused_{0};
  std::size_t cancels_unresolved_{0};
  std::size_t cancels_indeterminate_{0};
  std::optional<BrokerRefusal> last_broker_refusal_;
  std::optional<strategy::Rejection> last_unbuildable_;
  std::size_t phase_transitions_{0};
  std::size_t sends_outside_permitted_phase_{0};
  bool abandoned_{false};
  std::optional<PollKind> abandoned_kind_;
  bool order_list_read_{false};
  std::size_t withdrawn_{0};
  std::size_t flattened_{0};
  std::size_t closing_placed_{0};
  std::size_t closing_rejected_{0};
  std::size_t closing_refused_{0};
  std::size_t closing_withheld_{0};
  std::size_t closing_indeterminate_{0};
  std::size_t closing_unavailable_{0};
  std::size_t closing_unresolved_{0};
  std::size_t closing_unconfirmed_{0};
  std::chrono::nanoseconds closing_waited_{0};
  bool closing_wait_expired_{false};
  bool closing_interrupted_{false};
  std::optional<Obligation> owed_stop_;

  std::size_t closing_contended_{0};

  std::size_t closing_unsizable_{0};

  bool closing_source_unverified_{false};

  bool closing_complete() const noexcept {
    return closing_rejected_ == 0 && closing_refused_ == 0 && closing_withheld_ == 0 &&
      closing_indeterminate_ == 0 && closing_unavailable_ == 0 && closing_unresolved_ == 0 &&
      closing_contended_ == 0 && closing_unsizable_ == 0 && !closing_source_unverified_ &&
      closing_unconfirmed_ == 0 && !closing_interrupted_;
  }

  std::size_t stop_handler_refused_{0};
  std::size_t stop_handler_sent_{0};

  std::vector<std::string> notes_;

  std::size_t attempted(PollKind kind) const noexcept {
    return attempted_[static_cast<int>(kind)];
  }

  std::size_t failed(PollKind kind) const noexcept { return failed_[static_cast<int>(kind)]; }
};

template <typename Runner>
class ManualSource {
public:
  static constexpr bool provides_working_orders = true;

  void queue(PollKind kind, PollOutcome outcome) {
    queued_[static_cast<int>(kind)].push_back(outcome);
  }

  void set_default(PollKind kind, PollOutcome outcome) {
    defaults_[static_cast<int>(kind)] = outcome;
  }

  void set_order_list(PollOutcome outcome) { order_list_ = outcome; }

  void set_buying_power_currencies(std::vector<Currency> currencies) {
    currencies_ = std::move(currencies);
  }

  void set_calendar(TradingCalendar calendar) {
    calendar_ = std::move(calendar);
    defaults_[static_cast<int>(PollKind::CALENDAR)] = PollOutcome::UPDATED;
  }

  const std::vector<Currency>& buying_power_currencies() const noexcept { return currencies_; }

  std::size_t order_list_reads() const noexcept { return order_list_reads_; }

  std::size_t polls(PollKind kind) const noexcept { return polls_[static_cast<int>(kind)]; }

  void set_price_limits(InstrumentId instrument, PriceLimits limits) {
    price_limits_.emplace_back(std::move(instrument), std::move(limits));
    defaults_[static_cast<int>(PollKind::PRICE_LIMITS)] = PollOutcome::UPDATED;
  }

  PollOutcome poll(PollKind kind, Runner& runner_instance) {
    const int at = static_cast<int>(kind);
    ++polls_[at];
    auto outcome = defaults_[at];
    if (!queued_[at].empty()) {
      outcome = queued_[at].front();
      queued_[at].erase(queued_[at].begin());
    }
    // The one kind whose answer goes somewhere: the loop reads the calendar back out of the context,
    // so a source that did not put it there would leave every phase assertion in this header testing
    // an empty calendar.
    if (kind == PollKind::CALENDAR && calendar_ && outcome == PollOutcome::UPDATED) {
      runner_instance.context().set_calendar(*calendar_);
    }
    // The second kind whose answer goes somewhere -- onto the runner rather than into the context,
    // which is where reconciliation reads a band from. Installed on every successful poll, as a real
    // source does, so a test can age one by letting the loop's clock run.
    if (kind == PollKind::PRICE_LIMITS && outcome == PollOutcome::UPDATED) {
      for (const auto& band : price_limits_) {
        runner_instance.set_price_limits(band.first, band.second);
      }
    }
    return outcome;
  }

  PollOutcome read_order_list(Runner&) {
    ++order_list_reads_;
    return order_list_;
  }
 
private:
  // Every kind's default outcome, filled rather than listed.
  //
  // It used to be a list with one entry per `PollKind`, and `PollOutcome::UPDATED` is the zero value
  // -- so adding an enumerator silently defaulted the new kind to "new data arrived" in every test
  // that had not set it. Filling cannot be one short.
  static std::array<PollOutcome, poll_kind_count> unchanged_defaults() {
    std::array<PollOutcome, poll_kind_count> filled{};
    filled.fill(PollOutcome::UNCHANGED);
    return filled;
  }
 
  std::vector<PollOutcome> queued_[poll_kind_count];
  std::array<PollOutcome, poll_kind_count> defaults_{unchanged_defaults()};
  PollOutcome order_list_{PollOutcome::UPDATED};
  std::vector<Currency> currencies_;
  std::optional<TradingCalendar> calendar_;
  std::vector<std::pair<InstrumentId, PriceLimits>> price_limits_;
  std::size_t polls_[poll_kind_count] = {};
  std::size_t order_list_reads_{0};
};

template <typename Runner, typename Source, typename Clock = SteadyClock>
class PollingLoop {
public:
  PollingLoop(Runner& runner, Source& source, PollSchedule schedule = {}, Clock clock = {})
  : runner_(&runner), source_(&source), schedule_(std::move(schedule)), clock_(std::move(clock)) {}

  const Clock& clock() const noexcept { return clock_; }
  Clock& clock() noexcept { return clock_; }

  const PollSchedule& schedule() const noexcept { return schedule_; }
 
private:
  // Every monotone instruction-outcome figure the runner holds, read at one instant.
  //
  // Twelve of them, read together rather than one call at a time, because they are differenced
  // against each other's snapshot: a set taken at two instants describes one interval, and twelve
  // separately captured values do not.
  struct InstructionOutcomes {
    std::size_t unbuildable_{0};
    std::size_t unconfirmed_timed_out_{0};
    std::size_t reconcile_withheld_{0};
    std::size_t guard_refused_{0};
    std::size_t binding_refused_{0};
    std::size_t broker_refused_{0};
    std::size_t indeterminate_{0};
    std::size_t cancels_rejected_{0};
    std::size_t cancels_refused_{0};
    std::size_t cancels_unresolved_{0};
    std::size_t cancels_indeterminate_{0};
    std::size_t plans_withheld_{0};
    std::size_t instructions_unavailable_{0};
  };
 
  // Reads them all from the runner.
  InstructionOutcomes instruction_outcomes() const {
    InstructionOutcomes at;
    at.unbuildable_ = runner_->orders_unbuildable();
    at.unconfirmed_timed_out_ = runner_->orders_unconfirmed_timed_out();
    at.reconcile_withheld_ = runner_->orders_reconcile_withheld();
    at.guard_refused_ = runner_->orders_guard_refused();
    at.binding_refused_ = runner_->orders_binding_refused();
    at.broker_refused_ = runner_->orders_broker_refused();
    at.indeterminate_ = runner_->orders_indeterminate();
    at.cancels_rejected_ = runner_->cancels_rejected();
    at.cancels_refused_ = runner_->cancels_refused();
    at.cancels_unresolved_ = runner_->cancels_unresolved();
    at.cancels_indeterminate_ = runner_->cancels_indeterminate();
    at.plans_withheld_ = runner_->plans_withheld();
    at.instructions_unavailable_ = runner_->instructions_unavailable();
    return at;
  }
 
  // Writes the difference between two snapshots onto `report`.
  static void note_instruction_outcomes(
    LoopReport& report,
    const InstructionOutcomes& before,
    const InstructionOutcomes& after
  ) {
    report.orders_unbuildable_ = after.unbuildable_ - before.unbuildable_;
    report.orders_unconfirmed_timed_out_ = after.unconfirmed_timed_out_ - before.unconfirmed_timed_out_;
    report.orders_reconcile_withheld_ = after.reconcile_withheld_ - before.reconcile_withheld_;
    report.orders_guard_refused_ = after.guard_refused_ - before.guard_refused_;
    report.orders_binding_refused_ = after.binding_refused_ - before.binding_refused_;
    report.orders_broker_refused_ = after.broker_refused_ - before.broker_refused_;
    report.orders_indeterminate_ = after.indeterminate_ - before.indeterminate_;
    report.cancels_rejected_ = after.cancels_rejected_ - before.cancels_rejected_;
    report.cancels_refused_ = after.cancels_refused_ - before.cancels_refused_;
    report.cancels_unresolved_ = after.cancels_unresolved_ - before.cancels_unresolved_;
    report.cancels_indeterminate_ = after.cancels_indeterminate_ - before.cancels_indeterminate_;
    report.plans_withheld_ = after.plans_withheld_ - before.plans_withheld_;
    report.instructions_unavailable_ = after.instructions_unavailable_ - before.instructions_unavailable_;
  }
 
public:

  void set_maximum_portfolio_age(std::chrono::nanoseconds age) {
    maximum_portfolio_age_ = age;
    // Pushed down as the **policy**, not as a verdict. The runner re-evaluates it against whatever
    // snapshot is in hand when it settles, which is the only version that cannot describe a
    // portfolio that has since been replaced -- a source callback can feed, and settle, between this
    // loop's own checks.
    runner_->set_maximum_portfolio_age(age);
  }

  void set_closing_wait(std::chrono::nanoseconds limit) { closing_wait_ = limit; }

  void set_owed_order_limit(std::chrono::nanoseconds limit) { owed_order_limit_ = limit; }

  void allow_unbounded_portfolio_age() {
    allow_unbounded_portfolio_age_ = true;
    maximum_portfolio_age_.reset();
    runner_->set_maximum_portfolio_age(std::nullopt);
  }

  void set_maximum_price_limits_age(std::optional<std::chrono::nanoseconds> age) {
    runner_->set_maximum_price_limits_age(age);
  }

  void allow_unbounded_price_limits_age() {
    runner_->allow_unbounded_price_limits_age();
  }

  void allow_absent_portfolio() {
    allow_absent_portfolio_ = true;
    // Pushed down for the same reason the age policy is: this loop's hold is not in force at every
    // point an instruction can be settled, so the runner enforces the absence rule itself and an
    // opt-out that stopped at the hold would not actually opt out of anything.
    runner_->allow_absent_portfolio();
  }

  void set_session_window(strategy::SessionWindow window) {
    session_window_ = std::move(window);
  }

  const strategy::SessionWindow& session_window() const noexcept { return session_window_; }

  void set_failure_policy(FailurePolicy policy) { failure_policy_ = std::move(policy); }

  void set_control_file(FileStopChannel channel) { control_ = std::move(channel); }

  LoopReport run(StopSource& stop, std::size_t max_iterations = 0) {
    LoopReport report;
    // Carried in rather than left false: the field says whether the order list has been read at
    // least once, and the loop remembers that across runs. A second `run` on the same loop had been
    // reporting it as unread, which reads as a startup gate that never opened.
    report.order_list_read_ = order_list_read_;
    // **Differenced around the run, not accumulated from the passes.** An order can leave the
    // process from inside a source's `poll`, which calls `Runner::feed` itself and discards the
    // report it gets back, so a per-pass figure would only ever see what a stop did. The runner's
    // counter is monotonic for exactly this reason, and reading it at both ends is how
    // `RiskGuard::total_sent` is already used around a callback. Taken per `run` so a second run on
    // the same loop reports its own sends rather than the first one's as well.
    const auto outside_phase_before = runner_->sends_outside_permitted_phase();
    // Differenced for the reason above: both are monotone on the runner, which outlives this call.
    const auto bands_withheld_before = runner_->price_bands_withheld();
    const auto unclamped_rejected_before = runner_->unclamped_orders_rejected();
    const auto unclamped_refused_before = runner_->unclamped_orders_refused();
    // **The polling path's own figures, taken before any stop is honoured.** `honour` runs the stop
    // handler and the closing pass through the same `Runner::settle`, so a snapshot taken at the end
    // of `run` would count a refused flatten here *and* in `closing_refused_` -- one failure reported
    // twice, under a contract that says this family describes the polling path.
    const auto outcomes_before = instruction_outcomes();
    auto outcomes_after = outcomes_before;
    // **The record travels with the counters, and for the same reason.** Read after `honour`, a
    // refusal from the *closing* pass would have replaced it while the count still described a
    // polling one -- a report naming an order none of its figures counted.
    std::optional<BrokerRefusal> refusal_at_boundary;
    std::optional<strategy::Rejection> unbuildable_at_boundary;
    // Differenced around the run for the same reason, and the reason is sharper here: the source
    // survives `run` returning, so an absolute figure would report the previous run's charges again.
    const auto charged_before = [&]() {
      if constexpr (source_states_charged_costs<Source>::value) {
        return source_->charged_costs();
      } else {
        return ChargedCosts{};
      }
    }();
    for (auto& due : next_due_) {
      due.reset();
    }
    for (auto& count : consecutive_failures_) {
      count = 0;
    }
    closing_.reset();
    warmup_discarded_at_start_ = warmup_discarded();
    if constexpr (runner_counts_kept_drops<Runner>::value) {
      kept_dropped_at_start_ = runner_->kept_intents_dropped();
    }
 
    while (max_iterations == 0 || report.iterations_ < max_iterations) {
      const auto iteration = step(stop, report);
      ++report.iterations_;
      if (iteration.gate_) {
        ++report.blocked_;
      }
      report.final_gate_ = iteration.gate_;
      report.final_phase_ = iteration.phase_;
      if (iteration.phase_changed_) {
        ++report.phase_transitions_;
      }
      last_ = iteration;
 
      if (report.abandoned_) {
        if (!closing_) {
          outcomes_after = instruction_outcomes();
          refusal_at_boundary = runner_->last_broker_refusal();
          unbuildable_at_boundary = runner_->last_unbuildable();
        }
        // As for a stop: retained work lives for one run. A run resumed after an outage of unknown
        // length must not place what was decided before it.
        runner_->clear_retained_targets();
        if (const auto intent = stop.intent()) {
          interrupt_stop(*intent, report);
        } else if (owed_past_limit(clock_.now())) {
          // Giving up took `POLL_FAILING`, which lets a withdrawal through. The hold for an order nobody
          // can account for (`set_owed_order_limit`) is the stronger one and stays, with no stop
          // requested: no read has decided yet, and a later run on this loop takes it up again.
          hold(strategy::HaltReason::STOPPING);
          report.final_gate_ = runner_->guard().blocking_reason();
        }
        break;
      }
      if (const auto intent = stop.intent()) {
        if (!closing_) {
          // Sampled here, before the stop runs anything, for the reason stated where this began. Once,
          // when the stop is first seen: what a strategy asks for while the stop waits is refused by
          // the hold, and those refusals describe the stop, not the polling path.
          outcomes_after = instruction_outcomes();
          refusal_at_boundary = runner_->last_broker_refusal();
          unbuildable_at_boundary = runner_->last_unbuildable();
          begin_stop(report);
        }
        // A run's last permitted iteration honours the stop rather than leave it pending.
        const bool last_iteration = max_iterations != 0 && report.iterations_ >= max_iterations;
        if (!last_iteration && closing_should_wait(*intent, report)) {
          clock_.wait_for(closing_until_next());
          continue;
        }
        finish_stop(*intent, report);
        report.stopped_by_ = intent;
        break;
      }
      clock_.wait_for(until_next(iteration.at_));
    }
 
    // Deliberately **not** released here. A run that ended because an operator asked it to, or
    // because a poll kept failing, must leave the runner unable to send: releasing the halt on the
    // way out would reopen the gate at exactly the moment the loop stopped watching it. A caller who
    // genuinely wants to trade again resumes the guard themselves, which is a deliberate act.
    //
    // A healthy run that simply exhausted `max_iterations` never halted anything, so there is
    // nothing to leave behind in that case.
    report.sends_outside_permitted_phase_ =
      runner_->sends_outside_permitted_phase() - outside_phase_before;
    report.price_bands_withheld_ = runner_->price_bands_withheld() - bands_withheld_before;
    report.unclamped_orders_rejected_ =
      runner_->unclamped_orders_rejected() - unclamped_rejected_before;
    report.unclamped_orders_refused_ =
      runner_->unclamped_orders_refused() - unclamped_refused_before;
    // Taken at the loop's exit for a run nothing stopped, and at the stop for one something did.
    if (!report.stopped_by_ && !report.abandoned_) {
      outcomes_after = instruction_outcomes();
      refusal_at_boundary = runner_->last_broker_refusal();
      unbuildable_at_boundary = runner_->last_unbuildable();
    }
    note_instruction_outcomes(report, outcomes_before, outcomes_after);
    // **Carried only when this run's polling saw one**, which the count already decides: any refusal
    // during this run overwrote whatever the runner held before it, so a non-zero count means the
    // record is this run's. Comparing the records instead would need an equality `BrokerRefusal` has
    // no reason to have.
    if (report.orders_broker_refused_ != 0) {
      report.last_broker_refusal_ = refusal_at_boundary;
    }
    // The same rule for the reconciliation's own refusals.
    if (report.orders_unbuildable_ != 0) {
      report.last_unbuildable_ = unbuildable_at_boundary;
    }
    // **Said only when a band was actually held back**, not merely because no policy was stated. A
    // run that registers no band and polls none has nothing to decide, and a note telling it to
    // state a policy would be advice about a mechanism it is not using. Once per loop, like the
    // portfolio's own note, because a run that ignores it does not need it every iteration.
    //
    // **Asked of the runner rather than remembered here.** This loop borrows a mutable runner and
    // does not own its policy: a maximum stated directly on the runner before this loop existed
    // would have read as "nobody stated one", and a withdrawal made directly on the runner would
    // have been suppressed by this loop's memory of its own wrapper being called. One policy, one
    // place that answers for it.
    if (report.price_bands_withheld_ > 0 && !runner_->price_limits_age_policy_stated() &&
        !said_no_band_age_policy_) {
      said_no_band_age_policy_ = true;
      report.notes_.emplace_back(
        "no price band age policy: call set_maximum_price_limits_age or "
        "allow_unbounded_price_limits_age"
      );
    }
    if constexpr (source_states_charged_costs<Source>::value) {
      report.charged_costs_ = source_->charged_costs().since(charged_before);
    }
    note_sellable_convention(report);
    note_unestablished_phase(report);
    note_on_fill_never_fires(report);
    note_listing_outpaces_holdings(report);
    note_warmup_discarded(report);
    note_kept_intents_dropped(report);
    return report;
  }

  const IterationReport& last() const noexcept { return last_; }
 
private:
  // Puts the sellable-convention probe's verdict in the report, when it reached one.
  //
  // The probe answers a question no broker's documentation is known to -- whether the sellable figure
  // already excludes resting sells -- and it answers it as a by-product of ordinary polling. An
  // operator should not have to write code to read that, so a run that learned something says so,
  // with the counts behind it because one observation is evidence rather than proof.
  //
  // Silent when nothing was observed, which is most runs: a note saying "undetermined" on every run
  // that never sold anything is noise, and noise is how a report stops being read.
  void note_sellable_convention(LoopReport& report) const {
    const auto evidence = runner_->sellable_convention().overall();
    if (evidence.total() == 0) {
      return;
    }
    report.notes_.push_back(
      "broker's sellable figure appears " +
      std::string(strategy::sellable_convention_name(evidence.verdict())) + " (" +
      std::to_string(evidence.net_) + " net, " + std::to_string(evidence.gross_) + " gross)"
    );
  }
 
  // Returns how many instructions the runner has discarded as history, where it counts them.
  std::size_t warmup_discarded() const {
    if constexpr (runner_counts_warmup<Runner>::value) {
      return runner_->warmup_instructions_discarded();
    } else {
      return 0;
    }
  }
 
  // Says, once per run, how many instructions were decided on bars fed as history and not sent.
  //
  // Only the newest bar of each instrument's read may send (`Runner::feed_warmup`), so a cold start's
  // history and a backlog after an outage are shown to the strategy and nothing decided on them goes
  // out. Worth a line because it is also what a bar cadence slower than the bars' interval costs: a
  // poll bringing five minute bars acts on the newest one only.
  void note_warmup_discarded(LoopReport& report) const {
    const auto discarded = warmup_discarded() - warmup_discarded_at_start_;
    if (discarded == 0) {
      return;
    }
    report.notes_.push_back(
      std::to_string(discarded) + " instruction(s) decided on bars fed as history were not sent: "
      "only the newest bar of each instrument's read may send (poll bars no slower than their interval)"
    );
  }
 
  // Says, once per run, how many order intents kept for the phase were dropped at release.
  void note_kept_intents_dropped(LoopReport& report) const {
    if constexpr (runner_counts_kept_drops<Runner>::value) {
      const auto dropped = runner_->kept_intents_dropped() - kept_dropped_at_start_;
      if (dropped != 0) {
        report.notes_.push_back(
          std::to_string(dropped) + " order intent(s) kept for the market phase were not sent at the "
          "open: the instrument's holding had changed since they were decided, or a stop now rested"
        );
      }
    } else {
      static_cast<void>(report);
    }
  }
 
  // Says, once per run, that the strategy's `on_fill` never fires on this runner.
  //
  // See `Runner::on_fill_never_fires`. On every run rather than only the first, because a report is
  // read on its own, and a note that appeared only in a run nobody kept would be no note at all.
  void note_on_fill_never_fires(LoopReport& report) const {
    if constexpr (Runner::on_fill_never_fires) {
      report.notes_.push_back(
        "strategy defines on_fill; it never fires on this runner (fills are not observed live: "
        "read positions from holdings, see writing-a-strategy.md)"
      );
    } else {
      static_cast<void>(report);
    }
  }
 
  // Says, once per run, that working orders are polled more often than holdings.
  //
  // Not refused: it is a legitimate choice for a strategy that mostly watches its orders. But every
  // plan made between a listing and the next holdings read is withheld (`PollSchedule::working_orders_`
  // says why), and an operator reading a report full of `portfolio_superseded` should find the cause
  // stated beside it.
  void note_listing_outpaces_holdings(LoopReport& report) const {
    const auto& listing = schedule_.working_orders_;
    const auto& holdings = schedule_.holdings_;
    if (listing && holdings && *listing < *holdings) {
      report.notes_.push_back(
        "working orders are polled more often than holdings; plans made between a listing and the "
        "next holdings read are withheld as portfolio_superseded"
      );
    }
  }
 
  // Runs one iteration: poll what is due, then let the runner act if the gate is open.
  IterationReport step(StopSource& stop, LoopReport& report) {
    IterationReport iteration;
    iteration.at_ = clock_.now();
 
    // Read before polling, so an operator's stop is honoured without first spending requests on a
    // broker the loop is about to walk away from.
    check_control_file(stop, report);
    // A stop discards what was retained for a better account read: nothing decided before it is to be
    // replayed after it.
    if (stop.requested()) {
      runner_->clear_retained_targets();
    }
 
    // **Installed before anything can send, and announced only after the polls.** The two halves are
    // separate on purpose and neither is optional.
    //
    // A source's `poll` calls `Runner::feed`, which settles, so an order can leave the process from
    // inside a poll -- the same fact the gate below is computed early for. If the phase were installed
    // after the polls, the first bar arriving after the close would be settled while the context still
    // said `OPEN`, and an opening order would pass a phase limit that exists to refuse it. Once per
    // session, on the close.
    //
    // Announcing the change is the opposite case: `on_phase` may place orders, and it should see this
    // iteration's polls rather than the previous one's -- fresh prices, a fresh portfolio, and a gate
    // re-evaluated against them. That is what the replay's step does (`backtest/engine.hpp`, "The price moves with the portfolio"), with
    // the same argument: a handler deciding on a mark from one moment beside a price from another is
    // not looking at a fresher picture, it is looking at an incoherent one.
    install_phase(iteration);
 
    // **Before anything this loop calls can feed, because several of them can.** The runner measures
    // a snapshot's age against the later of this and its own decision instant, and its decision
    // instant is a bar's period end -- which on a daily bar is yesterday's close. Set here, a plan
    // settled from inside `read_order_list` or a poll is aged against the clock rather than against
    // the event that triggered it.
    runner_->set_observed_at(iteration.at_);
    // Stated here too, and for the same reason: the runner enforces the absence rule at the moment
    // an instruction is decided, which is the only place that is inside every callback this loop
    // makes. A run that opted out says so, and the runner then places as it did before.
    runner_->require_known_portfolio(!allow_absent_portfolio_);
 
    // **Established before any poll runs, because a poll is what feeds the runner.** A source's
    // `poll` calls `Runner::feed` itself -- `live::SessionSource::poll_bars` does, per instrument --
    // and `feed` settles, so an order can leave the process from inside a poll. Computing the gate
    // afterwards would mean the startup gate and the staleness gate were installed one poll too
    // late, and on the very first iteration they would not be installed at all: a bar arriving
    // before the first order-list read could place an order against a position a predecessor may
    // have left, which is the exact hazard the startup gate exists for.
    hold(gate_for(stop, iteration.at_, report, /*may_read=*/true));
 
    for (const auto kind : all_poll_kinds) {
      if (!is_due(kind, iteration.at_)) {
        continue;
      }
      // **Re-established immediately before every poll**, because a poll is a point at which an order
      // can leave the process and anything that ran before it may have replaced the calendar the phase
      // was derived from. Three things can have: a previous poll in this same loop, the calendar poll
      // in particular since it is polled first, and `gate_for`'s out-of-band `read_order_list` above.
      //
      // Before rather than after, which is the correction: "after every poll" closed the first two and
      // left the third, because the pre-loop install runs *before* `gate_for` and nothing re-derived
      // the phase between that call and the first poll. Placing it here makes the rule positional
      // instead of enumerated -- every send site is preceded by an install, and no list of which polls
      // can change what has to stay correct.
      //
      // The cost is a date comparison per due poll.
      install_phase(iteration);
      // **And the gate, for the same reason.** The polls before this one may have read the account: on
      // a cold start the hold for an account never read was installed before the polls, and without
      // re-evaluating it here it would still stand when the bars read after this iteration's holdings
      // are fed -- withholding the first decision against an account that has, by then, been read.
      // `may_read` is false: the order list is attempted once per iteration, above.
      hold(gate_for(stop, iteration.at_, report, /*may_read=*/false));
      const auto watch = watch_phase(iteration);
      // **Sampled before the poll, not after it.** A poll can send: `Runner::feed` settles, and a
      // working-order poll now feeds `on_order_update` as well as the bar and quote polls feeding
      // their handlers. An order a handler places during the working-order poll was sent *after*
      // the listing was fetched and cannot be in it -- so counting placements afterwards would
      // stamp that listing as though it covered the order, and `closing_source_verified` would
      // then approve a flatten against a picture missing a resting order. The count taken here is
      // the one that was true when the listing was read.
      const auto placements_before_poll = runner_->placements_at_broker();
      const auto outcome = source_->poll(kind, *runner_);
      if (kind == PollKind::WORKING_ORDERS) {
        working_asked_at_ = clock_.monotonic_now();
      } else if (kind == PollKind::HOLDINGS) {
        holdings_asked_at_ = clock_.monotonic_now();
      }
      note_revised_phase(std::string(poll_kind_name(kind)) + " poll", watch, report);
      // `UNCHANGED` counts, because a listing that is identical to the last one is still a listing
      // -- but only on a source that publishes them at all, which `note_working_orders_read`
      // enforces: a broker without the inquiry answers `UNCHANGED` without asking anybody.
      if (kind == PollKind::WORKING_ORDERS &&
          (outcome == PollOutcome::UPDATED || outcome == PollOutcome::UNCHANGED)) {
        note_working_orders_read(iteration.at_, placements_before_poll);
      }
      ++report.attempted_[static_cast<int>(kind)];
      iteration.polls_.emplace_back(kind, outcome);
      schedule_next(kind, iteration.at_);
      note_outcome(kind, outcome, report);
      if (outcome == PollOutcome::UPDATED) {
        iteration.fed_ = true;
      }
      if (report.abandoned_) {
        // Recorded before returning, so the report does not claim sending was permitted while the
        // guard `note_outcome` just halted says otherwise.
        iteration.gate_ = runner_->guard().blocking_reason();
        return iteration;
      }
    }
 
    // Observed after the polls, not before: a poll can refresh the listing without a bar arriving,
    // and that is precisely when an order of ours begins showing as resting. Observing first would
    // read every iteration's portfolio one iteration late. A repeat observation of unchanged state
    // concludes nothing, so a quiet iteration costs a scan and teaches the probe nothing.
    runner_->observe_sellable_convention();
 
    // Re-evaluated after the polls so the report and the next iteration reflect what those polls
    // changed -- a holdings refresh should lift the staleness hold now, not one iteration later.
    // The order list is deliberately not attempted a second time: one attempt per iteration, and
    // the gate simply stands until the next one.
    // The last send site: the announcement below settles what `on_phase` asks for, and the final poll
    // may have replaced the calendar after its own pre-poll install.
    install_phase(iteration);
    hold(gate_for(stop, iteration.at_, report, /*may_read=*/false));
 
    // Announced last, so the handler runs on everything this iteration established. A change is
    // announced once, not every iteration it persists for.
    if (announced_ != iteration.phase_) {
      // **The clock moves with the phase, and only forwards.** `Runner::feed(MarketPhase)` does not
      // advance `Context::now_`; only a bar or a quote does, from a broker's stamp. So a daily-bar run
      // would hand `on_phase` an instant hours old, and `RiskGuard` files a send at
      // `GuardInputs::now_` and counts sends within `rate_window_` of it -- the phase order and the
      // next bar's order would be filed at what the guard thinks are two unrelated moments, and two
      // orders would pass a one-order window. The replay's step (`backtest/engine.hpp`) documents the same hazard
      // for the replay.
      //
      // Forwards only: a broker stamp ahead of this machine's clock must not be walked backwards,
      // because a handler deciding at an instant *earlier* than the bar it has already been shown is
      // the same incoherence with the sign flipped. The skew is the broker's and is not this loop's
      // to quietly correct.
      //
      // A bar fed earlier in this same iteration set `now_` to its own period end, which on a daily bar
      // is yesterday's close. Its order is still counted in the rate window at this iteration's
      // reading, beside this one: the runner counts a send at the later of the decision instant and
      // the reading this loop gave it.
      if (runner_->context().now() < iteration.at_) {
        runner_->context().set_now(iteration.at_);
      }
      announced_ = iteration.phase_;
      iteration.phase_changed_ = true;
      // Watched for the same reason a poll is: `Context::set_calendar` is public, so an `on_phase`
      // handler can revise the calendar and submit in one call, and `settle` runs before this returns.
      const auto watch = watch_phase(iteration);
      runner_->feed(iteration.phase_);
      note_revised_phase("on_phase handler", watch, report);
    }
 
    // **Last of all the iteration's decisions**, so every handler that could restate the instrument --
    // the bars, the quotes, the listing's updates, and `on_phase` just above -- has spoken first, and
    // its instruction has replaced the retained target; a retry before `on_phase` could send an entry
    // the phase change was about to countermand. And after every account poll and the gate they
    // refreshed, so what is retried is sized against this iteration's holdings, listing and budget,
    // under this iteration's holds -- inside the holdings poll it ran before the gate lifted and before
    // buying power was read.
    runner_->retry_retained_targets();
 
    // Reported from the guard rather than from what the loop just computed, so the report names the
    // most fundamental hold in force whoever owns it -- an operator's halt, or a journal that
    // failed, outranks anything the loop decided.
    iteration.gate_ = runner_->guard().blocking_reason();
    return iteration;
  }
 
  // What the phase and the send count were before the loop handed control to somebody else.
  struct PhaseWatch {
    std::chrono::system_clock::time_point at_{};
    strategy::MarketPhase phase_{strategy::MarketPhase::UNKNOWN};
    std::size_t sends_{0};
  };
 
  // Records the phase and the send count, for comparison after a call into user code.
  PhaseWatch watch_phase(const IterationReport& iteration) const {
    return PhaseWatch{iteration.at_, iteration.phase_, runner_->guard().total_sent()};
  }
 
  // Counts and names a call that sent an order and left the phase changed behind it.
  //
  // **The one hazard here the loop cannot prevent and can only witness.** It re-derives the phase before
  // every call it makes that could send, but nothing of its runs *inside* such a call: a
  // source that replaces the calendar and then feeds, or an `on_phase` handler that does the same before
  // submitting, settles an order against the phase as it stood when the call began. So the loop looks
  // afterwards, which turns an invisible violation into a line in the report -- the same choice
  // `closing_source_unverified_` makes about a listing it could not trust.
  //
  // **Both halves are required, and each one removes a wrong answer.** The phase must have moved, or an
  // ordinary send would be reported. Something must have been sent, or an ordinary calendar poll would be
  // reported on every iteration -- which is why this no longer excludes the calendar poll by kind. That
  // exclusion was a false-positive workaround and it hid this case: a source obeying
  // the letter of the contract, writing the calendar in the calendar poll and *then* feeding.
  //
  // `total_sent()` rather than the guard's rate-window deque, because that one prunes and can fall, so
  // it cannot be differenced across a call.
  void note_revised_phase(const std::string& what, const PhaseWatch& before, LoopReport& report) {
    if (runner_->guard().total_sent() == before.sends_) {
      return;
    }
    if (phase_at(before.at_) == before.phase_) {
      return;
    }
    ++report.sends_under_a_revised_phase_;
    if (report.sends_under_a_revised_phase_ == 1) {
      // Once per run, not once per call: a strategy doing this on every bar would otherwise bury every
      // other note under thousands of copies of this one.
      // **Worded to exactly what was observed**, which is co-occurrence and not causation. Two cases
      // the stronger claim would have got wrong: a call that sent *first* and revised the calendar
      // afterwards, where no order used the stale answer; and a call that only cancelled, since
      // `Runner::send_cancel` never consults the phase at all. Both still break the writer contract and
      // are worth reporting, so the note states the facts and names the check rather than the verdict.
      report.notes_.push_back(
        "the " + what + " changed the market phase and sent a request in the same call; check whether "
        "that request used the earlier phase. Only the calendar poll may write the trading calendar, "
        "and it must not send");
    }
  }
 
  // Computes the phase for this iteration and installs it, so nothing can send under a stale one.
  //
  // **Called immediately before every point at which this loop hands control to the source**: once
  // before `gate_for`, which performs an out-of-band order-list read; once before each due poll; and
  // once before the phase announcement. So the invariant is exactly this, and no stronger:
  // **whenever the loop calls into the source, `Context::phase_` is derived from the calendar as it
  // stood at that call.**
  //
  // **What that deliberately does not promise**, because it cannot: a source is free to replace the
  // calendar and then feed the runner inside a single `poll`, and no loop code runs in between, so
  // that order would settle against the phase as it stood when the poll began. `PollKind::CALENDAR`
  // documents the contract that rules it out, `SessionSource` satisfies it -- it writes the calendar in
  // the calendar poll and nowhere else -- and `note_revised_phase` reports a source that
  // breaks it, because a violation nobody is told about is one nobody fixes. The stronger at-send-site
  // invariant does not hold.
  //
  // The instant is the iteration's own `at_` throughout, never re-read from the clock, so a poll that
  // took a second does not move the session boundary mid-iteration.
  void install_phase(IterationReport& iteration) {
    iteration.phase_ = phase_at(iteration.at_);
    // Recorded here rather than derived later, and it outlives a `run` for the same reason `announced_`
    // does: "this loop has never known the phase" is a claim about the loop, not about one call.
    if (iteration.phase_ != strategy::MarketPhase::UNKNOWN) {
      phase_ever_known_ = true;
    }
    runner_->context().set_phase(iteration.phase_);
  }
 
  // Returns the phase at `at`, from the configured window and whatever calendar has arrived.
  //
  // The composition `ReplayClock` already performs, with a polled calendar in place of a replay's
  // per-bar answer. An uncovered date is `UNKNOWN` and never "closed": `TradingCalendar` draws that
  // distinction because assuming closed stops a strategy on a day that is in fact open, and assuming
  // open lets one order into a holiday.
  //
  // The date is the *market's*, not this process's. `SessionWindow` owns the offset and derives both
  // answers from it, so the day the calendar is asked about and the day the session boundaries are
  // computed for cannot disagree.
  strategy::MarketPhase phase_at(std::chrono::system_clock::time_point at) const {
    const auto& calendar = runner_->context().calendar();
    return session_window_.phase(at, calendar.is_trading_day(session_window_.local_date(at)));
  }
 
  // The clause naming unreadable calendar rows, or nothing when there were none.
  //
  // A row whose date the broker sent unreadably is dropped, so the day it was for is uncovered and
  // the note above would otherwise send an operator to check a span or a broker that is fine. The
  // count says the broker *did* answer, and what it answered could not be read.
  std::string unreadable_calendar_rows() const {
    const auto unreadable = runner_->context().calendar().unreadable_days_;
    if (unreadable == 0) {
      return {};
    }
    return "; the broker sent " + std::to_string(unreadable) + " calendar row" +
      (unreadable == 1 ? "" : "s") + " whose date could not be read";
  }
 
  // Says so when a run never established the phase, and names which of the three reasons applies.
  //
  // Silent otherwise, like `note_sellable_convention`: a line on every healthy run is how a report
  // stops being read. But a run that placed nothing because the phase was never known must not look
  // like a quiet market -- and this is **not** a hold on the guard, deliberately. A hold would also
  // stop cancellations, which `Runner::send_cancel` refuses on any blocking reason while never
  // consulting the phase at all; taking one because a reference-data poll failed would make an
  // unknown phase more restrictive than a closed one, and a withdrawal only ever reduces exposure.
  // `RiskGuard` already refuses every placement that opens or increases a position with
  // `PHASE_NOT_PERMITTED`, and permits one that closes a held position -- which is exactly what a
  // hold would wrongly stop, for the same reason it would wrongly stop a cancellation.
  void note_unestablished_phase(LoopReport& report) const {
    if (report.iterations_ == 0 || report.final_phase_ != strategy::MarketPhase::UNKNOWN) {
      return;
    }
    // **"Never" is a claim, so it is tracked rather than derived.** A loop can establish the phase and
    // then lose it -- the local date rolls past what the calendar covers, or the broker revises it --
    // and telling an operator it was never established would send them looking for a configuration
    // fault that is not there.
    //
    // This used to read `report.phase_transitions_ > 0`, which was wrong: `announced_`
    // outlives a `run` while the report's counter restarts at zero, so a run with nothing left to
    // announce reports no transitions and would claim the phase had never been known -- of a loop that
    // spent its first run open. The verdict was right in the case that was tested and the reasoning was
    // wrong, which is the worse of the two.
    const bool ever_known = phase_ever_known_;
    // Three reasons, not four. "The poll never came due" was here and is unreachable: `run` resets
    // `next_due_`, so a configured cadence makes the calendar due on the first iteration of every
    // run. A branch describing a state that cannot happen is worse than no branch, because a reader
    // who sees it believes it can.
    std::string reason;
    if (!schedule_.calendar_) {
      reason = "no calendar cadence is configured";
    } else if (report.failed(PollKind::CALENDAR) == report.attempted(PollKind::CALENDAR)) {
      reason = "every calendar poll failed";
    } else if (runner_->context().calendar().days_.empty()) {
      // What the loop can actually tell: a poll reported success and no calendar arrived. It does
      // **not** claim to know why, because it cannot distinguish a broker that publishes none -- where
      // the request is not even compiled -- from a source that fetched one and installed nothing.
      reason = "a calendar poll succeeded but no calendar arrived (this broker may publish none)";
      reason += unreadable_calendar_rows();
    } else {
      // Named, not merely called uncovered. "The calendar does not cover today" sends a reader to
      // check a span they cannot see; the date is the one thing that turns this note into an action.
      const auto date = session_window_.local_date(last_.at_);
      reason = "the calendar does not cover " + std::to_string(date.year_) + "-" +
        std::to_string(date.month_) + "-" + std::to_string(date.day_);
      reason += unreadable_calendar_rows();
    }
    report.notes_.push_back(
      std::string(ever_known ?
        "the market phase stopped being established (" :
        "the market phase was never established (") +
      reason + "); a guard permitting only known phases refused every placement opening a position" +
      (ever_known ? " after that" : ""));
  }
 
  // Attempts the startup order-list read once, and returns whether it succeeded.
  //
  // The startup gate's read and a pending stop's both come here, so the two cannot drift apart.
  bool read_order_list_once(std::chrono::system_clock::time_point at, LoopReport& report) {
    // Watched like a poll, because it is the same kind of thing: a call into the source, from which a
    // source that revised the calendar and sent would reach the identical hazard through a different
    // door. The phase in force here is the one `step` installed immediately before calling this.
    const auto watch = PhaseWatch{at, phase_at(at), runner_->guard().total_sent()};
    // As in `step`, and here the startup read dispatches nothing at all -- but the count is taken
    // before the call for the same reason, so that the rule holds positionally rather than by an
    // argument about which reads can send.
    const auto placements_before_read = runner_->placements_at_broker();
    const auto outcome = source_->read_order_list(*runner_);
    // The working-order endpoint, where the source has one: a closing wait counts this as asking it.
    working_asked_at_ = clock_.monotonic_now();
    note_revised_phase("order-list read", watch, report);
    if (outcome != PollOutcome::UPDATED && outcome != PollOutcome::UNCHANGED) {
      return false;
    }
    order_list_read_ = true;
    report.order_list_read_ = true;
    order_list_at_ = at;
    // The startup read is a working-order read *on a source that has the inquiry*. On one that
    // does not, it reads order history for the journal and never populates `working_`, so
    // `note_working_orders_read` deliberately does nothing and a closing pass stays unverified.
    note_working_orders_read(at, placements_before_read);
    return true;
  }
 
  // Returns which of the loop's *own* conditions blocks sending, reading the order list if due.
  //
  // Only conditions the loop computes, because a hold is released by whoever took it. `OPERATOR`
  // belongs to a person and `LOCAL_FAILURE` to the runner; a loop that took either would be a
  // second holder of someone else's hold -- the defect `HaltReason` exists to prevent,
  // reintroduced one layer up.
  //
  // **Always evaluated, even while someone else's hold stands.** Short-circuiting on a foreign hold
  // would leave the loop's own hold un-maintained underneath it, so the moment an operator resumed,
  // the guard would be wide open against a portfolio nobody had rechecked.
  //
  // Called twice per iteration -- before the polls, which is what actually protects anything, and
  // after them so the report is current. `may_read` is false on the second call so the order-list
  // read is attempted at most once per iteration.
  std::optional<strategy::HaltReason> gate_for(
    StopSource& stop,
    std::chrono::system_clock::time_point at,
    LoopReport& report,
    bool may_read
  ) {
    if (stop.requested()) {
      // **A withdraw or a flatten reads the broker's order list first, if it never has.** Each builds
      // its instructions from the picture, and a stop pending before the first iteration -- a process
      // started with `flatten` in its control file -- used to return here before the startup read, so
      // its closing pass was unverified and a flatten closed nothing, for any position. That also left
      // no way to flatten a position behind a resting stop: the in-process flatten withdraws the stop
      // and, rightly, sends nothing beside it, and the restart that should finish the job could not.
      // The read is the startup gate's own, under its conditions; a quiesce sends nothing and needs no
      // picture, so it does not spend the request.
      //
      // **`STOPPING` on every path, the failed read included.** `ORDER_LIST_UNREAD` lets reducing
      // orders through (`halt_blocks`), and nothing but the stop's own may go out once one is pending.
      // Held before the read as well, because the source is a template parameter and may feed.
      const auto intent = stop.intent();
      // Once a stop is waiting, on the wait's own floor: a read that failed is not asked again at
      // every tick. The iteration that takes the stop up reads as any other.
      const bool on_the_floor = !closing_ || !asked_within_floor(PollKind::WORKING_ORDERS);
      if (intent && stop_intent_closes(*intent) && !order_list_read_ && may_read && on_the_floor &&
          !runner_->guard().halted(strategy::HaltReason::LOCAL_FAILURE)) {
        hold(strategy::HaltReason::STOPPING);
        read_order_list_once(at, report);
      }
      return strategy::HaltReason::STOPPING;
    }
    if (!order_list_read_) {
      // Not attempted when the caller says not to, and not attempted at all while the runner's
      // journal is broken: a runner that cannot record what it does must not be sent looking for
      // more work to fail at. Either way the gate stands, which is the safe answer.
      //
      // `STOPPING` rather than `ORDER_LIST_UNREAD` on each path here while an order nobody can account
      // for is past the limit (below): the stronger hold, which also refuses a withdrawal, must not be
      // traded for the weaker one because a read failed. And held before the read, since the source is
      // a template parameter and its read may feed.
      const bool owed = owed_past_limit(clock_.now());
      const auto unread = owed ? strategy::HaltReason::STOPPING : strategy::HaltReason::ORDER_LIST_UNREAD;
      if (!may_read || runner_->guard().halted(strategy::HaltReason::LOCAL_FAILURE)) {
        return unread;
      }
      if (owed) {
        hold(strategy::HaltReason::STOPPING);
      }
      if (!read_order_list_once(at, report)) {
        return unread;
      }
    }
    // **An order nobody can account for, owed too long, stops the run's placements**
    // (`set_owed_order_limit`), in two steps. Once the limit has passed on the clock, this gate holds
    // `STOPPING`; it is evaluated before every poll, so an order-list read that begins after a check
    // that found the limit passed, and whatever its handlers ask for, runs under the hold (what was
    // already under way at the limit runs on the check before it). Once a read in an iteration that
    // began after the limit has succeeded and left the order unaccounted for, the stop is requested:
    // that iteration's every check found the limit passed, so that read ran under the hold. A read
    // whose sweep names the order lifts the hold instead.
    if (auto owed = owed_too_long()) {
      report.notes_.push_back(owed_stop_note(*owed));
      report.owed_stop_ = std::move(owed);
      stop.request(StopIntent::QUIESCE);
      return strategy::HaltReason::STOPPING;
    }
    if (owed_past_limit(clock_.now())) {
      return strategy::HaltReason::STOPPING;
    }
    // Measured from the snapshot's own `as_of_` rather than from when the poll returned, because
    // that is when the broker says the positions were true. A poll that succeeded at noon carrying
    // a snapshot the broker took at 11:00 is an hour stale, and the poll's own timestamp would call
    // it fresh.
    const auto as_of = runner_->context().portfolio().as_of_;
    if (as_of == std::chrono::system_clock::time_point{}) {
      // A default-constructed `Portfolio` carries the epoch, and no broker ever reports a snapshot
      // taken in 1970, so the epoch means **never populated** -- which is not the same as old, and
      // is the more dangerous of the two: an empty portfolio is the *permissive* answer to every
      // question the send path asks it, so a run that has never read the account sizes an entry as
      // though it were flat.
      //
      // **No longer inside the age check.** It was, and hoisting it was not a one-line change:
      // the hold would reach cancellations, and an order that cannot be withdrawn keeps resting and
      // can fill. That objection is answered rather than avoided -- `strategy::halt_blocks` lets a
      // named withdrawal through a hold that only means this runner is uninformed, and
      // `Runner::plan_withhold_reason` stops the *derived* cancellation that objection overlooked.
      //
      // The opt-out reaches the age comparison below as well, not just this hold. The epoch is 1970
      // and therefore older than any bound an operator would state, so an absence opt-out that did
      // not skip the comparison would simply move the refusal to `PORTFOLIO_STALE`.
      if (!allow_absent_portfolio_) {
        return strategy::HaltReason::PORTFOLIO_ABSENT;
      }
      // Opted out, so the age comparison is skipped rather than fallen through to -- and the checks
      // after this one still run, because an absent portfolio says nothing about buying power.
    }
 
    // **A run must state how old a snapshot may be, or state that it accepts any age.**
    //
    // This was optional, and optional meant that a run which never thought about holdings sent
    // orders sized against whatever the portfolio happened to contain -- including nothing at all.
    // Fail-open by default was the defect. The library states no default
    // duration of its own, because any constant it picked would be a risk policy it has no basis
    // for: how stale a position may be before it must not be sized against is the operator's
    // judgement about their own market and cadence, not this library's.
    // **Checked after absence, not before it.** Both can be true at once, and "you have never read
    // this account" is the more useful of the two to be told: the missing policy is a configuration
    // fact the note below states in full, while absence is a fact about the data every ceiling and
    // every target subtraction is about to be measured against.
    if (!maximum_portfolio_age_ && !allow_unbounded_portfolio_age_) {
      if (!said_no_age_policy_) {
        said_no_age_policy_ = true;
        report.notes_.emplace_back(
          "no portfolio age policy: call set_maximum_portfolio_age or allow_unbounded_portfolio_age"
        );
      }
      return strategy::HaltReason::PORTFOLIO_STALE;
    }
    if (maximum_portfolio_age_ && as_of != std::chrono::system_clock::time_point{} &&
        at - as_of > *maximum_portfolio_age_) {
      return strategy::HaltReason::PORTFOLIO_STALE;
    }
    // **A runner that keeps its orders until the broker accounts for them needs the listing.** A holdings
    // poll refreshes positions and keeps `working_`; only a working-order poll replaces it. A runner
    // that sends to a broker keeps each order it sent in `working_` until that listing arrives
    // (`Runner::persist_reservation`), so with no listing -- none published, none polled, or one
    // older than the policy -- an order that expired or was refused would stay "working" for ever,
    // a target would stop being chased, and a filled one would be cancelled again on every pass.
    // Held, with the missing piece named once, whatever the portfolio age policy: opting out of a
    // holdings age is not opting out of a listing.
    if constexpr (runner_keeps_sent_orders<Runner>::value) {
      if (const auto missing = listing_missing(at)) {
        if ((said_listing_ & missing->first) == 0) {
          said_listing_ |= missing->first;
          report.notes_.emplace_back(missing->second);
        }
        return strategy::HaltReason::PORTFOLIO_STALE;
      }
    }
    if (runner_->guard().limits().enforce_buying_power_) {
      // Checked only when the caller asked for the limit. A loop with no buying-power cadence and
      // no check configured is an ordinary loop, and must not be held for a reading nobody wants.
      if (const auto unread = buying_power_unread(at)) {
        return *unread;
      }
    }
    return std::nullopt;
  }
 
  // Returns whether some budget the run needs is missing or too old to size against.
  //
  // **Every configured currency must have its own fresh reading**, not merely one of them. A source
  // that names its currencies is asked, because an account trading two and answering for one is a
  // partial answer: the per-order guard would still refuse the unread currency, but the loop would
  // go on reporting an open gate while a budget it was told to watch never arrived at all.
  //
  // A source that names none falls back to "any reading at all", which is the most the loop can
  // check without knowing what was asked for.
  std::optional<strategy::HaltReason> buying_power_unread(
    std::chrono::system_clock::time_point at
  ) const {
    const auto& readings = runner_->context().portfolio().buying_power_;
    // Aged against the reading's own stamp, for the same reason the portfolio is: a figure that
    // arrived an hour ago is an hour old however recently the poll succeeded.
    const auto usable = [&](const BuyingPower& reading) {
      return !maximum_buying_power_age_ ||
        at - reading.received_at_ <= *maximum_buying_power_age_;
    };
    if constexpr (source_names_buying_power_currencies<Source>::value) {
      const auto& wanted = source_->buying_power_currencies();
      if (!wanted.empty()) {
        for (const auto& currency : wanted) {
          bool found = false;
          for (const auto& reading : readings) {
            if (reading.cash_.currency_ == currency && usable(reading)) {
              found = true;
              break;
            }
          }
          if (!found) {
            return strategy::HaltReason::BUYING_POWER_UNREAD;
          }
        }
        return std::nullopt;
      }
    }
    if (readings.empty()) {
      return strategy::HaltReason::BUYING_POWER_UNREAD;
    }
    for (const auto& reading : readings) {
      if (!usable(reading)) {
        return strategy::HaltReason::BUYING_POWER_UNREAD;
      }
    }
    return std::nullopt;
  }
 
  // Sets how old a buying-power reading may be before the gate closes on it.
public:
  void set_maximum_buying_power_age(std::chrono::nanoseconds age) {
    maximum_buying_power_age_ = age;
  }
 
private:
  // Reads the control file, escalating `stop` when it names an intent.
  void check_control_file(StopSource& stop, LoopReport& report) {
    if (!control_.enabled()) {
      return;
    }
    if (const auto intent = control_.poll()) {
      stop.request(*intent);
      return;
    }
    if (!control_.unrecognized().empty()) {
      // Reported rather than guessed at, because picking the nearest intent would decide how much of
      // an operator's position to close on their behalf.
      report.notes_.push_back(
        "control file names no intent: '" + control_.unrecognized() + "'");
    }
  }
 
  // Counts an outcome, and abandons the run when one kind keeps failing.
  void note_outcome(PollKind kind, PollOutcome outcome, LoopReport& report) {
    const int at = static_cast<int>(kind);
    if (outcome == PollOutcome::FAILED) {
      ++report.failed_[at];
      ++consecutive_failures_[at];
      if (consecutive_failures_[at] >= failure_policy_.consecutive_limit_) {
        report.abandoned_ = true;
        report.abandoned_kind_ = kind;
        report.notes_.push_back(
          std::string("gave up after repeated failures polling ") + std::string(poll_kind_name(kind)));
        hold(strategy::HaltReason::POLL_FAILING);
      }
      return;
    }
    switch (outcome) {
      case PollOutcome::FAILED:
        // Counted above.
        return;
      case PollOutcome::RATE_LIMITED:
        // The limiter working says nothing either way, so the count stands: a kind alternating
        // failure and refusal is still broken.
        return;
      case PollOutcome::UPDATED:
      case PollOutcome::UNCHANGED:
        // A poll that answered, with something new or with nothing new, ends a run of failures.
        consecutive_failures_[at] = 0;
        return;
    }
  }
 
  // Enters a stop, once per run: the hold, the strategy's notice, and the end of whatever it had pending.
  // A stop that closes may then wait (`closing_should_wait`) before `finish_stop` builds its pass.
  void begin_stop(LoopReport& report) {
    closing_ = ClosingWait{};
    closing_->began_ = clock_.monotonic_now();
    closing_->deadline_ = closing_->began_ + std::chrono::duration_cast<std::chrono::steady_clock::duration>(
      closing_wait_ > std::chrono::nanoseconds::zero() ? closing_wait_ : std::chrono::nanoseconds::zero());
    hold(strategy::HaltReason::STOPPING);
    // Whenever the stop was requested -- before the iteration or during one of its polls -- nothing
    // retained before it is to be replayed after it, by this run or a resumed one. A target waiting on
    // a resting stop is the same kind of work.
    runner_->clear_retained_targets();
    runner_->clear_deferred_targets();
 
    // **The strategy is told first, and cannot trade on the way out.** `on_stop` is dispatched here,
    // while the hold stands, so every instruction it asks for is refused by the guard and counted.
    // It must not be dispatched later: from the moment the loop queues its own closing orders, a
    // strategy's order would be planned against the same snapshot and settled in the same window,
    // going through none of the sizing the closing pass does -- a second full-size close beside the
    // first, or a placement during a `WITHDRAW`, which promises none.
    // **Not watched, and it does not need to be.** `on_stop` runs with a mutable context and can revise
    // the calendar like any other handler, but `hold(STOPPING)` above is already in force, so every
    // instruction it asks for is refused before anything is sent -- which `stop_handler_sent_` asserts as
    // a report field precisely so that ordering cannot be broken silently. The revision it may have made
    // *is* answered, one call later: the closing derive below picks it up.
    //
    // For the record, the complete list of calls this loop makes out of itself is: `source_->poll`,
    // `source_->read_order_list`, `source_->buying_power_currencies` (a const accessor that is handed no
    // runner and so can neither feed nor write a calendar), `runner_->observe_sellable_convention` (which
    // reads the portfolio and dispatches nothing), this one, and `settle_closing`. The two that can send
    // under a phase are watched.
    const auto parting = runner_->notify_stop();
    // `withheld_plans_` counts here for the same reason it counts in `Report::quiet`: an `on_stop`
    // handler may state a target, `STOPPING` is held by then, and a plan that would have cancelled
    // something is withheld whole. Left out of this total, the loop would report that a parting
    // instruction met no resistance while the instruction never went anywhere.
    report.stop_handler_refused_ = parting.refused_.size() + parting.refused_cancel_.size() +
      parting.rejected_.size() + parting.withheld_.size() + parting.withheld_plans_.size() +
      parting.unavailable_.size();
    report.stop_handler_sent_ = parting.would_place_.size() + parting.would_cancel_.size();
  }
 
  // Leaves the runner unable to send when a run is abandoned with a stop pending, and says the stop's
  // closing pass did not run. The abandonment took `POLL_FAILING`, under which a named withdrawal still
  // goes; a stop asked for more than that.
  void interrupt_stop(StopIntent intent, LoopReport& report) {
    hold(strategy::HaltReason::STOPPING);
    // The stop may never have been taken up (`begin_stop`): a target waiting on a resting stop order is
    // not to go out after it either.
    runner_->clear_deferred_targets();
    report.stopped_by_ = intent;
    if (closing_) {
      report.closing_waited_ = std::chrono::duration_cast<std::chrono::nanoseconds>(
        clock_.monotonic_now() - closing_->began_);
    }
    if (stop_intent_closes(intent)) {
      report.closing_interrupted_ = true;
      report.notes_.push_back(
        std::string("the run was abandoned for failing polls with a ") + std::string(stop_intent_name(intent)) +
        " pending: its closing pass did not run, nothing was withdrawn or closed by it, and the runner is"
        " left unable to send; read the broker's order list and holdings");
    }
  }
 
  // Carries out `intent`, which is where the three stops actually differ.
  void finish_stop(StopIntent intent, LoopReport& report) {
    report.closing_waited_ = std::chrono::duration_cast<std::chrono::nanoseconds>(
      clock_.monotonic_now() - closing_->began_);
    closing_->reads_wanted_ = false;
    if (!stop_intent_closes(intent)) {
      return;
    }
    // **Whatever a strategy queued without settling goes nowhere.** The closing instructions below are
    // settled with the loop's hold lifted, and anything else still on the queue would be settled with
    // them. Settled here, under the hold, it is refused like the rest of what a strategy asks during a stop.
    {
      const auto stale = runner_->settle_closing();
      report.stop_handler_refused_ += stale.refused_.size() + stale.refused_cancel_.size() +
        stale.rejected_.size() + stale.withheld_.size() + stale.withheld_plans_.size() + stale.unavailable_.size();
      report.stop_handler_sent_ += stale.would_place_.size() + stale.would_cancel_.size();
    }
    // Nothing decided under the hold is kept, so this clears nothing today. It stands so that a change
    // to that rule cannot let kept work out with the closing orders below.
    runner_->clear_retained_targets();
    runner_->clear_deferred_targets();
 
    // **Both intents state their instructions outright rather than expressing them as a target.**
    //
    // A withdraw always had to: there is no target to reconcile against, so the working orders are
    // named one by one. A flatten used to state a zero `TargetPosition` and let reconciliation do
    // the cancelling, which was wrong twice over -- reconciliation keeps a resting order that points
    // the same way as the target, so the flatten could send nothing at all, and it skipped orders on
    // instruments with no position to target. Stating both halves here means one withdrawal per
    // resting order, no instruction derived from another, and no case where the two disagree.
    //
    // Recorded before anything is queued, because it is a statement about the picture the pass is
    // built from rather than about any instruction in it.
    report.closing_source_unverified_ = !closing_source_verified(clock_.now());
 
    // Sized from what is working, which counts an order this runner sent and holdings do not show yet
    // only once that row is rebuilt: a sell that filled while holdings lag is otherwise sold again.
    runner_->refresh_unconfirmed();
    // Counted before the pass, so its own new orders are not: what a serialized runner owes that no
    // listing or holdings explain is exposure this pass cannot withdraw or size around.
    report.closing_unconfirmed_ = runner_->obligations_unexplained();
    const auto& portfolio = runner_->context().portfolio();
    if (!stop_intent_closes_positions(intent)) {
      for (const auto& working : portfolio.working_) {
        // Already withdrawn by this runner, and the broker took the request: asked again it refuses.
        if (runner_->withdrawal_accepted_for(working)) {
          continue;
        }
        runner_->context().cancel(working.ref_, "stop-withdraw");
        ++report.withdrawn_;
      }
    } else {
      // **A flatten withdraws everything, then closes only what it cannot double-sell.**
      //
      // Two ways to get this wrong, and the library has now had both. Expressing the flatten as a
      // zero `TargetPosition` lets reconciliation treat a resting same-side order as progress and
      // send nothing at all -- a limit sell that may never fill is not being out of the market.
      // Sending the *whole* holding instead ignores that order, and since a cancellation is a
      // request rather than an outcome, the resting sell can fill while its cancellation is still
      // pending: ten long becomes ten short. The second error is much worse than the first, because
      // it manufactures a position in the opposite direction rather than leaving the one that was
      // already there.
      //
      // So the close is sized to what cannot double up: the holding minus whatever is still working
      // on the same side. Every working order is withdrawn either way, so a cancellation that
      // completes leaves part of the position open rather than a resting order nobody is watching,
      // and a cancellation that fails because the order filled leaves the account flat. Neither
      // outcome can overshoot, and the pass says it could not prove which one happened.
      for (const auto& working : portfolio.working_) {
        if (runner_->withdrawal_accepted_for(working)) {
          continue;
        }
        runner_->context().cancel(working.ref_, "stop-flatten");
        ++report.withdrawn_;
      }
      // **One close per instrument, sized from the net of its rows.** This looped over rows, and a
      // broker reporting one instrument as several lots is ordinary -- so it went wrong twice at
      // once. Rows of +100 and -40 are a holding of +60, and it submitted SELL 100 *and* BUY 40:
      // outside the session the phase rule refuses both, so the operator's flatten sent nothing
      // although SELL 60 was permitted all along. And the deduction below subtracts what is working
      // on the closing side, which is a figure for the whole instrument -- subtracted once per row,
      // it was charged twice, so rows of +60 and +40 against a working SELL 20 sent 40 and 20 and
      // left 20 of the position behind. Neither is reachable once the instrument is the unit.
      std::vector<InstrumentId> closed;
      for (const auto& position : portfolio.positions_) {
        bool seen = false;
        for (const auto& already : closed) {
          if (already == position.instrument_) {
            seen = true;
            break;
          }
        }
        if (seen) {
          continue;
        }
        closed.push_back(position.instrument_);
 
        // The net, not this row. A snapshot whose rows will not add up cannot say what flattening
        // would take, and guessing is the one thing a closing pass must not do.
        const auto net = portfolio.net_quantity_of(position.instrument_);
        if (!net) {
          // **Not `closing_contended_`.** That one means a competing order, and its note sends an
          // operator to the order list. What failed here is the holding's own arithmetic.
          ++report.closing_unsizable_;
          continue;
        }
        if (net->coefficient() == 0) {
          // Flat on the net, however many rows say so. A closed position can sit in a snapshot at
          // zero, and two rows can cancel; an order for none of it is one the broker would reject,
          // for nothing.
          continue;
        }
        if (report.closing_source_unverified_) {
          // **Nothing is placed against a listing that cannot be trusted.** The deduction below
          // subtracts what is *known* to be working on the closing side, and an unverified picture
          // is precisely one where that is not all of it: an order this runner sent after the last
          // listing, or a source that cannot enumerate orders at all. Sending a full-size close
          // beside an order nobody can see is the same double sell, reached by a different door --
          // and reporting it afterwards does not recall the instruction.
          //
          // The withdrawals above still went, because cancelling only ever reduces exposure. What
          // is left is the position, held and reported, which is the safe end of every branch here.
          ++report.closing_contended_;
          continue;
        }
        const bool held_long = decimal_less(Decimal64{}, *net);
        const auto magnitude = held_long ?
          std::optional<Decimal64>(*net) :
          decimal_subtract(Decimal64{}, *net);
        if (!magnitude) {
          // **A representable net can have no representable magnitude**: two rows summing to the
          // most negative `Decimal64` have no positive counterpart, so the size of the buy that
          // would close them cannot be stated. This used to `continue` in silence, which moved no
          // count -- so `closing_complete()` returned true over a position the pass had never
          // sized, which is the one thing that call promises it will not do.
          ++report.closing_unsizable_;
          continue;
        }
        const OrderSide closing_side = held_long ? OrderSide::SELL : OrderSide::BUY;
        const auto contending = portfolio.working_quantity(position.instrument_, closing_side);
        const auto sendable = decimal_subtract(*magnitude, contending);
        if (!sendable) {
          // The same silence for the same reason: a size that cannot be computed is not a size of
          // zero, and a pass that skipped it did not knowingly send nothing. **Unlike the two cases
          // above, this one can fail with both inputs representable** -- a held magnitude and a
          // working quantity whose scales cannot be aligned -- which is why the note names the
          // working orders as well as the holdings.
          ++report.closing_unsizable_;
          continue;
        }
        if (contending.coefficient() != 0) {
          // Sized down, and the reduction is a fact about what this pass could not establish rather
          // than about what the operator asked for. Counted so `closing_complete()` cannot claim a
          // flatten that may have left part of the position open.
          ++report.closing_contended_;
        }
        if (decimal_less(*sendable, Decimal64{}) || sendable->coefficient() == 0) {
          // Everything held is already working on the closing side. Sending anything at all here is
          // the over-close, so the withdrawal above is the whole of this instrument's flatten.
          continue;
        }
        runner_->context().submit(
          strategy::OrderIntent::market(
            position.instrument_,
            closing_side,
            strategy::SizeSpec::quantity(*sendable)
          ).with_tag("stop-flatten")
        );
        ++report.flattened_;
      }
    }
 
    // Lifted so the closing instructions can actually go out -- a withdraw that cannot send is not a
    // withdraw -- and closed again immediately afterwards. The window is exactly one `stop()` call
    // wide, and it is the only point at which a stop in progress can send anything.
    //
    // Only the loop's *own* hold lifts. An operator who halted by hand, or a runner whose journal
    // failed, still refuses every closing instruction -- placements through `RiskGuard::check` and
    // cancellations through the same holds in `Runner::send_cancel` -- and the closing counts say
    // so, so `closing_complete()` is false and the notes say the position is still held. A stop
    // arriving by signal or control file therefore cannot send anything through a halt somebody had
    // already set.
    hold(std::nullopt);
    // **Derived once more, immediately before the only call that can send during a stop.** `on_stop` ran
    // a few lines above with a mutable context, so it may have revised the calendar -- and unlike a
    // strategy's write inside `on_phase`, this one the loop *can* answer, because `settle_closing` is a
    // separate call it makes itself. Without this, a handler that revised the calendar to closed
    // saw the flatten go out under the `OPEN` it replaced.
    //
    // At `clock_.now()` rather than the iteration's instant, because a stop is honoured after the
    // iteration and a closing pass should be checked against the session as it is now.
    const auto closing_at = clock_.now();
    runner_->context().set_phase(phase_at(closing_at));
    const auto watch = PhaseWatch{closing_at, phase_at(closing_at), runner_->guard().total_sent()};
    const auto closing = runner_->settle_closing();
    note_revised_phase("closing pass", watch, report);
    hold(strategy::HaltReason::STOPPING);
    // What the pass itself caused to be kept (a close the phase or the broker's hours refused) is not to
    // go out later either: a stop makes each request once.
    runner_->clear_retained_targets();
    runner_->clear_deferred_targets();
 
    report.closing_placed_ = closing.would_place_.size();
    // **Both halves of every bucket.** A closing instruction is a placement or a cancellation, and
    // each can fail in the same five ways; counting only the placement half is how a stop that sent
    // nothing reported itself complete.
    report.closing_rejected_ = closing.rejected_.size() + closing.rejected_cancel_.size();
    report.closing_refused_ = closing.refused_.size() + closing.refused_cancel_.size();
    // Whole plans too: a serialized runner withholds a closing order's plan while the instrument owes one.
    report.closing_withheld_ = closing.withheld_.size() + closing.withheld_plans_.size();
    report.closing_indeterminate_ =
      closing.indeterminate_.size() + closing.indeterminate_cancel_.size();
    report.closing_unavailable_ = closing.unavailable_.size() + closing.unavailable_cancel_.size();
    report.closing_unresolved_ = closing.unresolved_cancel_.size();
 
    // Named individually rather than summarized, because "your flatten did not go out" is the one
    // note in this whole report an operator must not have to infer.
    if (!closing.refused_.empty() || !closing.refused_cancel_.empty()) {
      report.notes_.push_back(
        "a guard refused " + std::to_string(report.closing_refused_) +
        " closing instruction(s); the position is still held");
    }
    if (!closing.rejected_.empty()) {
      // Deliberately does not name who decided. The bucket holds three things -- reconciliation
      // could not build the instruction, the broker refused it, or the binding refused to send it
      // -- and an operator reading a stop report needs the consequence, which is the same for all
      // three. The report's own entries carry which it was, on `RejectReason` for a placement and
      // on `by_broker_` for a cancellation.
      report.notes_.push_back(
        std::to_string(closing.rejected_.size()) +
        " closing instruction(s) were not placed; the position is still held");
    }
    if (!closing.rejected_cancel_.empty()) {
      report.notes_.push_back(
        std::to_string(closing.rejected_cancel_.size()) +
        " closing cancellation(s) were refused; read the broker's order list for their state");
    }
    if (report.closing_withheld_ != 0) {
      report.notes_.push_back(
        "withheld " + std::to_string(report.closing_withheld_) +
        " closing instruction(s) pending reconciliation, or until an earlier order is accounted for;"
        " read the broker's order list");
    }
    if (report.closing_unconfirmed_ != 0) {
      report.notes_.push_back(
        std::to_string(report.closing_unconfirmed_) +
        " order(s) this runner sent were not yet accounted for by holdings or a listing, so the closing pass"
        " could not withdraw or size around them; read the broker's order list and holdings");
    }
    if (report.closing_wait_expired_) {
      report.notes_.push_back(
        "the stop waited " +
        std::to_string(std::chrono::duration_cast<std::chrono::seconds>(report.closing_waited_).count()) +
        " s for holdings and the order list to account for this runner's orders, or for a marked"
        " instrument to be resolved, and that did not happen; the closing pass was built from what was in"
        " hand (set_closing_wait)");
    }
    if (report.closing_indeterminate_ != 0) {
      report.notes_.push_back(
        std::to_string(report.closing_indeterminate_) +
        " closing instruction(s) have an unknown outcome; read the broker's order list");
    }
    if (report.closing_unavailable_ != 0) {
      report.notes_.push_back(
        "local machinery could not send " + std::to_string(report.closing_unavailable_) +
        " closing instruction(s); no broker was asked and the position is still held");
    }
    if (report.stop_handler_refused_ != 0) {
      report.notes_.push_back(
        "the strategy's on_stop asked for " + std::to_string(report.stop_handler_refused_) +
        " instruction(s) during the stop; none were sent, because an operator's stop is not a"
        " moment for a strategy to trade");
    }
    if (report.closing_contended_ != 0) {
      report.notes_.push_back(
        report.closing_source_unverified_ ?
          "no closing order was sent for " + std::to_string(report.closing_contended_) +
            " position(s), because an order this loop cannot see could close them too; they are"
            " still held and the withdrawals above are all that went" :
          std::to_string(report.closing_contended_) +
            " position(s) were closed only for the part no working order was already selling or"
            " buying; if those withdrawals succeed, part of the position is still held");
    }
    if (report.closing_unsizable_ != 0) {
      // **Names both inputs, because the three causes do not share one.** Two of them are the
      // holdings rows alone -- they will not add up, or the net has no representable closing
      // magnitude. The third subtracts what is already working on the closing side, and that
      // subtraction can fail with both inputs perfectly representable: a held quantity of
      // 1,000,000,000,000,000,000 and a working close of 0.1 cannot be aligned. An earlier note
      // blamed the holdings snapshot alone and would have sent an operator to check one of two
      // inputs. It still does not say "contended", which means a competing order reduced the close;
      // here nothing could be sized at all.
      report.notes_.push_back(
        "no closing order could be sized for " + std::to_string(report.closing_unsizable_) +
        " position(s): the holdings quantities would not add up or fit, or could not be combined"
        " with what is already working on the closing side; nothing was sent for them and they are"
        " still held");
    }
    if (report.closing_source_unverified_) {
      report.notes_.push_back(
        std::string("the stop could not verify what was outstanding: ") + unverified_source_reason() +
        "; orders or positions a previous process left may remain");
    }
    if (report.closing_unresolved_ != 0) {
      report.notes_.push_back(
        std::to_string(report.closing_unresolved_) +
        " closing cancellation(s) named an order the portfolio does not hold; read the broker's"
        " order list");
    }
  }
 
  // Takes `gate`'s hold on the guard and releases the one this loop took before it.
  //
  // There is no "did I halt it" flag any more, and that is the point: the loop names the hold it
  // takes and releases only that one, so an operator halting by hand into a loop's closed gate is
  // no longer invisible, and the loop reopening its gate no longer lifts their halt with it. The
  // hold is released and retaken rather than diffed, because taking a hold is idempotent.
  //
  // Only the loop's *own* conditions are held. `OPERATOR` belongs to a person and `LOCAL_FAILURE`
  // to the runner; both outlive any gate the loop computes, so they are reported and never taken or
  // released here.
  void hold(std::optional<strategy::HaltReason> gate) {
    if (gate && !owns(*gate)) {
      gate.reset();
    }
    if (held_ == gate) {
      return;
    }
    if (held_) {
      runner_->guard().resume(*held_);
    }
    if (gate) {
      runner_->guard().halt(*gate);
    }
    held_ = gate;
  }
 
  // Says what is missing for a runner that keeps its sent orders until the broker accounts for them, as a
  // note bit and the note, or nothing when the listing is in hand.
  std::optional<std::pair<unsigned, std::string>> listing_missing(
    std::chrono::system_clock::time_point at
  ) const {
    if constexpr (!source_provides_working_orders<Source>::value) {
      (void)at;
      return std::make_pair(1u, std::string(
        "this runner keeps the orders it sends until a working-order listing or holdings account for "
        "them, and the source publishes no listing: placing holds (declare provides_working_orders and "
        "poll the listing)"
      ));
    } else {
      if (!schedule_.working_orders_) {
        return std::make_pair(2u, std::string(
          "no working-order cadence: this runner keeps the orders it sends until a working-order "
          "listing or holdings account for them, so placing holds until PollSchedule::working_orders_ "
          "is set"
        ));
      }
      if (maximum_portfolio_age_ && working_orders_at_ &&
          at - *working_orders_at_ > *maximum_portfolio_age_) {
        return std::make_pair(4u, std::string(
          "working-order listing older than the portfolio age policy: poll working orders "
          "(PollSchedule::working_orders_) more often than that age, by at least a tick; placing "
          "holds until it is read"
        ));
      }
      return std::nullopt;
    }
  }
 
  // Returns the oldest order nobody can account for that was sent longer than the limit before `at`: one
  // no sweep has named, or one held for an operator (`set_owed_order_limit`). None with the limit off, or
  // from a runner that is not serialized.
  std::optional<Obligation> owed_longer_than_limit(std::chrono::system_clock::time_point at) const {
    std::optional<Obligation> oldest;
    if (owed_order_limit_ <= std::chrono::nanoseconds::zero()) {
      return oldest;
    }
    for (auto& owed : runner_->obligations()) {
      if (!obligation_is_unaccounted(owed.state_)) {
        continue;
      }
      if (at - owed.sent_at_ > owed_order_limit_ && (!oldest || owed.sent_at_ < oldest->sent_at_)) {
        oldest = std::move(owed);
      }
    }
    return oldest;
  }
 
  // Whether such an order is past the limit as of `now`: placing is held from here on.
  bool owed_past_limit(std::chrono::system_clock::time_point now) const {
    return owed_longer_than_limit(now).has_value();
  }
 
  // The order the run stops for: one still unaccounted for after the last successful order-list read,
  // where that read's iteration began longer after the order's send than the limit (`order_list_at_` is
  // the iteration's instant, which is not later than any check made in it). Measured to that read and
  // not to now, so that a loop reading the list less often than the limit gives a sweep its chance to
  // name the order first. None before the order list was read.
  std::optional<Obligation> owed_too_long() const {
    if (!order_list_read_ || !order_list_at_) {
      return std::nullopt;
    }
    return owed_longer_than_limit(*order_list_at_);
  }
 
  // The note for a run the loop stopped for `owed`.
  static std::string owed_stop_note(const Obligation& owed) {
    return "stopped placing: the " + std::string(owed.side_ == OrderSide::BUY ? "buy" : "sell") + " of " +
      owed.quantity_.str() + " " + owed.instrument_.symbol_ + " sent " + format_iso8601_utc(owed.sent_at_) +
      " (journal id " + owed.journal_id_ + ") is still owed, and " +
      (owed.state_ == ObligationState::OPERATOR ? "is held for an operator" : "no sweep has found its order") +
      ". Nothing was cancelled or closed. Read the broker's order list and holdings for it, then account for "
      "it with Runner::release_obligations; until then every start stops here again";
  }
 
  // Returns whether a stop that closes should let the loop read again before its pass is built.
  //
  // True while a read could still give the pass what it lacks and the wait has not reached its limit;
  // `STOPPING` is held throughout (`begin_stop`, `gate_for`), so the loop only reads meanwhile: what a
  // strategy asks for is refused. See `set_closing_wait` for what it waits for and what it does not.
  bool closing_should_wait(StopIntent intent, LoopReport& report) {
    if (!stop_intent_closes(intent) || closing_wait_ <= std::chrono::nanoseconds::zero()) {
      return false;
    }
    // A source that lists no working orders can never verify the pass, however long the loop reads:
    // waiting would only delay the same result.
    if constexpr (!source_provides_working_orders<Source>::value) {
      return false;
    }
    // Under a person's hold or the runner's own, the pass can send nothing whatever the reads show.
    if (runner_->guard().halted(strategy::HaltReason::OPERATOR) ||
        runner_->guard().halted(strategy::HaltReason::LOCAL_FAILURE)) {
      return false;
    }
    const auto monotonic = clock_.monotonic_now();
    report.closing_waited_ = std::chrono::duration_cast<std::chrono::nanoseconds>(monotonic - closing_->began_);
    // As the pass does before it counts: an order holdings now show is no longer owed.
    runner_->refresh_unconfirmed();
    closing_->reads_wanted_ = closing_needs_evidence(clock_.now());
    if (!closing_->reads_wanted_) {
      return false;
    }
    if (monotonic >= closing_->deadline_) {
      report.closing_wait_expired_ = true;
      closing_->reads_wanted_ = false;
      return false;
    }
    return true;
  }
 
  // Whether a further read could change what the closing pass would do: the picture it is built from is
  // not verified, an order of this runner awaits an account, or an instrument that is held is marked.
  bool closing_needs_evidence(std::chrono::system_clock::time_point at) const {
    if (!closing_source_verified(at) || runner_->obligations_awaiting_account() != 0) {
      return true;
    }
    // A mark on something held, unless only an operator can lift it. Held by the instrument's net, as the
    // pass sizes it; rows that cannot be netted count as held.
    const auto& portfolio = runner_->context().portfolio();
    for (const auto& position : portfolio.positions_) {
      const auto net = portfolio.net_quantity_of(position.instrument_);
      if ((!net || net->coefficient() != 0) && runner_->ledger().requires_reconcile(position.instrument_) &&
          !obligation_needs_operator(runner_->obligation_state(position.instrument_))) {
        return true;
      }
    }
    return false;
  }
 
  // Whether a closing wait may read `kind` (working orders or holdings) now: it wants the read, and that
  // endpoint was last asked at least `closing_reread_` ago, whoever asked -- the ordinary cadence, the
  // startup read, or this wait. The iteration in which the limit passes may still read, like the startup
  // read: the pass that follows is built from it, and nothing reads after it.
  bool closing_read_due(PollKind kind) const {
    if (!closing_ || !closing_->reads_wanted_) {
      return false;
    }
    return !asked_within_floor(kind);
  }
 
  // Whether `kind`'s endpoint (working orders or holdings) was last asked less than `closing_reread_` ago.
  bool asked_within_floor(PollKind kind) const {
    const auto& asked = kind == PollKind::WORKING_ORDERS ? working_asked_at_ : holdings_asked_at_;
    return asked && clock_.monotonic_now() - *asked < closing_reread_;
  }
 
  // Returns how long a waiting stop sleeps: until the next read it may make, its limit, an ordinary poll
  // of another kind, or the tick, whichever is first. The ordinary due times of the two closing reads are
  // left out: the wait reads them on its own floor, and an overdue one would otherwise spin the loop.
  std::chrono::nanoseconds closing_until_next() const {
    auto wait = schedule_.tick_;
    const auto wall = clock_.now();
    for (const auto kind : all_poll_kinds) {
      const auto& due = next_due_[static_cast<int>(kind)];
      if (!due || kind == PollKind::WORKING_ORDERS || kind == PollKind::HOLDINGS) {
        continue;
      }
      wait = std::min(wait, std::chrono::duration_cast<std::chrono::nanoseconds>(*due - wall));
    }
    const auto monotonic = clock_.monotonic_now();
    const auto until = [&](std::chrono::steady_clock::time_point when) {
      return std::chrono::duration_cast<std::chrono::nanoseconds>(when - monotonic);
    };
    wait = std::min(wait, until(closing_->deadline_));
    for (const auto* asked : {&working_asked_at_, &holdings_asked_at_}) {
      wait = std::min(wait, *asked ? until(**asked + closing_reread_) : std::chrono::nanoseconds::zero());
    }
    return wait > std::chrono::nanoseconds::zero() ? wait : std::chrono::nanoseconds::zero();
  }
 
  // Returns whether the loop knows what a closing pass at `at` would have to send.
  //
  // Two sources, both required. The broker's own order list must have been read once, which is what
  // the startup gate exists for, and what a stop requested before the first iteration attempts itself
  // (a failed attempt leaves it unread). And the
  // holdings snapshot -- which is where a `WITHDRAW`'s working orders and a `FLATTEN`'s positions
  // actually come from -- must exist, be within the configured age, and be **no older than the
  // working-order listing**: a fill between the two reads leaves the listing before it reaches the
  // holdings, so a flatten would see neither the order nor the shares it bought.
  bool closing_source_verified(std::chrono::system_clock::time_point at) const {
    if (!order_list_read_) {
      return false;
    }
    // The working-order picture is its own source with its own freshness, and `as_of_` is not its
    // timestamp: `SessionSource::poll_holdings` refreshes positions and `as_of_` while *keeping* the
    // working orders it already had, and only a working-order poll replaces those. So a stop can
    // have a snapshot stamped a second ago whose resting orders were last read an hour ago.
    if (!working_orders_at_) {
      return false;
    }
    // And the decisive one, which no timestamp catches: an order this runner sent after that read
    // cannot be in it. A listing is out of date the moment the next placement is accepted, and an
    // indeterminate send counts too, because the order may exist.
    if (runner_->placements_at_broker() != placements_at_working_read_) {
      return false;
    }
    if (maximum_portfolio_age_ && at - *working_orders_at_ > *maximum_portfolio_age_) {
      return false;
    }
    const auto as_of = runner_->context().portfolio().as_of_;
    if (as_of == std::chrono::system_clock::time_point{}) {
      return false;
    }
    // On one clock: both stamps are the local arrival time of their replies, which
    // `working_orders_at_` above is not.
    if (runner_->context().portfolio().listing_newer_than_holdings()) {
      return false;
    }
    return !maximum_portfolio_age_ || at - as_of <= *maximum_portfolio_age_;
  }
 
  // Records that the working-order picture was just refreshed, with what it could not contain.
  //
  // Does nothing when the source does not publish working orders, so a stop on such a broker stays
  // unverified rather than being told a history read or a no-op poll refreshed a listing it never
  // built.
  void note_working_orders_read(
    std::chrono::system_clock::time_point at, std::size_t placements_before_poll
  ) {
    if constexpr (source_provides_working_orders<Source>::value) {
      working_orders_at_ = at;
      order_list_at_ = at;
      // **The caller's count, taken before the poll ran.** Reading it here instead would include
      // an order an `on_order_update` handler placed *during* the poll -- an order sent after the
      // listing was fetched, which the listing therefore cannot contain.
      placements_at_working_read_ = placements_before_poll;
    } else {
      (void)at;
      (void)placements_before_poll;
    }
  }
 
  // Returns which half of the closing source is missing, for the operator's note.
  std::string unverified_source_reason() const {
    if constexpr (!source_provides_working_orders<Source>::value) {
      return "this source does not publish the broker's outstanding orders";
    }
    if (!order_list_read_ || !working_orders_at_) {
      return "the broker's order list has not been read";
    }
    if (runner_->placements_at_broker() != placements_at_working_read_) {
      return "an order was sent after the working orders were last read";
    }
    if (runner_->context().portfolio().as_of_ == std::chrono::system_clock::time_point{}) {
      return "no holdings snapshot has arrived";
    }
    if (runner_->context().portfolio().listing_newer_than_holdings()) {
      return "the holdings were read before the working orders, so a fill between them is in neither";
    }
    if (maximum_portfolio_age_ &&
        clock_.now() - *working_orders_at_ > *maximum_portfolio_age_) {
      return "the working orders were read longer ago than the configured maximum";
    }
    return "the holdings snapshot is older than the configured maximum";
  }
 
  // Returns whether `kind` is due at `at`.
  bool is_due(PollKind kind, std::chrono::system_clock::time_point at) const {
    // A waiting stop reads the two closing kinds on its own floor, whatever their cadence (a disabled
    // one included), and nothing reads them faster than that meanwhile. The iteration that takes the
    // stop up is not waiting yet: it reads as any other.
    if (closing_ && closing_->reads_wanted_ &&
        (kind == PollKind::WORKING_ORDERS || kind == PollKind::HOLDINGS)) {
      return closing_read_due(kind);
    }
    const auto interval = schedule_.interval_for(kind);
    if (!interval) {
      return false;
    }
    const auto& due = next_due_[static_cast<int>(kind)];
    return !due || at >= *due;
  }
 
  // Records when `kind` is next due.
  void schedule_next(PollKind kind, std::chrono::system_clock::time_point at) {
    const auto interval = schedule_.interval_for(kind);
    if (!interval) {
      return;
    }
    next_due_[static_cast<int>(kind)] =
      at + std::chrono::duration_cast<std::chrono::system_clock::duration>(*interval);
  }
 
  // Returns how long to wait before the next iteration.
  //
  // The soonest due poll, capped at the tick so a stop through the control file is noticed within
  // one tick however sparse the polling is. Never negative: a loop that has fallen behind catches up
  // rather than pausing once per overdue poll.
  std::chrono::nanoseconds until_next(std::chrono::system_clock::time_point at) const {
    auto wait = schedule_.tick_;
    for (const auto kind : all_poll_kinds) {
      const auto& due = next_due_[static_cast<int>(kind)];
      if (!due) {
        continue;
      }
      const auto remaining =
        std::chrono::duration_cast<std::chrono::nanoseconds>(*due - at);
      if (remaining < wait) {
        wait = remaining;
      }
    }
    return wait > std::chrono::nanoseconds::zero() ? wait : std::chrono::nanoseconds::zero();
  }
 
  Runner* runner_;
  Source* source_;
  PollSchedule schedule_;
  Clock clock_;
  FileStopChannel control_;
  FailurePolicy failure_policy_;
  strategy::SessionWindow session_window_{strategy::SessionWindow::krx()};
  // The phase most recently announced, which is not the same as the phase in force.
  //
  // `UNKNOWN` to begin with rather than "nothing announced yet", because that is genuinely what the
  // loop knows before its first calendar arrives -- and it means a run that starts knowing nothing and
  // then learns the market is open announces that as a change, which is exactly right.
  strategy::MarketPhase announced_{strategy::MarketPhase::UNKNOWN};
  // Whether this loop has ever derived a phase that was not `UNKNOWN`, across every `run`.
  bool phase_ever_known_{false};
  std::optional<std::chrono::nanoseconds> maximum_portfolio_age_;
  // How long a stop that closes may wait for the reads that account for this runner's orders, and how
  // often it repeats them.
  std::chrono::nanoseconds closing_wait_{std::chrono::seconds(30)};
  std::chrono::nanoseconds owed_order_limit_{std::chrono::minutes(10)};
  static constexpr std::chrono::seconds closing_reread_{5};
  // A stop in progress in this run: when it was entered and when its wait ends, on the monotonic reading,
  // and whether the wait currently wants the two closing reads.
  struct ClosingWait {
    std::chrono::steady_clock::time_point began_{};
    std::chrono::steady_clock::time_point deadline_{};
    bool reads_wanted_{false};
  };
  std::optional<ClosingWait> closing_;
  // When each closing read's endpoint was last asked, on the monotonic reading: by its ordinary poll, by
  // the startup order-list read (the working-order endpoint), or by a closing wait.
  std::optional<std::chrono::steady_clock::time_point> working_asked_at_;
  std::optional<std::chrono::steady_clock::time_point> holdings_asked_at_;
  // Whether the missing-band-policy note has been said; it is said once per loop.
  bool said_no_band_age_policy_{false};
  // Whether the operator stated that any snapshot age is acceptable; see the setter.
  bool allow_unbounded_portfolio_age_{false};
  // Whether the operator stated that this run may act with no snapshot at all; see the setter.
  bool allow_absent_portfolio_{false};
  // Whether the missing-age-policy note has been reported, so it is said once and not per poll.
  bool said_no_age_policy_{false};
  // Which of `listing_missing`'s notes have been given, so each is said once per loop.
  unsigned said_listing_{0};
  // The runner's discarded-history count when this run began; see `note_warmup_discarded`.
  std::size_t warmup_discarded_at_start_{0};
  // The runner's dropped-kept-intent count when this run began; see `note_kept_intents_dropped`.
  std::size_t kept_dropped_at_start_{0};
  std::optional<std::chrono::nanoseconds> maximum_buying_power_age_;
  std::optional<std::chrono::system_clock::time_point> next_due_[poll_kind_count];
  std::size_t consecutive_failures_[poll_kind_count] = {};
  bool order_list_read_{false};
  // Returns whether `reason` is a condition this loop computes, and may therefore hold.
  static constexpr bool owns(strategy::HaltReason reason) noexcept {
    switch (reason) {
      // A person's hold and a local failure are released by their owners, never by the loop.
      case strategy::HaltReason::OPERATOR:
      case strategy::HaltReason::LOCAL_FAILURE:
        return false;
      case strategy::HaltReason::STOPPING:
      case strategy::HaltReason::ORDER_LIST_UNREAD:
      case strategy::HaltReason::PORTFOLIO_ABSENT:
      case strategy::HaltReason::PORTFOLIO_STALE:
      case strategy::HaltReason::BUYING_POWER_UNREAD:
      case strategy::HaltReason::POLL_FAILING:
        return true;
    }
    return false;
  }
 
  // When the working-order picture was last refreshed, or nothing while it never has been.
  std::optional<std::chrono::system_clock::time_point> working_orders_at_;
  // The instant of the iteration in which the source last read the broker's order list, of either kind:
  // a working-order listing, or the startup read of a source without that inquiry (which reads order
  // history for its journal). The sweep that can name an order runs inside that read.
  std::optional<std::chrono::system_clock::time_point> order_list_at_;
  // `Runner::placements_at_broker()` as of that read, so a later send invalidates it.
  std::size_t placements_at_working_read_{0};
 
  // The hold this loop currently owns, and the only one it will ever release.
  std::optional<strategy::HaltReason> held_;
  IterationReport last_;
};
 
} // namespace runner
} // namespace algotradelib