Fence unmanaged lifecycle by snapshot time

Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>

tests/adapters/portfolio/test_unmanaged_position_store.py

@@ -161,13 +161,83 @@ class TestCRUDRoundTrip:
         audit = store.list_audit(instance_key="inst-A", symbol="008110")
         assert [row["event_type"] for row in audit] == ["detected"]
 
-    def test_concurrent_observe_and_resolve_leave_consistent_state_and_audit(self, pg_clean):
+    def test_older_absent_snapshot_applied_after_newer_observation_is_ignored(self, pg_unmanaged_store):
+        t1 = datetime(2026, 9, 7, 0, 0, tzinfo=UTC)
+        t2 = t1 + timedelta(minutes=1)
+        t3 = t2 + timedelta(minutes=1)
+
+        pg_unmanaged_store.observe(
+            instance_key="inst-A",
+            symbol="008110",
+            quantity=3,
+            observed_at=t1,
+        )
+        observed = pg_unmanaged_store.observe(
+            instance_key="inst-A",
+            symbol="008110",
+            quantity=5,
+            observed_at=t3,
+        )
+        resolved = pg_unmanaged_store.resolve_absent(
+            instance_key="inst-A",
+            observed_symbols=set(),
+            observed_at=t2,
+        )
+
+        assert observed["transition"] == "quantity_changed"
+        assert resolved == []
+        state = pg_unmanaged_store.get_state(instance_key="inst-A", symbol="008110")
+        assert state is not None
+        assert state["lifecycle_state"] == "active"
+        assert state["quantity"] == 5
+        assert [row["event_type"] for row in pg_unmanaged_store.list_audit(
+            instance_key="inst-A",
+            symbol="008110",
+        )] == ["detected", "quantity_changed"]
+
+    def test_older_observation_applied_after_newer_resolution_is_ignored(self, pg_unmanaged_store):
+        t1 = datetime(2026, 9, 7, 0, 0, tzinfo=UTC)
+        t2 = t1 + timedelta(minutes=1)
+        t3 = t2 + timedelta(minutes=1)
+
+        pg_unmanaged_store.observe(
+            instance_key="inst-A",
+            symbol="008110",
+            quantity=3,
+            observed_at=t1,
+        )
+        pg_unmanaged_store.resolve_absent(
+            instance_key="inst-A",
+            observed_symbols=set(),
+            observed_at=t3,
+        )
+        stale = pg_unmanaged_store.observe(
+            instance_key="inst-A",
+            symbol="008110",
+            quantity=5,
+            observed_at=t2,
+        )
+
+        assert stale["transition"] == "stale"
+        state = pg_unmanaged_store.get_state(instance_key="inst-A", symbol="008110")
+        assert state is not None
+        assert state["lifecycle_state"] == "resolved"
+        assert state["quantity"] == 3
+        assert [row["event_type"] for row in pg_unmanaged_store.list_audit(
+            instance_key="inst-A",
+            symbol="008110",
+        )] == ["detected", "resolved"]
+
+    def test_concurrent_observe_and_resolve_respects_newest_snapshot_timestamp(self, pg_clean):
         from trex_core.adapters.portfolio.unmanaged_position_store import (
             PostgreSQLUnmanagedPositionStore,
         )
 
         store = PostgreSQLUnmanagedPositionStore(pg_clean)
-        store.observe(instance_key="inst-A", symbol="008110", quantity=3)
+        t1 = datetime(2026, 9, 7, 0, 0, tzinfo=UTC)
+        t2 = t1 + timedelta(minutes=1)
+        t3 = t2 + timedelta(minutes=1)
+        store.observe(instance_key="inst-A", symbol="008110", quantity=3, observed_at=t1)
 
         with ThreadPoolExecutor(max_workers=2) as pool:
             observe_future = pool.submit(
@@ -175,32 +245,24 @@ class TestCRUDRoundTrip:
                 instance_key="inst-A",
                 symbol="008110",
                 quantity=5,
+                observed_at=t3,
             )
             resolve_future = pool.submit(
                 PostgreSQLUnmanagedPositionStore(pg_clean).resolve_absent,
                 instance_key="inst-A",
                 observed_symbols=set(),
+                observed_at=t2,
             )
             observe_result = observe_future.result()
             resolve_result = resolve_future.result()
 
         state = store.get_state(instance_key="inst-A", symbol="008110")
         assert state is not None
