fix(infonet): fail closed on malformed authoritative snapshots

backend/services/infonet/markets/event_selection.py

@@ -0,0 +1,81 @@
+"""Canonical selection helpers for market events.
+
+The Infonet hashchain is already an append-only ordered sequence. Market
+application code must preserve that iteration order instead of deriving a
+second history from partially trusted ``timestamp`` or ``sequence`` fields.
+
+Authoritative event selection therefore always chooses the first matching
+event in hashchain order. Callers validate that event separately and fail
+closed; they must not skip a malformed first authoritative event in favour of
+a later replacement.
+"""
+
+from __future__ import annotations
+
+import math
+from typing import Any, Iterable
+
+
+def payload(event: dict[str, Any]) -> dict[str, Any]:
+    value = event.get("payload")
+    return value if isinstance(value, dict) else {}
+
+
+def market_id(event: dict[str, Any]) -> str:
+    return str(payload(event).get("market_id") or "")
+
+
+def finite_float(value: Any) -> float | None:
+    try:
+        parsed = float(value)
+    except (TypeError, ValueError, OverflowError):
+        return None
+    return parsed if math.isfinite(parsed) else None
+
+
+def safe_int(value: Any) -> int | None:
+    try:
+        return int(value)
+    except (TypeError, ValueError, OverflowError):
+        return None
+
+
+def has_valid_ordering(event: dict[str, Any]) -> bool:
+    """Whether an event carries usable ordering metadata."""
+    return finite_float(event.get("timestamp")) is not None and safe_int(
+        event.get("sequence")
+    ) is not None
+
+
+def events_for_market(
+    market_id_value: str,
+    chain: Iterable[dict[str, Any]],
+) -> list[dict[str, Any]]:
+    """Filter one market while preserving canonical hashchain iteration order."""
+    return [
+        event
+        for event in chain
+        if isinstance(event, dict) and market_id(event) == market_id_value
+    ]
+
+
+def first_authoritative_event(
+    events: Iterable[dict[str, Any]],
+    event_type: str,
+) -> dict[str, Any] | None:
+    """Return the first matching event without skipping malformed metadata."""
+    return next(
+        (event for event in events if event.get("event_type") == event_type),
+        None,
+    )
+
+
+__all__ = [
+    "events_for_market",
+    "finite_float",
+    "first_authoritative_event",
+    "has_valid_ordering",
+    "market_id",
+    "payload",
+    "safe_int",
+]

backend/services/infonet/markets/lifecycle.py

@@ -24,6 +24,13 @@ from enum import Enum
 from typing import Any, Iterable
 
 from services.infonet.config import CONFIG
+from services.infonet.markets.event_selection import (
+    events_for_market,
+    finite_float,
+    first_authoritative_event,
+    has_valid_ordering,
+    payload,
+)
 
 
 class MarketStatus(str, Enum):
@@ -37,24 +44,17 @@ class MarketStatus(str, Enum):
 _SECONDS_PER_HOUR = 3600.0
 
 
