Resolve main merge and add unmanaged store race tests

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

tests/adapters/portfolio/test_unmanaged_position_store.py

@@ -1,6 +1,7 @@
 """PostgreSQLUnmanagedPositionStore integration tests."""
 from __future__ import annotations
 
+from concurrent.futures import ThreadPoolExecutor
 from datetime import UTC, datetime, timedelta
 
 import pytest
@@ -136,3 +137,70 @@ class TestCRUDRoundTrip:
         assert reopened["state"]["last_seen_at"] == t3
         assert reopened["state"]["operator_action"] is None
         assert reopened["state"]["evidence_ref"] is None
+
+    def test_concurrent_first_observe_keeps_single_detected_audit(self, pg_clean):
+        from trex_core.adapters.portfolio.unmanaged_position_store import (
+            PostgreSQLUnmanagedPositionStore,
+        )
+
+        def _observe():
+            return PostgreSQLUnmanagedPositionStore(pg_clean).observe(
+                instance_key="inst-A",
+                symbol="008110",
+                quantity=3,
+            )
+
+        with ThreadPoolExecutor(max_workers=2) as pool:
+            results = list(pool.map(lambda _n: _observe(), range(2)))
+
+        assert sorted(result["transition"] for result in results) == ["detected", "unchanged"]
+        store = PostgreSQLUnmanagedPositionStore(pg_clean)
+        state = store.get_state(instance_key="inst-A", symbol="008110")
+        assert state is not None
+        assert state["lifecycle_state"] == "active"
+        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):
+        from trex_core.adapters.portfolio.unmanaged_position_store import (
+            PostgreSQLUnmanagedPositionStore,
+        )
+
+        store = PostgreSQLUnmanagedPositionStore(pg_clean)
+        store.observe(instance_key="inst-A", symbol="008110", quantity=3)
+
+        with ThreadPoolExecutor(max_workers=2) as pool:
+            observe_future = pool.submit(
+                PostgreSQLUnmanagedPositionStore(pg_clean).observe,
+                instance_key="inst-A",
+                symbol="008110",
+                quantity=5,
+            )
+            resolve_future = pool.submit(
+                PostgreSQLUnmanagedPositionStore(pg_clean).resolve_absent,
+                instance_key="inst-A",
+                observed_symbols=set(),
+            )
+            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")
+        event_types = [row["event_type"] for row in audit]
+        assert event_types[0] == "detected"
+        assert event_types in (
+            ["detected", "quantity_changed", "resolved"],
+            ["detected", "resolved", "detected"],
+        )
+        if state["lifecycle_state"] == "resolved":
+            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"