-        audit = store.list_audit(instance_key="inst-A", symbol="008110")
+        assert observe_result["transition"] == "quantity_changed"
-        event_types = [row["event_type"] for row in audit]
+        assert resolve_result == []
-        assert event_types[0] == "detected"
+        assert state["lifecycle_state"] == "active"
-        assert event_types in (
+        assert state["quantity"] == 5
-            ["detected", "quantity_changed", "resolved"],
+        assert [row["event_type"] for row in store.list_audit(
-            ["detected", "resolved", "detected"],
+            instance_key="inst-A",
-        )
+            symbol="008110",
-        if state["lifecycle_state"] == "resolved":
+        )] == ["detected", "quantity_changed"]
-            assert observe_result["transition"] == "quantity_changed"
-            assert len(resolve_result) == 1
-            assert state["quantity"] == 5
-            assert event_types[-1] == "resolved"
-        else:
-            assert state["lifecycle_state"] == "active"
-            assert observe_result["transition"] == "detected"
-            assert len(resolve_result) == 1
-            assert state["quantity"] == 5
-            assert event_types[-1] == "detected"

tests/application/test_unmanaged_position_lifecycle.py

@@ -140,6 +140,49 @@ def test_in_memory_store_preserves_first_seen_when_resolved_symbol_reappears():
     assert reopened["state"]["evidence_ref"] is None
 
 
+def test_in_memory_store_ignores_stale_resolve_after_newer_observation():
+    store = InMemoryUnmanagedPositionStore()
+    t1 = datetime(2026, 9, 7, 0, 0, tzinfo=UTC)
+    t2 = t1 + timedelta(minutes=1)
+    t3 = t2 + timedelta(minutes=1)
+
+    store.observe(instance_key=INSTANCE, symbol="008110", quantity=3, observed_at=t1)
+    store.observe(instance_key=INSTANCE, symbol="008110", quantity=5, observed_at=t3)
+    resolved = store.resolve_absent(
+        instance_key=INSTANCE,
+        observed_symbols=set(),
+        observed_at=t2,
+    )
+
+    assert resolved == []
+    state = store.get_state(instance_key=INSTANCE, symbol="008110")
+    assert state is not None
+    assert state["lifecycle_state"] == "active"
+    assert state["quantity"] == 5
+    assert store.list_audit(instance_key=INSTANCE, symbol="008110")[-1]["event_type"] == "quantity_changed"
+
+
+def test_in_memory_store_ignores_stale_observation_after_newer_resolution():
+    store = InMemoryUnmanagedPositionStore()
+    t1 = datetime(2026, 9, 7, 0, 0, tzinfo=UTC)
+    t2 = t1 + timedelta(minutes=1)
+    t3 = t2 + timedelta(minutes=1)
+
+    store.observe(instance_key=INSTANCE, symbol="008110", quantity=3, observed_at=t1)
+    store.resolve_absent(instance_key=INSTANCE, observed_symbols=set(), observed_at=t3)
+    stale = store.observe(instance_key=INSTANCE, symbol="008110", quantity=5, observed_at=t2)
+
+    assert stale["transition"] == "stale"
+    state = store.get_state(instance_key=INSTANCE, symbol="008110")
+    assert state is not None
+    assert state["lifecycle_state"] == "resolved"
+    assert state["quantity"] == 3
+    assert [row["event_type"] for row in store.list_audit(instance_key=INSTANCE, symbol="008110")] == [
+        "detected",
+        "resolved",
+    ]
+
+
 def test_reconcile_core_holdings_suppresses_repeated_unmanaged_alarm_spam():
     repo, svc = _wire()
     emitter, transport = _make_emitter()

trex_core/adapters/portfolio/unmanaged_position_store.py

@@ -21,6 +21,18 @@ class PostgreSQLUnmanagedPositionStore:
         self._psycopg = _require_psycopg(psycopg_module)
         self._dsn = dsn
 
