fix(engine): close #731 and #732 — pin two ON CONFLICT windows, and deliver the finding nobody could receive

keel/notifications.py

@@ -108,8 +108,13 @@ def events_from_state(
     """
     events: list[NotificationEvent] = []
 
+    notified: set[str] = set()
     for finding in attestation_findings:
-        if finding.name in _ATTESTATION_FINDINGS and finding.status in (doctor.WARN, doctor.FAIL):
+        if (
+            finding.name in _ATTESTATION_FINDINGS
+            and finding.name not in notified
+            and finding.status in (doctor.WARN, doctor.FAIL)
+        ):
             events.append(
                 notification_event(
                     "attestation.expiring",
@@ -119,7 +124,13 @@ def events_from_state(
                     detail=finding.detail,
                 )
             )
-            break  # one event per cycle: the finding list carries one rail-17 verdict
+            # One event PER FINDING NAME, not one per cycle. This was a `break`, written when the
+            # registry held rail 17 alone and the list genuinely carried one verdict. With rail 22
+            # in it (#732) a break makes a cash-posture problem invisible whenever a withdrawals
+            # problem also exists -- the same silence, one layer down. The names are unique across
+            # the gatherers, so the guard below is about not repeating one, never about choosing
+            # between two.
+            notified.add(finding.name)
 
     for finding in rail_findings:
         if finding.name in _ARMED_RAIL_FINDINGS and finding.status != doctor.OK:
@@ -229,11 +240,26 @@ def notify_after_cycle(
 
         venue = current_venue() or guards.DEFAULT_VENUE
         subscription = repo.get_broker_subscription(venue)
-        attestation = doctor.attestation_findings(
-            subscription=subscription,
-            withdrawals_attested_at=int(repo.get_state("withdrawals_attested_at", default=0) or 0),
-            now_ts=now_ts,
-        )
+        attestation = [
+            *doctor.attestation_findings(
+                subscription=subscription,
+                withdrawals_attested_at=int(
+                    repo.get_state("withdrawals_attested_at", default=0) or 0
+                ),
+                now_ts=now_ts,
+            ),
+            # #732. `attest.cash_posture` was in `_ATTESTATION_FINDINGS` and this call was not
+            # here, so the registration was real and the delivery path was not: an operator who
+            # wired a webhook for it would never have been told.
+            #
+            # It is the finding that fires when the account is attested MARGIN-ENABLED, when the
+            # posture attestation has expired, or when it was attested with no due date at all --
+            # three states in which rail 22 has stopped letting the agent enter positions. The
+            # symptom otherwise is SILENCE: an agent that looks healthy and never trades again.
+            *doctor.cash_posture_findings(
+                repo.get_venue_cash_posture(venue), venue=venue, now_ts=now_ts
+            ),
+        ]
         rails = doctor.rail_state_findings(
             kill_switch=bool(repo.get_state("kill_switch", default=False)),
             streak_halt_until=int(repo.get_state("streak_halt_until", default=0) or 0),

tests/data/test_repository.py

@@ -798,3 +798,154 @@ def test_get_screen_exceptions_is_scoped_to_the_asset(repo):
 
     assert repo.get_screen_exceptions("PAXG") == {"history": "paxg reason"}
     assert "SOL" not in repo.get_screen_exceptions("PAXG")
+
+
+# -- the two remaining ON CONFLICT windows (#731) -------------------------------------------------
+#
+# `broker_subscriptions` and `venue_cash_postures` both carry `attest_due_ts` and both are written
+# by an `INSERT ... ON CONFLICT DO UPDATE SET`. Both clauses set it, so both are correct -- and
+# deleting either line left the whole suite green, which is the same hole #718 closed for the asset
+# and instrument tables and left open here.
+#
+# `venue_cash_postures.attest_due_ts` is RAIL 22's INPUT. A re-attestation that silently carried
+# the old window forward would keep an expired posture reading as current, and rail 22 would stop
+# vetoing when it should -- the failure direction that costs money rather than opportunity.
+
+
+def _subscription(**overrides):
+    from decimal import Decimal
+
+    from keel_core.subscription import BrokerSubscription, SubscriptionStatus
+
+    fields = {
+        "venue": "coinbase",
+        "tier_name": "advanced",
+        "free_volume_usd": Decimal("10000"),
+        "pacing": "monthly",
+        "subscription_usd_month": Decimal("30"),
+        "status": SubscriptionStatus.ACTIVE,
+        "attested_at": 1_000,
+        "attest_due_ts": 2_000,
+    }
+    fields.update(overrides)
+    return BrokerSubscription(**fields)
+
+
+def _posture(**overrides):
+    from keel_core.cash_posture import CashPostureState, VenueCashPosture
+
+    fields = {
+        "venue": "coinbase",
+        "state": CashPostureState.ATTESTED,
+        "attested_posture": "SPOT_CASH",
+        "attested_ts": 1_000,
+        "attest_due_ts": 2_000,
+        "refuted_ts": None,
+        "refuted_reason": None,
+        "credential_fingerprint": "fp-1",
+    }
+    fields.update(overrides)
+    return VenueCashPosture(**fields)
+
+
+def test_reattesting_a_subscription_overwrites_its_window(repo):
+    """`attest_due_ts` is NOT NULL here, so the slip is even less visible than on the nullable
+    columns: the window can never read as absent, only as stale."""
+    repo.upsert_broker_subscription(_subscription(attest_due_ts=2_000))
+    assert repo.get_broker_subscription("coinbase").attest_due_ts == 2_000
+
+    repo.upsert_broker_subscription(_subscription(attested_at=3_000, attest_due_ts=9_000))
+    assert repo.get_broker_subscription("coinbase").attest_due_ts == 9_000
+
+
+def test_reattesting_a_subscription_does_not_clobber_its_other_columns(repo):
+    """The other half of the ON CONFLICT check: a clause that set the window and dropped a
+    neighbour would pass the test above."""
+    from decimal import Decimal
+
+    repo.upsert_broker_subscription(_subscription())
+    repo.upsert_broker_subscription(
+        _subscription(tier_name="pro", free_volume_usd=Decimal("50000"), attest_due_ts=9_000)
+    )
+
+    stored = repo.get_broker_subscription("coinbase")
+    assert stored.tier_name == "pro"
+    assert stored.free_volume_usd == Decimal("50000")
+    assert stored.attest_due_ts == 9_000
+
+
+def test_reattesting_a_cash_posture_overwrites_its_window(repo):
+    """RAIL 22'S INPUT. `doctor.cash_posture_findings` FAILS an expired posture and rail 22 vetoes
+    live entries on it -- so a window carried forward through a re-attestation is a rail that
+    stops vetoing when it should."""
+    repo.upsert_venue_cash_posture(_posture(attest_due_ts=2_000))
+    assert repo.get_venue_cash_posture("coinbase").attest_due_ts == 2_000
+
+    repo.upsert_venue_cash_posture(_posture(attested_ts=3_000, attest_due_ts=9_000))
+    assert repo.get_venue_cash_posture("coinbase").attest_due_ts == 9_000
+
+
+def test_a_cash_posture_reattested_with_no_window_does_not_inherit_the_old_one(repo):
+    """`VenueCashPosture`'s own docstring: "a record with no due date is a claim that never
+    expires, which this record does not permit". A NULL that inherited the previous window would
+    be exactly the claim it refuses, wearing a date nobody stated this time."""
+    repo.upsert_venue_cash_posture(_posture(attest_due_ts=2_000))
+    repo.upsert_venue_cash_posture(_posture(attested_ts=3_000, attest_due_ts=None))
+
+    assert repo.get_venue_cash_posture("coinbase").attest_due_ts is None
+
+
+def test_reattesting_a_cash_posture_CLEARS_a_previous_refutation(repo):
+    """Every nullable column on this row has to be settable back to NULL by a re-attestation, and
+    the refutation columns are the ones that matter: a posture the venue refuted, then re-attested
+    after the operator fixed it, would otherwise keep reading as refuted forever.
+
+    The method's own docstring states this rule for `credential_fingerprint` -- "could not clear
+    it would let a stale fingerprint outlive the record it described" -- and the same is true of
+    `refuted_ts` and `refuted_reason`. Dropping any of the three from `DO UPDATE SET` left the
+    suite green.
+    """
+    from keel_core.cash_posture import CashPostureState
+
+    repo.upsert_venue_cash_posture(
+        _posture(
+            state=CashPostureState.REFUTED,
+            refuted_ts=1_500,
+            refuted_reason="INTX portfolio present",
+        )
+    )
+    stored = repo.get_venue_cash_posture("coinbase")
+    assert stored.refuted_ts == 1_500
+
+    repo.upsert_venue_cash_posture(
+        _posture(attested_ts=3_000, attest_due_ts=9_000, refuted_ts=None, refuted_reason=None)
+    )
+
+    stored = repo.get_venue_cash_posture("coinbase")
+    assert stored.refuted_ts is None
+    assert stored.refuted_reason is None
+    assert stored.state == CashPostureState.ATTESTED
+
+
+def test_reattesting_a_cash_posture_does_not_clobber_its_other_columns(repo):
+    repo.upsert_venue_cash_posture(_posture())
+    repo.upsert_venue_cash_posture(
+        _posture(attest_due_ts=9_000, credential_fingerprint="fp-2", refuted_reason=None)
+    )
+
+    stored = repo.get_venue_cash_posture("coinbase")
+    assert stored.attest_due_ts == 9_000
+    assert stored.credential_fingerprint == "fp-2"
+
+
+def test_reattesting_a_cash_posture_overwrites_the_posture_ITSELF(repo):
+    """The column the rail actually reads. `attested_posture` is `SPOT_CASH` or `MARGIN_ENABLED`,
+    and rail 22 vetoes on the second -- so an operator who re-attests after moving off margin, and
+    whose re-attestation silently kept `MARGIN_ENABLED`, would go on being vetoed with a record
+    saying the opposite of what they stated. Every other column on this row was pinned and this
+    one was not."""
+    repo.upsert_venue_cash_posture(_posture(attested_posture="MARGIN_ENABLED"))
+    assert repo.get_venue_cash_posture("coinbase").attested_posture == "MARGIN_ENABLED"
+
+    repo.upsert_venue_cash_posture(_posture(attested_ts=3_000, attested_posture="SPOT_CASH"))
+    assert repo.get_venue_cash_posture("coinbase").attested_posture == "SPOT_CASH"

tests/test_notifications.py

@@ -24,6 +24,7 @@
 from keel.commands.doctor import attestation_findings, rail_state_findings
 from keel.config import AutoTradeConfig, Caps, Config, MarketDataConfig
 from keel.notifications import (
+    _ATTESTATION_FINDINGS,
     ALLOWANCE_NEARING_USED_PCT,
     UnplacedSetup,
     events_from_state,
@@ -226,6 +227,10 @@ def __init__(self, *, withdrawals_attested_at: int, held: tuple[str, ...] = ())
         self._withdrawals_attested_at = withdrawals_attested_at
         self._held = held
         self.state_writes: list[tuple[str, object]] = []
+        #: #732's read. `None` is the unattested posture, which `cash_posture_findings` reports
+        #: as its own finding -- so the default here is a book where nobody has attested rather
+        #: than one where the question is not asked.
+        self.cash_posture: object | None = None
 
     def get_state(self, key: str, default: object = None) -> object:
         if key == "withdrawals_attested_at":
@@ -241,6 +246,9 @@ def get_state(self, key: str, default: object = None) -> object:
     def get_broker_subscription(self, venue: str):  # None: rail 14 is out of scope here
         return None
 
+    def get_venue_cash_posture(self, venue: str):
+        return self.cash_posture
+
     def held_products(self) -> list[str]:
         return list(self._held)
 
@@ -271,6 +279,12 @@ def _transport(url: str, body: bytes) -> None:
 def test_notify_after_cycle_reads_doctor_seams_and_sends_only_opted_in_events():
     calls: list[tuple[str, str]] = []
     repo = _Repo(withdrawals_attested_at=NOW - 5 * DAY)  # rail 17: 2 days remain
+    # A HEALTHY posture, so this test stays about rail 17 alone. #732 wired
+    # `cash_posture_findings` into the same path, and the default double carries no posture at
+    # all -- which `doctor` reports as "cash posture never attested", a FAIL, because rail 22
+    # vetoes on it. That is a real second event, not a fixture artefact, and it belongs to the
+    # tests below rather than to this one.
+    repo.cash_posture = _healthy_posture()
     settings = NotificationSettings(events=frozenset({"attestation.expiring"}))
     config = _config_with(settings)
 
@@ -418,3 +432,140 @@ def _config_with(settings: NotificationSettings) -> Config:
         auto_trade=AutoTradeConfig(),
         notifications=settings,
     )
+
+
+def _healthy_posture():
+    """An attested, in-date cash posture -- rail 22 quiet."""
+    from keel_core.cash_posture import CashPostureState, VenueCashPosture
+
+    return VenueCashPosture(
+        venue="coinbase",
+        state=CashPostureState.ATTESTED,
+        attested_posture="SPOT_CASH",
+        attested_ts=NOW - DAY,
+        attest_due_ts=NOW + 200 * DAY,
+        refuted_ts=None,
+        refuted_reason=None,
+        credential_fingerprint="fp-1",
+    )
+
+
+# -- the registry and the call site must agree (#732) ----------------------------------------------
+
+
+def _lapsed_posture_repo(*, withdrawals_attested_at: int) -> _Repo:
+    """A book where BOTH registered attestation findings are unhealthy at once."""
+    from keel_core.cash_posture import CashPostureState, VenueCashPosture
+
+    repo = _Repo(withdrawals_attested_at=withdrawals_attested_at)
+    repo.cash_posture = VenueCashPosture(
+        venue="coinbase",
+        state=CashPostureState.ATTESTED,
+        attested_posture="SPOT_CASH",
+        attested_ts=NOW - 200 * DAY,
+        attest_due_ts=NOW - DAY,  # lapsed
+        refuted_ts=None,
+        refuted_reason=None,
+        credential_fingerprint="fp-1",
+    )
+    return repo
+
+
+def test_every_registered_attestation_finding_is_actually_deliverable():
+    """THE CLASS, not the instance, and driven through the REAL `notify_after_cycle`.
+
+    `_ATTESTATION_FINDINGS` is an opt-in registry; the call site is a hand-written list of doctor
+    gatherers. Two lists that must agree with nothing making them agree -- and they did not:
+    `attest.cash_posture` was registered and never produced, so an operator who wired a webhook
+    for it would never have been told.
+
+    That matters more than a missing warning. It fires when the account is attested
+    MARGIN-ENABLED, when the posture attestation has expired, or when it was attested with no due
+    date at all -- three states in which rail 22 has stopped letting the agent enter positions,
+    where the symptom otherwise is SILENCE.
+
+    A test asserting the two lists match by name would be a third list. This makes every
+    registered finding unhealthy at once and asserts each one ARRIVES, so a registration with no
+    gatherer fails here rather than in production quiet.
+    """
+    calls: list[tuple[str, str]] = []
+    repo = _lapsed_posture_repo(withdrawals_attested_at=NOW - 5 * DAY)
+    config = _config_with(NotificationSettings(events=frozenset({"attestation.expiring"})))
+
+    notify_after_cycle(
+        repo,
+        config,
+        _LoopResult(),
+        NOW,
+        url="https://alerts.example/hook",
+        transport=_recording_transport(calls),
+    )
+
+    delivered = " ".join(body for _url, body in calls)
+    for name in _ATTESTATION_FINDINGS:
+        assert name in delivered, f"{name} is registered and nothing delivers it"
+
+
+def test_both_attestation_rails_are_notified_when_both_are_unhealthy():
+    """The `break` said "one event per cycle: the finding list carries one rail-17 verdict" --
+    true when the registry held rail 17 alone. With rail 22 in it, a break makes a cash-posture
+    problem invisible whenever a withdrawals problem also exists: the same silence, one layer
+    down."""
+    calls: list[tuple[str, str]] = []
+    repo = _lapsed_posture_repo(withdrawals_attested_at=NOW - 5 * DAY)
+    config = _config_with(NotificationSettings(events=frozenset({"attestation.expiring"})))
+
+    sent = notify_after_cycle(
+        repo,
+        config,
+        _LoopResult(),
+        NOW,
+        url="https://alerts.example/hook",
+        transport=_recording_transport(calls),
+    )
+
+    assert sent == 2, f"expected one event per unhealthy rail, got {sent}"
+    delivered = " ".join(body for _url, body in calls)
+    assert "rail 17" in delivered
+    assert "rail 22" in delivered
+
+
+def test_a_healthy_cash_posture_notifies_nothing():
+    """The other direction: an alert that fired on a healthy posture is the alert nobody reads."""
+    calls: list[tuple[str, str]] = []
+    repo = _Repo(withdrawals_attested_at=NOW - DAY)  # rail 17 comfortably in date
+    repo.cash_posture = _healthy_posture()
+    config = _config_with(NotificationSettings(events=frozenset({"attestation.expiring"})))
+
+    sent = notify_after_cycle(
+        repo,
+        config,
+        _LoopResult(),
+        NOW,
+        url="https://alerts.example/hook",
+        transport=_recording_transport(calls),
+    )
+
+    assert sent == 0
+    assert calls == []
+
+
+def test_a_deployment_that_never_attested_a_posture_is_told(monkeypatch):
+    """`cash_posture_findings(None)` is a FAIL -- "cash posture never attested" -- and rail 22
+    vetoes live entries on it. This is the standing case #732 is really about: nothing lapsed,
+    nothing broke, the agent simply cannot enter and had no way to say so."""
+    calls: list[tuple[str, str]] = []
+    repo = _Repo(withdrawals_attested_at=NOW - DAY)  # rail 17 fine; no posture at all
+    config = _config_with(NotificationSettings(events=frozenset({"attestation.expiring"})))
+
+    sent = notify_after_cycle(
+        repo,
+        config,
+        _LoopResult(),
+        NOW,
+        url="https://alerts.example/hook",
+        transport=_recording_transport(calls),
+    )
+
+    assert sent == 1
+    assert "rail 22" in calls[0][1]