feat(clustering): mark un-clusterable-fit rows in the contract_anomaly feed

services/clustering/backend/app/service/pipeline.py

@@ -191,20 +191,16 @@ def on_download(msg: str) -> None:
             "run_id"
         ]
 
-        # Surface this fit's flagged verdicts to the TMS as contract_anomaly
-        # rows (the host_ch sidecar path; a non-host_ch source would skip this).
-        if get_settings().host_backed:
-            from app.service.publish import publish_contract_anomaly
-
-            publish_contract_anomaly(repo, target, network=get_settings().cardano_network)
-
         # Clusterability of THIS fit: the fraction of the training window that
         # landed in some cluster (1 - n_noise/n_points). Persisted so the scheduler
         # and UI can tell "the shape does not cluster at these params" (a re-fit is
         # futile, drift is structural) from "genuinely drifted" (a re-fit
         # converges). Computed from the in-hand cluster dict, no extra query.
         # n_points here is >= _MIN_TXS_FOR_ANALYSIS (checked above), so it is never
         # 0; guarded anyway. last_fit_at marks this fit for the anti-flap cadence.
+        # Saved BEFORE the publish below so the contract_anomaly rows this fit emits
+        # are stamped with THIS fit's coverage (publish reads the row), not the
+        # previous fit's.
         n_points = int(cluster.get("n_points") or 0)
         n_noise = int(cluster.get("n_noise") or 0)
         fit_coverage = 1.0 - (n_noise / n_points) if n_points else 0.0
@@ -215,6 +211,14 @@ def on_download(msg: str) -> None:
             last_fit_at=int(time.time()),
         )
         repo.save_contract(contract)
+
+        # Surface this fit's flagged verdicts to the TMS as contract_anomaly
+        # rows (the host_ch sidecar path; a non-host_ch source would skip this).
+        if get_settings().host_backed:
+            from app.service.publish import publish_contract_anomaly
+
+            publish_contract_anomaly(repo, target, network=get_settings().cardano_network)
+
         set_stage("done", f"{n} txs · shape cluster + shape/graph anomaly", txs_done=n)
         return result
     except Exception as exc:

services/clustering/backend/app/service/publish.py

@@ -43,6 +43,7 @@
 from datetime import UTC, datetime, timedelta
 from typing import Any
 
+from app.config import _COVERAGE_UNKNOWN, get_settings
 from app.service.verdicts import (
     VERDICT_ANOMALY,
     VERDICT_MALICIOUS,
@@ -92,6 +93,10 @@ def _host_known_only(repo: Repo, target: str, hashes: set[str]) -> set[str]:
     "verdict",
     "model_id",
     "feature_set",
+    # 012: host-feed grouping marker (1 = row came from an un-clusterable fit).
+    # Evidence only; never changes which rows publish. Kept adjacent to
+    # feature_set so this list mirrors the physical table column order.
+    "unclusterable_fit",
     "evidence",
     "scored_at",
     "published_at",
@@ -149,6 +154,8 @@ def _publish_batch(
     network: str,
     feature_set: str,
     published_at: datetime,
+    *,
+    unclusterable: int = 0,
 ) -> set[str]:
     """Resolve and publish the canonical (System) batch fit's flagged verdicts.
     Returns the set of tx_hashes published (empty if the target has no System run
@@ -212,6 +219,7 @@ def _publish_batch(
                 verdict,
                 model_id,
                 feature_set,
+                unclusterable,
                 "{}",
                 stamp,
                 published_at,
@@ -232,6 +240,8 @@ def _publish_online(
     network: str,
     feature_set: str,
     published_at: datetime,
+    *,
+    unclusterable: int = 0,
 ) -> set[str]:
     """Copy the target's flagged ``tx_classifications`` (incrementally-scored new
     txs) into ``tx_contract_anomaly``, in-database (both in ``tms_clustering``).