+    @staticmethod
+    def _latest_snapshot_at(state: dict) -> datetime:
+        resolved_at = state.get("resolved_at")
+        last_seen_at = state.get("last_seen_at")
+        if isinstance(resolved_at, datetime) and (
+            not isinstance(last_seen_at, datetime) or resolved_at >= last_seen_at
+        ):
+            return resolved_at
+        if isinstance(last_seen_at, datetime):
+            return last_seen_at
+        return state.get("updated_at") or state.get("created_at") or datetime.min.replace(tzinfo=UTC)
+
     def observe(
         self,
         *,
@@ -43,6 +55,13 @@ class PostgreSQLUnmanagedPositionStore:
                     row = cur.fetchone()
                     previous = self._row_to_state(row) if row is not None else None
                     previous_quantity = previous.get("quantity") if previous is not None else None
+                    if previous is not None and now < self._latest_snapshot_at(previous):
+                        conn.rollback()
+                        return {
+                            "transition": "stale",
+                            "previous_quantity": previous_quantity,
+                            "state": previous,
+                        }
                     if previous is None:
                         transition = "detected"
                         cur.execute(
@@ -54,6 +73,13 @@ class PostgreSQLUnmanagedPositionStore:
                             (instance_key, symbol, quantity, now, now, now, now, now),
                         )
                     elif previous["lifecycle_state"] == "resolved":
+                        if now <= self._latest_snapshot_at(previous):
+                            conn.rollback()
+                            return {
+                                "transition": "stale",
+                                "previous_quantity": previous_quantity,
+                                "state": previous,
+                            }
                         transition = "detected"
                         cur.execute(
                             "UPDATE unmanaged_position_state "
@@ -63,29 +89,58 @@ class PostgreSQLUnmanagedPositionStore:
                             "    operator_actor = NULL, operator_updated_at = NULL, "
                             "    updated_at = %s "
                             "WHERE instance_key = %s AND symbol = %s "
+                            "  AND lifecycle_state = 'resolved' AND resolved_at < %s "
                             f"RETURNING {_STATE_COLUMNS}",
-                            (quantity, now, now, now, instance_key, symbol),
+                            (quantity, now, now, now, instance_key, symbol, now),
                         )
                     elif int(previous["quantity"]) != int(quantity):
+                        if now <= previous["last_seen_at"]:
+                            conn.rollback()
+                            return {
+                                "transition": "stale",
+                                "previous_quantity": previous_quantity,
+                                "state": previous,
+                            }
                         transition = "quantity_changed"
                         cur.execute(
                             "UPDATE unmanaged_position_state "
                             "SET quantity = %s, last_seen_at = %s, quantity_changed_at = %s, "
                             "    resolved_at = NULL, lifecycle_state = 'active', updated_at = %s "
                             "WHERE instance_key = %s AND symbol = %s "
+                            "  AND lifecycle_state = 'active' AND last_seen_at < %s "
                             f"RETURNING {_STATE_COLUMNS}",
-                            (quantity, now, now, now, instance_key, symbol),
+                            (quantity, now, now, now, instance_key, symbol, now),
                         )
                     else:
                         transition = "unchanged"
-                        cur.execute(
+                        if now > previous["last_seen_at"]:
-                            "UPDATE unmanaged_position_state "
+                            cur.execute(
-                            "SET last_seen_at = %s, updated_at = %s "
+                                "UPDATE unmanaged_position_state "
-                            "WHERE instance_key = %s AND symbol = %s "
+                                "SET last_seen_at = %s, updated_at = %s "
-                            f"RETURNING {_STATE_COLUMNS}",
+                                "WHERE instance_key = %s AND symbol = %s "
-                            (now, now, instance_key, symbol),
+                                "  AND lifecycle_state = 'active' AND quantity = %s AND last_seen_at < %s "
-                        )
+                                f"RETURNING {_STATE_COLUMNS}",
-                    state = self._row_to_state(cur.fetchone())
+                                (now, now, instance_key, symbol, quantity, now),
+                            )
+                            fetched = cur.fetchone()
+                            state = previous if fetched is None else self._row_to_state(fetched)
+                        else:
+                            conn.rollback()
+                            return {
+                                "transition": "unchanged",
+                                "previous_quantity": previous_quantity,
+                                "state": previous,
+                            }
+                    if transition != "unchanged":
+                        fetched = cur.fetchone()
+                        if fetched is None:
+                            conn.rollback()
+                            return {
+                                "transition": "stale",
+                                "previous_quantity": previous_quantity,
+                                "state": previous,
+                            }
+                        state = self._row_to_state(fetched)
                     if transition != "unchanged":
                         note = (
                             f"previous_quantity={previous_quantity}"
@@ -139,14 +194,20 @@ class PostgreSQLUnmanagedPositionStore:
             for state in active:
                 if state["symbol"] in observed_symbols:
                     continue
+                if now <= state["last_seen_at"]:
+                    continue
                 cur.execute(
                     "UPDATE unmanaged_position_state "
                     "SET lifecycle_state = 'resolved', resolved_at = %s, updated_at = %s "
                     "WHERE instance_key = %s AND symbol = %s "
+                    "  AND lifecycle_state = 'active' AND last_seen_at < %s "
                     f"RETURNING {_STATE_COLUMNS}",
-                    (now, now, instance_key, state["symbol"]),
+                    (now, now, instance_key, state["symbol"], now),
                 )
-                resolved_state = self._row_to_state(cur.fetchone())
+                row = cur.fetchone()
+                if row is None:
+                    continue
+                resolved_state = self._row_to_state(row)
                 resolved.append(resolved_state)
                 self._insert_audit(
                     cur,

trex_core/application/portfolio.py

@@ -625,6 +625,18 @@ class InMemoryUnmanagedPositionStore:
         self._states: dict[tuple[str, str], dict] = {}
         self._audit: list[dict] = []
 
+    @staticmethod
+    def _latest_snapshot_at(state: dict) -> datetime:
+        resolved_at = state.get("resolved_at")
+        last_seen_at = state.get("last_seen_at")
+        if isinstance(resolved_at, datetime) and (
+            not isinstance(last_seen_at, datetime) or resolved_at >= last_seen_at
+        ):
+            return resolved_at
+        if isinstance(last_seen_at, datetime):
+            return last_seen_at
+        return state.get("updated_at") or state.get("created_at") or datetime.min.replace(tzinfo=UTC)
+
     def observe(
         self,
         *,
@@ -642,7 +654,21 @@ class InMemoryUnmanagedPositionStore:
             preserved = {} if existing is None else {
                 "created_at": existing.get("created_at", now),
             }
+            if existing is not None:
+                latest_snapshot_at = self._latest_snapshot_at(existing)
+                if now < latest_snapshot_at:
+                    return {
+                        "transition": "stale",
+                        "previous_quantity": previous_quantity,
+                        "state": dict(existing),
+                    }
             if existing is None or existing.get("lifecycle_state") == "resolved":
+                if existing is not None and now == self._latest_snapshot_at(existing):
+                    return {
+                        "transition": "stale",
+                        "previous_quantity": previous_quantity,
+                        "state": dict(existing),
+                    }
                 transition = "detected"
                 state = {
                     "instance_key": instance_key,
@@ -666,6 +692,12 @@ class InMemoryUnmanagedPositionStore:
                 }
                 self._states[key] = state
             elif int(existing["quantity"]) != int(quantity):
+                if now <= existing["last_seen_at"]:
+                    return {
+                        "transition": "stale",
+                        "previous_quantity": previous_quantity,
+                        "state": dict(existing),
+                    }
                 transition = "quantity_changed"
                 existing["quantity"] = quantity
                 existing["last_seen_at"] = now
@@ -676,8 +708,9 @@ class InMemoryUnmanagedPositionStore:
                 state = existing
             else:
                 transition = "unchanged"
-                existing["last_seen_at"] = now
+                if now > existing["last_seen_at"]:
-                existing["updated_at"] = now
+                    existing["last_seen_at"] = now
+                    existing["updated_at"] = now
                 state = existing
             if transition != "unchanged":
                 self._audit.append(
@@ -722,6 +755,8 @@ class InMemoryUnmanagedPositionStore:
                     continue
                 if state["symbol"] in observed_symbols:
                     continue
+                if now <= state["last_seen_at"]:
+                    continue
                 state["lifecycle_state"] = "resolved"
                 state["resolved_at"] = now
                 state["updated_at"] = now

trex_core/application/reconciliation.py

@@ -29,6 +29,7 @@ import hashlib
 import logging
 import math
 import re
+from datetime import UTC, datetime
 from typing import cast
 
 from trex_core.application.alarm_emitter import AlarmEmitter
@@ -407,6 +408,7 @@ def reconcile_core_holdings(
     Returns {key: 'error'|'qty_warning'|'ok'|'unmanaged_position'|
     'managed_non_holding_divergence'} for telemetry.
     """
+    snapshot_observed_at = datetime.now(tz=UTC)
     broker_qty: dict[str, int] = {
         item.symbol: int(item.hldg_qty) for item in broker.get_balance()
     }
@@ -507,6 +509,7 @@ def reconcile_core_holdings(
                             instance_key=instance_key,
                             symbol=sym,
                             quantity=bq,
+                            observed_at=snapshot_observed_at,
                         )
                     except Exception:  # noqa: BLE001
                         _log.exception(
@@ -555,6 +558,7 @@ def reconcile_core_holdings(
             for state in unmanaged_position_store.resolve_absent(
                 instance_key=instance_key,
                 observed_symbols=active_unmanaged_symbols,
+                observed_at=snapshot_observed_at,
             ):
                 _log.info(
                     "unmanaged position resolved: symbol=%s qty=%d instance_key=%s",