From be2e73a2d660dd3df48233033959d1718349267e Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 12 Sep 2026 13:31:32 +0000 Subject: [PATCH] fix: harden unmanaged position lifecycle writes Co-authored-by: genichin <4282227+genichin@users.noreply.github.com> --- tests/runtime/test_cli_operator.py | 47 ++++++ .../portfolio/unmanaged_position_store.py | 159 +++++++++--------- 2 files changed, 130 insertions(+), 76 deletions(-) diff --git a/tests/runtime/test_cli_operator.py b/tests/runtime/test_cli_operator.py index 51c3610..91d3775 100644 --- a/tests/runtime/test_cli_operator.py +++ b/tests/runtime/test_cli_operator.py @@ -741,6 +741,53 @@ def test_unmanaged_record_text_output(monkeypatch, capsys): assert "evidence_ref=kis-balance:2026-09-11" in out +def test_unmanaged_record_main_dispatch(monkeypatch, capsys): + class _Store: + def __init__(self, *_args, **_kwargs): + pass + + def record_operator_action(self, **kwargs): + return { + "instance_key": "default", + "symbol": kwargs["symbol"], + "quantity": 3, + "lifecycle_state": "active", + "first_seen_at": "2026-09-07T00:00:00+00:00", + "last_seen_at": "2026-09-11T00:00:00+00:00", + "quantity_changed_at": "2026-09-07T00:00:00+00:00", + "resolved_at": None, + "operator_action": kwargs["action"], + "evidence_ref": kwargs["evidence_ref"], + "operator_note": kwargs["note"], + "operator_actor": kwargs["actor"], + "operator_updated_at": "2026-09-11T00:00:00+00:00", + "created_at": "2026-09-07T00:00:00+00:00", + "updated_at": "2026-09-11T00:00:00+00:00", + } + + monkeypatch.setattr( + "trex_core.adapters.portfolio.unmanaged_position_store.PostgreSQLUnmanagedPositionStore", + _Store, + ) + + rc = cli.main( + [ + "unmanaged", + "record", + "008110", + "investigating", + "--evidence-ref", + "kis-balance:2026-09-11", + "--json", + ] + ) + + assert rc == 0 + out = json.loads(capsys.readouterr().out) + assert out["result"] == "recorded" + assert out["state"]["symbol"] == "008110" + + # --------------------------------------------------------------------------- # REL-010 regression: _cmd_verify broker import must not raise ImportError # --------------------------------------------------------------------------- diff --git a/trex_core/adapters/portfolio/unmanaged_position_store.py b/trex_core/adapters/portfolio/unmanaged_position_store.py index 694ab4b..6e75469 100644 --- a/trex_core/adapters/portfolio/unmanaged_position_store.py +++ b/trex_core/adapters/portfolio/unmanaged_position_store.py @@ -31,82 +31,89 @@ class PostgreSQLUnmanagedPositionStore: actor: str = "reconcile", ) -> dict: now = observed_at or datetime.now(tz=UTC) - with self._psycopg.connect(self._dsn, autocommit=False) as conn, conn.cursor() as cur: - cur.execute( - f"SELECT {_STATE_COLUMNS} FROM unmanaged_position_state " - "WHERE instance_key = %s AND symbol = %s " - "FOR UPDATE", - (instance_key, symbol), - ) - 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 None: - transition = "detected" - cur.execute( - "INSERT INTO unmanaged_position_state " - f"({_STATE_COLUMNS}) " - "VALUES (%s, %s, %s, 'active', %s, %s, %s, NULL, " - " NULL, NULL, '', NULL, NULL, %s, %s) " - f"RETURNING {_STATE_COLUMNS}", - (instance_key, symbol, quantity, now, now, now, now, now), - ) - elif previous["lifecycle_state"] == "resolved": - transition = "detected" - cur.execute( - "UPDATE unmanaged_position_state " - "SET quantity = %s, lifecycle_state = 'active', " - " first_seen_at = %s, last_seen_at = %s, quantity_changed_at = %s, " - " resolved_at = NULL, updated_at = %s " - "WHERE instance_key = %s AND symbol = %s " - f"RETURNING {_STATE_COLUMNS}", - (quantity, now, now, now, now, instance_key, symbol), - ) - elif int(previous["quantity"]) != int(quantity): - 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 " - f"RETURNING {_STATE_COLUMNS}", - (quantity, now, now, now, instance_key, symbol), - ) - else: - transition = "unchanged" - cur.execute( - "UPDATE unmanaged_position_state " - "SET last_seen_at = %s, updated_at = %s " - "WHERE instance_key = %s AND symbol = %s " - f"RETURNING {_STATE_COLUMNS}", - (now, now, instance_key, symbol), - ) - state = self._row_to_state(cur.fetchone()) - if transition != "unchanged": - note = ( - f"previous_quantity={previous_quantity}" - if transition == "quantity_changed" and previous_quantity is not None - else "" - ) - self._insert_audit( - cur, - instance_key=instance_key, - symbol=symbol, - event_type=transition, - action=None, - quantity=quantity, - lifecycle_state=state["lifecycle_state"], - actor=actor, - evidence_ref=state.get("evidence_ref"), - note=note, - created_at=now, - ) - conn.commit() - return { - "transition": transition, - "previous_quantity": previous_quantity, - "state": state, - } + for _attempt in range(2): + try: + with self._psycopg.connect(self._dsn, autocommit=False) as conn, conn.cursor() as cur: + cur.execute( + f"SELECT {_STATE_COLUMNS} FROM unmanaged_position_state " + "WHERE instance_key = %s AND symbol = %s " + "FOR UPDATE", + (instance_key, symbol), + ) + 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 None: + transition = "detected" + cur.execute( + "INSERT INTO unmanaged_position_state " + f"({_STATE_COLUMNS}) " + "VALUES (%s, %s, %s, 'active', %s, %s, %s, NULL, " + " NULL, NULL, '', NULL, NULL, %s, %s) " + f"RETURNING {_STATE_COLUMNS}", + (instance_key, symbol, quantity, now, now, now, now, now), + ) + elif previous["lifecycle_state"] == "resolved": + transition = "detected" + cur.execute( + "UPDATE unmanaged_position_state " + "SET quantity = %s, lifecycle_state = 'active', " + " first_seen_at = %s, last_seen_at = %s, quantity_changed_at = %s, " + " resolved_at = NULL, updated_at = %s " + "WHERE instance_key = %s AND symbol = %s " + f"RETURNING {_STATE_COLUMNS}", + (quantity, now, now, now, now, instance_key, symbol), + ) + elif int(previous["quantity"]) != int(quantity): + 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 " + f"RETURNING {_STATE_COLUMNS}", + (quantity, now, now, now, instance_key, symbol), + ) + else: + transition = "unchanged" + cur.execute( + "UPDATE unmanaged_position_state " + "SET last_seen_at = %s, updated_at = %s " + "WHERE instance_key = %s AND symbol = %s " + f"RETURNING {_STATE_COLUMNS}", + (now, now, instance_key, symbol), + ) + state = self._row_to_state(cur.fetchone()) + if transition != "unchanged": + note = ( + f"previous_quantity={previous_quantity}" + if transition == "quantity_changed" and previous_quantity is not None + else "" + ) + self._insert_audit( + cur, + instance_key=instance_key, + symbol=symbol, + event_type=transition, + action=None, + quantity=quantity, + lifecycle_state=state["lifecycle_state"], + actor=actor, + evidence_ref=state.get("evidence_ref"), + note=note, + created_at=now, + ) + conn.commit() + return { + "transition": transition, + "previous_quantity": previous_quantity, + "state": state, + } + except self._psycopg.errors.UniqueViolation: + continue + raise RuntimeError( + "failed to persist unmanaged_position_state due to repeated concurrent insert race" + ) def resolve_absent( self,