@@ -251,7 +261,13 @@ def _publish_online(
     ``published_at`` is the reconciliation version (see the table doc)."""
     db = repo._db  # type: ignore[attr-defined]
     verdicts = ", ".join(f"'{v}'" for v in _PUBLISHED)
-    params = {"net": network, "tgt": target, "fs": feature_set, "pub": published_at}
+    params = {
+        "net": network,
+        "tgt": target,
+        "fs": feature_set,
+        "pub": published_at,
+        "unclust": unclusterable,
+    }
     # Built by concatenation (not an f-string) so the {name:Type} server-binding
     # placeholders stay literal while db / verdicts interpolate. Active-label
     # subqueries: FINAL + deleted=0 means a cleared label no longer applies.
@@ -289,7 +305,8 @@ def _publish_online(
         "INSERT INTO " + db + ".tx_contract_anomaly (" + ", ".join(_COLUMNS) + ") "
         "SELECT {net:String} AS network, toString(tx_hash), target, cluster_id, "
         "iso_score, lof_score, consensus, votes, " + verdict_expr + ", model_id, "
-        "toString(feature_set), '{}' AS evidence, toDateTime(scored_at), "
+        "toString(feature_set), {unclust:UInt8} AS unclusterable_fit, "
+        "'{}' AS evidence, toDateTime(scored_at), "
         "{pub:DateTime64(6)} AS published_at "
         "FROM "
         + db
@@ -309,6 +326,7 @@ def _retract_stale(
     *,
     keep: set[str],
     published_at: datetime,
+    unclusterable: int = 0,
 ) -> int:
     """Append a ``normal`` tombstone for every currently-published (non-normal) tx
     of this ``(network, target, feature_set)`` that is NOT in ``keep`` (the
@@ -363,6 +381,7 @@ def _retract_stale(
             VERDICT_NORMAL,
             "",
             feature_set,
+            unclusterable,
             "{}",
             published_at,
             published_at,
@@ -385,6 +404,7 @@ def _publish_labels(
     published_at: datetime,
     *,
     exclude: set[str],
+    unclusterable: int = 0,
 ) -> set[str]:
     """Publish malicious MANUAL labels that neither the online nor batch path can
     reach, and return the hashes this target's ``keep`` set must cover on their
@@ -430,6 +450,7 @@ def _publish_labels(
                 VERDICT_MALICIOUS,
                 "",
                 feature_set,
+                unclusterable,
                 "{}",
                 published_at,
                 published_at,
@@ -462,8 +483,25 @@ def publish_contract_anomaly(
     # gets a newer published_at and wins on FINAL, regardless of source times
     # and even across a backward wall-clock step (see _reconciliation_version).
     published_at = _reconciliation_version(repo, target, network)
-    online_flagged = _publish_online(repo, target, network, feature_set, published_at)
-    batch_flagged = _publish_batch(repo, target, network, feature_set, published_at)
+    # One per-contract clusterability marker stamped on every row this pass writes
+    # (positives AND tombstones), derived once from the frozen fit's coverage: the
+    # SAME signal the scheduler and UI use, so there is no second threshold to
+    # diverge (011/012). Evidence only for host-feed grouping; it never changes a
+    # verdict, a vote, or which txs publish. A never-fit/legacy row (coverage -1)
+    # is not un-clusterable, so it stamps 0, exactly as before this column existed.
+    contract = repo.get_contract(target)
+    fit_coverage = (
+        float(contract.get("fit_coverage", _COVERAGE_UNKNOWN))
+        if contract is not None
+        else _COVERAGE_UNKNOWN
+    )
+    unclusterable = 1 if get_settings().model_unclusterable(fit_coverage) else 0
+    online_flagged = _publish_online(
+        repo, target, network, feature_set, published_at, unclusterable=unclusterable
+    )
+    batch_flagged = _publish_batch(
+        repo, target, network, feature_set, published_at, unclusterable=unclusterable
+    )
     # Malicious manual labels the online/batch paths can't reach (never-scored,
     # non-cluster txs). Folded into keep so they are not tombstoned.
     label_flagged = _publish_labels(
@@ -473,6 +511,7 @@ def publish_contract_anomaly(
         feature_set,
         published_at,
         exclude=online_flagged | batch_flagged,
+        unclusterable=unclusterable,
     )
     retracted = _retract_stale(
         repo,
@@ -481,6 +520,7 @@ def publish_contract_anomaly(
         feature_set,
         keep=online_flagged | batch_flagged | label_flagged,
         published_at=published_at,
+        unclusterable=unclusterable,
     )
     db = repo._db  # type: ignore[attr-defined]
     rows = repo.client.query(  # type: ignore[attr-defined]

services/clustering/backend/app/storage/clickhouse/base.py

@@ -157,6 +157,7 @@ def _tx_scope_params(self, target: str) -> dict[str, Any]:
         ("contracts", "target_txs"),  # 010 (per-contract read window)
         ("contracts", "fit_coverage"),  # 011 (frozen-fit clusterability)
         ("contracts", "last_fit_at"),  # 011 (anti-flap re-fit cadence)
+        ("tx_contract_anomaly", "unclusterable_fit"),  # 012 (host-feed grouping marker)
     )
 
     def missing_schema_objects(self) -> list[str]:

services/clustering/backend/tests/test_publish.py

@@ -33,6 +33,7 @@
 _VERDICT_COL = _COLUMNS.index("verdict")
 _TX_HASH_COL = _COLUMNS.index("tx_hash")
 _PUBLISHED_AT_COL = _COLUMNS.index("published_at")
+_UNCLUST_COL = _COLUMNS.index("unclusterable_fit")
 _PUB = datetime(2026, 6, 23, 12, 0, 0)  # a fixed reconciliation version for tests
 
 
@@ -63,11 +64,17 @@ def command(self, sql: str, parameters: dict[str, Any] | None = None) -> None:
         self.command_params.append(parameters or {})
 
 
-def _repo(client: FakeClient) -> Any:
+def _repo(client: FakeClient, *, fit_coverage: float = -1.0) -> Any:
     # Identity host-membership: every hash is host-known, so these tests pin
     # the reconciliation logic itself (the bound is exercised separately below).
+    # get_contract carries fit_coverage so publish_contract_anomaly can derive the
+    # unclusterable marker; the -1 default is "not yet fit" (marker 0, pre-012
+    # behaviour), so every existing test is unaffected.
     return SimpleNamespace(
-        client=client, _db="tms", host_known_tx_hashes=lambda target, hashes: set(hashes)
+        client=client,
+        _db="tms",
+        host_known_tx_hashes=lambda target, hashes: set(hashes),
+        get_contract=lambda target: {"target": target, "fit_coverage": fit_coverage},
     )
 
 
@@ -221,6 +228,68 @@ def test_publish_reconciles_then_counts(monkeypatch: pytest.MonkeyPatch) -> None
     assert "verdict != {normal:String}" in fake.queries[-1]
 
 
+# --- 012: un-clusterable marker (evidence only; never changes the published set) ---
+
+
+def test_publish_online_stamps_unclusterable_flag() -> None:
+    # The INSERT...SELECT projects the per-pass flag as a bound param, and the
+    # returned published set is unchanged by it (recall: marker suppresses nothing).
+    fake = FakeClient([[("txB",)]])
+    published = _publish_online(_repo(fake), "addr1", "preprod", "shape", _PUB, unclusterable=1)
+    assert published == {"txB"}
+    assert "unclusterable_fit" in fake.commands[0]
+    assert fake.command_params[0]["unclust"] == 1
+
+
+def test_retract_stale_stamps_unclusterable_flag() -> None:
+    fake = FakeClient([[("txC",)]])  # current published non-normal, none kept
+    _retract_stale(
+        _repo(fake), "addr1", "preprod", "shape", keep=set(), published_at=_PUB, unclusterable=1
+    )
+    _table, rows, cols = fake.inserts[0]
+    assert cols == _COLUMNS
+    assert rows[0][_UNCLUST_COL] == 1
+    assert rows[0][_VERDICT_COL] == "normal"  # still a plain tombstone
+
+
+def test_publish_labels_stamps_unclusterable_flag() -> None:
+    fake = FakeClient([[("txMANUAL",)]])
+    _publish_labels(_repo(fake), "addr1", "preprod", "shape", _PUB, exclude=set(), unclusterable=1)
+    _table, rows, _cols = fake.inserts[0]
+    assert rows[0][_UNCLUST_COL] == 1
+    assert rows[0][_VERDICT_COL] == VERDICT_MALICIOUS
+
+
+def test_publish_derives_and_forwards_unclusterable_flag(monkeypatch: pytest.MonkeyPatch) -> None:
+    # publish_contract_anomaly derives ONE 0/1 flag from the contract's fit_coverage
+    # (below MIN_CLUSTER_COVERAGE -> 1) and forwards it to every write path, without
+    # changing which txs each path returns (the flagged set is passed through).
+    seen: dict[str, int | None] = {}
+
+    def _cap(name: str, ret: Any) -> Any:
+        def _f(*_a: Any, **k: Any) -> Any:
+            seen[name] = k.get("unclusterable")
+            return ret
+
+        return _f
+
+    monkeypatch.setattr("app.service.publish._publish_online", _cap("online", {"txA"}))
+    monkeypatch.setattr("app.service.publish._publish_batch", _cap("batch", set()))
+    monkeypatch.setattr("app.service.publish._publish_labels", _cap("labels", set()))
+    monkeypatch.setattr("app.service.publish._retract_stale", _cap("retract", 0))
+
+    # Below the 0.5 floor -> un-clusterable -> flag 1 everywhere.
+    fake = FakeClient([[(datetime(2000, 1, 1),)], [(1,)]])  # version max, final count
+    publish_contract_anomaly(_repo(fake, fit_coverage=0.1), "addr1", network="preprod")
+    assert seen == {"online": 1, "batch": 1, "labels": 1, "retract": 1}
+
+    # Clusterable fit -> flag 0 everywhere (byte-identical to pre-012).
+    seen.clear()
+    fake = FakeClient([[(datetime(2000, 1, 1),)], [(1,)]])
+    publish_contract_anomaly(_repo(fake, fit_coverage=0.9), "addr1", network="preprod")
+    assert seen == {"online": 0, "batch": 0, "labels": 0, "retract": 0}
+
+
 def test_label_change_triggers_host_projection_sync(monkeypatch: pytest.MonkeyPatch) -> None:
     # On the host_ch path, applying a label must reconcile the host projection so
     # the alert is retracted/raised immediately rather than on the next re-fit.

services/clustering/clickhouse/init/012_contract_anomaly_unclusterable.sql

@@ -0,0 +1,22 @@
+-- 012: mark contract_anomaly rows that came from an un-clusterable fit.
+--
+-- `unclusterable_fit` = 1 when the row's contract has a frozen fit with no usable
+-- cluster structure (fit_coverage < MIN_CLUSTER_COVERAGE, see migration 011). Such
+-- a contract's "anomaly" verdicts are outlier-detector signals against an
+-- un-clusterable baseline, not cluster-relative anomalies, and they arrive as a
+-- high-volume degenerate-window flood. This column lets the host feed GROUP and
+-- de-prioritize that flood into an honest "no stable clusters; N low-confidence
+-- outliers" summary.
+--
+-- It is EVIDENCE ONLY, never a suppression flag: every flagged row is still
+-- published, queryable, corroboration-eligible and individually inspectable, so
+-- recall is untouched (a mismarked row is still fully visible). The publisher
+-- derives it from the SAME per-contract coverage the scheduler/UI use (one
+-- signal, no second threshold to diverge), stamped uniformly on every row a
+-- reconciliation writes (positives and tombstones alike).
+--
+-- Additive and idempotent (ADD COLUMN IF NOT EXISTS), mirroring 011: pre-existing
+-- rows default to 0 (treated clusterable) until the next publish re-stamps them.
+
+ALTER TABLE tms.tx_contract_anomaly
+    ADD COLUMN IF NOT EXISTS unclusterable_fit UInt8 DEFAULT 0 AFTER feature_set;