From 84d81658a13f46b3c37b9c2efbf386ca9e556e09 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 13 Sep 2026 02:58:18 +0000 Subject: [PATCH] Resolve main merge and add unmanaged store race tests Co-authored-by: genichin <4282227+genichin@users.noreply.github.com> --- .../test_unmanaged_position_store.py | 68 +++++++++++++++++++ .../test_unmanaged_position_lifecycle.py | 1 + .../portfolio/unmanaged_position_store.py | 2 +- trex_core/application/portfolio.py | 8 +-- 4 files changed, 74 insertions(+), 5 deletions(-) diff --git a/tests/adapters/portfolio/test_unmanaged_position_store.py b/tests/adapters/portfolio/test_unmanaged_position_store.py index c03a2a8..91d3e92 100644 --- a/tests/adapters/portfolio/test_unmanaged_position_store.py +++ b/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" diff --git a/tests/application/test_unmanaged_position_lifecycle.py b/tests/application/test_unmanaged_position_lifecycle.py index ed94ed4..a6de1fd 100644 --- a/tests/application/test_unmanaged_position_lifecycle.py +++ b/tests/application/test_unmanaged_position_lifecycle.py @@ -98,6 +98,7 @@ def test_in_memory_store_tracks_lifecycle_and_operator_audit(): "operator_action", "resolved", ] + assert all("recorded_at" in row for row in audit) def test_in_memory_store_isolates_same_symbol_by_instance(): diff --git a/trex_core/adapters/portfolio/unmanaged_position_store.py b/trex_core/adapters/portfolio/unmanaged_position_store.py index 291842b..f2f2f38 100644 --- a/trex_core/adapters/portfolio/unmanaged_position_store.py +++ b/trex_core/adapters/portfolio/unmanaged_position_store.py @@ -258,7 +258,7 @@ class PostgreSQLUnmanagedPositionStore: "actor": row[4], "evidence_ref": row[5], "note": row[6], - "created_at": row[7], + "recorded_at": row[7], } for row in cur.fetchall() ] diff --git a/trex_core/application/portfolio.py b/trex_core/application/portfolio.py index 5be2aa8..ac5c985 100644 --- a/trex_core/application/portfolio.py +++ b/trex_core/application/portfolio.py @@ -660,7 +660,7 @@ class InMemoryUnmanagedPositionStore: "operator_note": "", "operator_actor": None, "operator_updated_at": None, - "created_at": now, + "recorded_at": now, **preserved, "updated_at": now, } @@ -695,7 +695,7 @@ class InMemoryUnmanagedPositionStore: if transition == "quantity_changed" and previous_quantity is not None else "" ), - "created_at": now, + "recorded_at": now, } ) return { @@ -737,7 +737,7 @@ class InMemoryUnmanagedPositionStore: "actor": actor, "evidence_ref": state.get("evidence_ref"), "note": "", - "created_at": now, + "recorded_at": now, } ) return resolved @@ -798,7 +798,7 @@ class InMemoryUnmanagedPositionStore: "actor": actor, "evidence_ref": evidence_ref, "note": note, - "created_at": now, + "recorded_at": now, } ) return dict(state)