apply_record reports APPLY_UPDATED at LWW parity, so replay_healed can never reach zero

[gregoryfoster]

Symptom

Twelve consecutive hourly replay passes reported a byte-identical result:

04:26:09  applied=3963 healed=3290
05:26:36  applied=3963 healed=3290
06:26:38  applied=3963 healed=3290
07:27:04  applied=3963 healed=3290
08:27:30  applied=3963 healed=3290
09:27:56  applied=3963 healed=3290
10:28:56  applied=3964 healed=3291
12:05:04  applied=3965 healed=3292
13:04:07  applied=3965 healed=3292
14:04:52  applied=3965 healed=3292
15:09:59  applied=3963 healed=3290
16:10:29  applied=3963 healed=3290

3,290 rows "healed" every hour, forever, with no convergence.

Cause

engine/read.py::apply_record returns APPLY_UPDATED on the PM-wins branch unconditionally, without comparing any field values:

lu_local = descriptor.last_updated(existing)
lu_pm = descriptor.last_updated(record)
if lu_local is not None and lu_pm is not None and lu_local > lu_pm:
    ...
    return APPLY_KEPT_LOCAL

row = await descriptor.upsert_from_pm(session, record, existing=existing)
self._anchors.adopt_remote_clock(descriptor, row, record)
return APPLY_UPDATED      # <-- no diff was performed

The guard is strictly >, so an equal clock falls through to the PM-wins branch. And equality is exactly the converged steady state: adopt_remote_clock deliberately mirrors PM's updated_at onto the local row so that "a freshly-cached row must not read as locally-newer" (engine/anchors.py). descriptor.last_updated reads updated_at on both sides.

So for every already-converged row: lu_local == lu_pm → tie → re-upsert → APPLY_UPDATED → counted in healed. Every pass. Permanently.

_apply_feed_page then does if outcome in (APPLY_INSERTED, APPLY_UPDATED): healed += 1.

Why it matters

sidecar.py documents replay_healed as the backstop's headline signal — "replay_healed > 0 = replay recovered events the live feed dropped". That signal is structurally incapable of reaching zero on a healthy system, so it detects nothing and cannot be alerted on. It is also why the runaway in #211 was invisible in the logs for 19 hours: the backstop looked like it was doing 3,290 useful repairs an hour while doing none.

Cost side: each tie also performs a real upsert_from_pm write of identical values — ~3,290 no-op row writes per pass.

Prior art in the same file

engine/anchors.py already fixed the analogous bug one level down. adopt_remote_clock skips the stamp at parity precisely because flagging unconditionally "turned each already-converged row into a no-op UPDATE writing an identical value (~12.7k/day across the anchored cohorts)". The outcome classification in apply_record never received the same parity treatment.

Suggested fix

Distinguish "PM record differed and we changed a column" from "re-upserted identical values" — e.g. have upsert_from_pm report whether any attribute actually changed (SQLAlchemy attribute history / session.is_modified) and return a new APPLY_NOOP outcome at parity. Then:

• healed counts only genuine repairs and can legitimately be zero (and alertable).
• The identical-value write disappears.
• The tie branch can skip the upsert entirely.

Note this is the read path and is distinct from #117, which covers write-path non-convergence (PM auto-attached re-sends).

Found while investigating #211; the sidecar is stopped and held pending both.

[gregoryfoster]

Shipped in batch/a → main (92bd3cc), orchestrated under #218. Gate: full suite 2307 passed (coverage 97.91%), integration tier 3 passed exit 0, two-session lock proof live.