-def _payload(event: dict[str, Any]) -> dict[str, Any]:
-    p = event.get("payload")
-    return p if isinstance(p, dict) else {}
-
-
-def _market_id(event: dict[str, Any]) -> str:
-    return str(_payload(event).get("market_id") or "")
-
-
-def _events_for_market(market_id: str, chain: Iterable[dict[str, Any]]) -> list[dict[str, Any]]:
-    out: list[dict[str, Any]] = []
-    for ev in chain:
-        if not isinstance(ev, dict):
-            continue
-        if _market_id(ev) == market_id:
-            out.append(ev)
-    out.sort(key=lambda e: (float(e.get("timestamp") or 0.0), int(e.get("sequence") or 0)))
-    return out
+def _snapshot_time(snapshot: dict[str, Any]) -> float | None:
+    """Return a valid snapshot time without masking malformed metadata."""
+    if not has_valid_ordering(snapshot):
+        return None
+    raw_timestamp = snapshot.get("timestamp")
+    # Preserve the pre-existing fallback only for an actually absent/zero
+    # timestamp. A present-but-malformed timestamp must fail closed instead
+    # of being disguised by a valid payload frozen_at.
+    if raw_timestamp in (None, "", 0, 0.0):
+        return finite_float(payload(snapshot).get("frozen_at"))
+    return finite_float(raw_timestamp)
 
 
 def compute_market_status(
@@ -71,24 +71,31 @@ def compute_market_status(
     ``chain_majority_time(chain)``) so every node converges on the same
     status.
     """
-    events = _events_for_market(market_id, chain)
+    events = events_for_market(market_id, chain)
     if not events:
         return MarketStatus.PREDICTING  # treated as not-yet-existing
 
-    create_event = next((e for e in events if e.get("event_type") == "prediction_create"), None)
-    if create_event is None:
+    create_event = first_authoritative_event(events, "prediction_create")
+    if create_event is None or not has_valid_ordering(create_event):
         return MarketStatus.PREDICTING
 
-    finalize = next((e for e in events if e.get("event_type") == "resolution_finalize"), None)
-    if finalize is not None:
-        outcome = _payload(finalize).get("outcome")
+    # The first finalize remains authoritative by chain order. A malformed
+    # first finalize cannot exert terminal authority, and a later finalize
+    # must not replace it; projection therefore falls through to the prior
+    # snapshot-backed phase.
+    finalize = first_authoritative_event(events, "resolution_finalize")
+    if finalize is not None and has_valid_ordering(finalize):
+        outcome = payload(finalize).get("outcome")
         return MarketStatus.INVALID if outcome == "invalid" else MarketStatus.FINAL
 
-    snapshot = next((e for e in events if e.get("event_type") == "market_snapshot"), None)
+    # Likewise, never let a later snapshot replace the first commitment.
+    snapshot = first_authoritative_event(events, "market_snapshot")
     if snapshot is None:
         return MarketStatus.PREDICTING
 
-    snapshot_ts = float(snapshot.get("timestamp") or _payload(snapshot).get("frozen_at") or 0.0)
+    snapshot_ts = _snapshot_time(snapshot)
+    if snapshot_ts is None:
+        return MarketStatus.PREDICTING
     evidence_close = snapshot_ts + float(CONFIG["evidence_window_hours"]) * _SECONDS_PER_HOUR
     if now < evidence_close:
         return MarketStatus.EVIDENCE
@@ -110,29 +117,37 @@ def should_advance_phase(
     - EVIDENCE → RESOLVING: just a status change (no chain event).
     - RESOLVING → FINAL/INVALID: emit ``resolution_finalize``.
     """
-    events = _events_for_market(market_id, chain)
+    events = events_for_market(market_id, chain)
     if not events:
         return None
 
-    create_event = next((e for e in events if e.get("event_type") == "prediction_create"), None)
-    if create_event is None:
+    create_event = first_authoritative_event(events, "prediction_create")
+    if create_event is None or not has_valid_ordering(create_event):
         return None
-    finalize = next((e for e in events if e.get("event_type") == "resolution_finalize"), None)
-    if finalize is not None:
-        return None  # already terminal
 
-    create_payload = _payload(create_event)
-    trigger_date = float(create_payload.get("trigger_date") or 0.0)
-    snapshot = next((e for e in events if e.get("event_type") == "market_snapshot"), None)
+    finalize = first_authoritative_event(events, "resolution_finalize")
+    if finalize is not None:
+        # Whether valid (terminal) or malformed (fail closed), do not let a
+        # later finalize replace the first authoritative chain event.
+        return None
+
+    create_payload = payload(create_event)
+    trigger_date = finite_float(create_payload.get("trigger_date"))
+    snapshot = first_authoritative_event(events, "market_snapshot")
 
     if snapshot is None:
         # PREDICTING — advance to EVIDENCE iff trigger_date has passed in
-        # majority chain time.
-        if now >= trigger_date:
+        # majority chain time. Invalid trigger metadata cannot authorize
+        # a phase transition.
+        if trigger_date is not None and now >= trigger_date:
             return (MarketStatus.PREDICTING, MarketStatus.EVIDENCE)
         return None
 
-    snapshot_ts = float(snapshot.get("timestamp") or _payload(snapshot).get("frozen_at") or 0.0)
+    snapshot_ts = _snapshot_time(snapshot)
+    if snapshot_ts is None:
+        # An invalid first snapshot cannot authorize a transition, and a
+        # later snapshot must not replace the commitment boundary.
+        return None
     evidence_close = snapshot_ts + float(CONFIG["evidence_window_hours"]) * _SECONDS_PER_HOUR
     resolution_close = evidence_close + float(CONFIG["resolution_window_hours"]) * _SECONDS_PER_HOUR
 

backend/services/infonet/markets/snapshot.py

@@ -15,41 +15,34 @@ evaluation. Once frozen:
   can't pre-mine before the boundary.
 
 The snapshot itself is **immutable** by spec — the producer emits it
-once and never updates it. Sprint 4 enforces immutability by ignoring
-any subsequent ``market_snapshot`` events with the same market_id
-(``find_snapshot`` returns the FIRST one). Tests assert this invariant.
+once and never updates it. The first snapshot in canonical hashchain
+append order is authoritative; a malformed first snapshot fails closed
+and cannot be replaced by a later one.
 """
 
 from __future__ import annotations
 
 import hashlib
 import json
-import math
 from typing import Any, Iterable
 
-
-def _payload(event: dict[str, Any]) -> dict[str, Any]:
-    p = event.get("payload")
-    return p if isinstance(p, dict) else {}
+from services.infonet.markets.event_selection import (
+    events_for_market,
+    finite_float,
+    first_authoritative_event,
+    has_valid_ordering,
+    payload,
+)
 
 
-def _finite_float(value: Any) -> float | None:
-    try:
-        parsed = float(value)
-    except (TypeError, ValueError, OverflowError):
-        return None
-    return parsed if math.isfinite(parsed) else None
+class InvalidAuthoritativeSnapshot(ValueError):
+    """Raised when the first on-chain snapshot exists but is malformed.
 
-
-def _events_for_market(market_id: str, chain: Iterable[dict[str, Any]]) -> list[dict[str, Any]]:
-    out: list[dict[str, Any]] = []
-    for ev in chain:
-        if not isinstance(ev, dict):
-            continue
-        if _payload(ev).get("market_id") == market_id:
-            out.append(ev)
-    out.sort(key=lambda e: (float(e.get("timestamp") or 0.0), int(e.get("sequence") or 0)))
-    return out
+    ``None`` from :func:`find_snapshot` is reserved for the distinct state in
+    which no snapshot exists. Raising here prevents downstream resolution and
+    eligibility code from interpreting a corrupted commitment as an absent
+    snapshot with relaxed restrictions.
+    """
 
 
 def build_snapshot(
@@ -60,21 +53,21 @@ def build_snapshot(
 ) -> dict[str, Any]:
     """Compute the snapshot payload deterministically from chain history.
 
-    Walks ``prediction_place`` events for ``market_id``, in chain order,
-    and produces the frozen counts / stake totals / predictor list /
-    yes-no probability state. The resulting dict is ready to be written
-    as the payload of a ``market_snapshot`` event.
+    Walks ``prediction_place`` events for ``market_id`` in canonical hashchain
+    append order and produces the frozen counts / stake totals / predictor list /
+    yes-no probability state. The resulting dict is ready to be written as the
+    payload of a ``market_snapshot`` event.
 
     ``frozen_at`` is the canonical commitment timestamp — typically
     ``chain_majority_time(chain)`` at the moment the producer decides
     to advance to EVIDENCE. Pass it explicitly so the function stays
     pure and deterministic.
     """
-    frozen_at_value = _finite_float(frozen_at)
+    frozen_at_value = finite_float(frozen_at)
     if frozen_at_value is None:
         raise ValueError("frozen_at must be finite")
 
-    events = _events_for_market(market_id, chain)
+    events = events_for_market(market_id, chain)
 
     predictor_ids: list[str] = []
     seen_predictors: set[str] = set()
@@ -88,7 +81,7 @@ def build_snapshot(
         node = ev.get("node_id")
         if not isinstance(node, str) or not node:
             continue
-        p = _payload(ev)
+        p = payload(ev)
         side = p.get("side")
         if side not in ("yes", "no"):
             continue
@@ -98,7 +91,7 @@ def build_snapshot(
             weight = 1.0  # Free pick = 1.0 virtual stake (RULES §5.2).
             staked_amount = 0.0
         else:
-            parsed_stake = _finite_float(stake)
+            parsed_stake = finite_float(stake)
             if parsed_stake is None or parsed_stake <= 0:
                 # Invalid paid predictions must not inflate participant
                 # counts or poison the frozen probability state.
@@ -165,21 +158,27 @@ def find_snapshot(
     market_id: str,
     chain: Iterable[dict[str, Any]],
 ) -> dict[str, Any] | None:
-    """Return the FIRST ``market_snapshot`` payload for ``market_id``.
+    """Return the first authoritative ``market_snapshot`` payload.
 
-    Subsequent ``market_snapshot`` events with the same market_id are
-    ignored — snapshots are immutable per RULES §2.2. This is a
-    structural enforcement, not just a convention; an attacker who
-    forges a second snapshot cannot influence resolution.
+    Selection follows canonical hashchain append order. ``None`` means no
+    snapshot exists. If the first snapshot exists but has malformed ordering
+    metadata, :class:`InvalidAuthoritativeSnapshot` is raised instead of
+    treating the corruption as absence or allowing a later snapshot to replace
+    the commitment.
     """
-    events = _events_for_market(market_id, chain)
-    for ev in events:
-        if ev.get("event_type") == "market_snapshot":
-            return _payload(ev)
-    return None
+    events = events_for_market(market_id, chain)
+    snapshot = first_authoritative_event(events, "market_snapshot")
+    if snapshot is None:
+        return None
+    if not has_valid_ordering(snapshot):
+        raise InvalidAuthoritativeSnapshot(
+            f"authoritative snapshot for market '{market_id}' has invalid ordering metadata"
+        )
+    return payload(snapshot)
 
 
 __all__ = [
+    "InvalidAuthoritativeSnapshot",
     "build_snapshot",
     "compute_snapshot_event_hash",
     "find_snapshot",

backend/services/infonet/tests/test_lifecycle_invalid_ordering.py

@@ -0,0 +1,154 @@
+"""Regression coverage for malformed market lifecycle ordering metadata."""
+
+from typing import Any
+
+import pytest
+
+from services.infonet.config import CONFIG
+from services.infonet.markets import MarketStatus, compute_market_status, should_advance_phase
+from services.infonet.markets.event_selection import events_for_market
+from services.infonet.markets.resolution import resolve_market
+from services.infonet.markets.snapshot import InvalidAuthoritativeSnapshot, find_snapshot
+
+
+def _create(trigger_date: Any = 200.0) -> dict[str, Any]:
+    return {
+        "event_type": "prediction_create",
+        "node_id": "creator",
+        "timestamp": 100.0,
+        "sequence": 1,
+        "payload": {"market_id": "m1", "trigger_date": trigger_date},
+    }
+
+
+def _snapshot(
+    timestamp: Any = 200.0, frozen_at: Any = 200.0, sequence: Any = 2
+) -> dict[str, Any]:
+    return {
+        "event_type": "market_snapshot",
+        "node_id": "creator",
+        "timestamp": timestamp,
+        "sequence": sequence,
+        "payload": {
+            "market_id": "m1",
+            "frozen_at": frozen_at,
+            "frozen_predictor_ids": ["predictor"],
+        },
+    }
+
+
+def _finalize(timestamp: Any = 300.0, sequence: Any = 3) -> dict[str, Any]:
+    return {
+        "event_type": "resolution_finalize",
+        "node_id": "creator",
+        "timestamp": timestamp,
+        "sequence": sequence,
+        "payload": {"market_id": "m1", "outcome": "yes"},
+    }
+
+
+def test_malformed_non_authoritative_metadata_does_not_break_projection() -> None:
+    chain = [
+        _create(),
+        _snapshot(),
+        {
+            "event_type": "prediction_place",
+            "node_id": "peer",
+            "timestamp": "not-a-timestamp",
+            "sequence": "not-a-sequence",
+            "payload": {"market_id": "m1"},
+        },
+    ]
+
+    assert compute_market_status("m1", chain, now=201.0) == MarketStatus.EVIDENCE
+
+
+def test_market_events_preserve_canonical_hashchain_order() -> None:
+    first = _snapshot(timestamp=300.0, sequence=5)
+    second = _snapshot(timestamp=200.0, sequence=2)
+
+    events = events_for_market("m1", [_create(), first, second])
+
+    assert events[1:] == [first, second]
+
+
+def test_lifecycle_and_find_snapshot_share_authoritative_append_order() -> None:
+    evidence_window = float(CONFIG["evidence_window_hours"]) * 3600.0
+    first_timestamp = 10.0 * evidence_window
+    first = _snapshot(timestamp=first_timestamp, frozen_at=first_timestamp, sequence=5)
+    second = _snapshot(timestamp=1.0, frozen_at=1.0, sequence=2)
+    chain = [_create(), first, second]
+
+    # Lifecycle must use the first appended snapshot. If snapshot.py re-sorted
+    # by timestamp/sequence, the much older second event would be selected and
+    # this same `now` would project RESOLVING instead of EVIDENCE.
+    now = first_timestamp + evidence_window / 2.0
+    assert compute_market_status("m1", chain, now=now) == MarketStatus.EVIDENCE
+    assert find_snapshot("m1", chain) == first["payload"]
+
+
+def test_invalid_first_snapshot_fails_closed_without_replacement() -> None:
+    chain = [
+        _create(),
+        _snapshot(timestamp="bad", frozen_at=200.0, sequence="bad"),
+        _snapshot(timestamp=200.0, frozen_at=200.0, sequence=3),
+    ]
+
+    assert compute_market_status("m1", chain, now=10_000.0) == MarketStatus.PREDICTING
+    assert should_advance_phase("m1", chain, now=10_000.0) is None
+    with pytest.raises(InvalidAuthoritativeSnapshot):
+        find_snapshot("m1", chain)
+
+
+def test_invalid_first_snapshot_halts_direct_resolution() -> None:
+    chain = [
+        _create(),
+        _snapshot(timestamp="bad", frozen_at=200.0, sequence="bad"),
+        _snapshot(timestamp=200.0, frozen_at=200.0, sequence=3),
+        {
+            "event_type": "resolution_stake",
+            "node_id": "resolver",
+            "timestamp": 400.0,
+            "sequence": 4,
+            "payload": {
+                "market_id": "m1",
+                "side": "yes",
+                "amount": 10.0,
+                "rep_type": "oracle",
+            },
+        },
+    ]
+
+    # A malformed authoritative commitment is distinct from no snapshot.
+    # Resolution must halt instead of treating it as an empty exclusion set.
+    with pytest.raises(InvalidAuthoritativeSnapshot):
+        resolve_market("m1", chain)
+
+
+def test_nonfinite_snapshot_timestamp_does_not_fall_back_to_frozen_at() -> None:
+    chain = [_create(), _snapshot(timestamp=float("nan"), frozen_at=200.0)]
+
+    assert compute_market_status("m1", chain, now=201.0) == MarketStatus.PREDICTING
+    assert should_advance_phase("m1", chain, now=10_000.0) is None
+    with pytest.raises(InvalidAuthoritativeSnapshot):
+        find_snapshot("m1", chain)
+
+
+def test_invalid_first_finalize_cannot_be_replaced_by_later_finalize() -> None:
+    chain = [
+        _create(),
+        _snapshot(),
+        _finalize(timestamp="bad", sequence="bad"),
+        _finalize(timestamp=300.0, sequence=4),
+    ]
+
+    # The malformed finalize has no terminal authority, so projection stays
+    # in the prior snapshot-backed phase. The later finalize cannot replace it.
+    assert compute_market_status("m1", chain, now=201.0) == MarketStatus.EVIDENCE
+    assert should_advance_phase("m1", chain, now=10_000.0) is None
+
+
+def test_invalid_trigger_date_cannot_authorize_phase_advance() -> None:
+    chain = [_create(trigger_date="not-a-date")]
+
+    assert should_advance_phase("m1", chain, now=10_000.0) is None