fix: guard unmanaged resolution on authoritative balances

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

tests/application/test_unmanaged_position_lifecycle.py

@@ -35,6 +35,11 @@ class RaisingBroker(FakeBroker):
         raise BrokerBalanceError("transport_failed")
 
 
+class FilteredBroker(FakeBroker):
+    def exposes_full_balance_snapshot(self) -> bool:
+        return False
+
+
 def _wire():
     repo = InMemorySlotRepository()
     return repo, SlotService(repo, instance_key=INSTANCE)
@@ -188,3 +193,22 @@ def test_broker_failure_does_not_resolve_unmanaged_state():
     state = store.get_state(instance_key=INSTANCE, symbol="008110")
     assert state is not None
     assert state["lifecycle_state"] == "active"
+
+
+def test_non_authoritative_broker_view_does_not_resolve_unmanaged_state():
+    repo, svc = _wire()
+    store = InMemoryUnmanagedPositionStore()
+    store.observe(instance_key=INSTANCE, symbol="008110", quantity=3)
+
+    decisions = reconcile_core_holdings(
+        slot_repo=repo,
+        slot_service=svc,
+        broker=FilteredBroker({}),
+        instance_key=INSTANCE,
+        unmanaged_position_store=store,
+    )
+
+    assert decisions == {}
+    state = store.get_state(instance_key=INSTANCE, symbol="008110")
+    assert state is not None
+    assert state["lifecycle_state"] == "active"

tests/runtime/test_cli_operator.py

@@ -593,6 +593,7 @@ def test_unmanaged_list_outputs_durable_state(monkeypatch, capsys):
     assert rc == 0
     out = json.loads(capsys.readouterr().out)
     assert out["count"] == 1
+    assert out["instance_key"] == "inst-1"
     assert out["states"][0]["symbol"] == "008110"
     assert out["states"][0]["evidence_ref"] == "kis-balance:2026-09-11"
 
@@ -688,6 +689,58 @@ def test_unmanaged_record_persists_operator_action(monkeypatch, capsys):
     assert out["state"]["evidence_ref"] == "kis-balance:2026-09-11"
 
 
+def test_unmanaged_record_text_output(monkeypatch, capsys):
+    class _Store:
+        def __init__(self, *_args, **_kwargs):
+            pass
+
+        def record_operator_action(self, **kwargs):
+            return {
+                "instance_key": "inst-1",
+                "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,
+    )
+    args = SimpleNamespace(
+        cmd="unmanaged",
+        unmanaged_cmd="record",
+        symbol="008110",
+        action="approved_exception",
+        evidence_ref="kis-balance:2026-09-11",
+        note="confirmed",
+        actor="operator",
+        instance="inst-1",
+        config=None,
+        config_path=None,
+        env_file=None,
+        json=False,
+        timeout=10.0,
+    )
+
+    rc = cli._cmd_unmanaged_record(args, SimpleNamespace(dsn="postgresql://x/y"))
+
+    assert rc == 0
+    out = capsys.readouterr().out
+    assert "recorded unmanaged position instance=inst-1 symbol=008110" in out
+    assert "evidence_ref=kis-balance:2026-09-11" in out
+
+
 # ---------------------------------------------------------------------------
 # REL-010 regression: _cmd_verify broker import must not raise ImportError
 # ---------------------------------------------------------------------------

trex_core/adapters/broker/ownership.py

@@ -13,7 +13,6 @@ import os
 from datetime import datetime
 from pathlib import Path
 from threading import Lock
-from typing import Optional
 
 from trex_core.application.broker_port import (
     BalanceItem,
@@ -134,8 +133,8 @@ class OwnershipAwareBroker:
         *,
         side: str,
         symbol: str,
-        order_date: Optional[str] = None,
-        submitted_at: Optional[datetime] = None,
+        order_date: str | None = None,
+        submitted_at: datetime | None = None,
     ) -> FillStatus:
         result = self._delegate.poll_fill(
             broker_order_id,
@@ -175,6 +174,10 @@ class OwnershipAwareBroker:
     def get_balance(self) -> list[BalanceItem]:
         return self._store.project(self._delegate.get_balance())
 
+    def exposes_full_balance_snapshot(self) -> bool:
+        """False: projected balances omit operator-owned/unmanaged broker positions."""
+        return False
+
     def account_scope_hash(self) -> str:
         return self._delegate.account_scope_hash()
 

trex_core/application/reconciliation.py

@@ -364,6 +364,18 @@ def _notify_unmanaged_position_transition(
     )
 
 
+def _broker_balances_are_authoritative_for_unmanaged_resolution(broker: BrokerPort) -> bool:
+    marker = getattr(broker, "exposes_full_balance_snapshot", None)
+    if callable(marker):
+        try:
+            return bool(marker())
+        except Exception:  # noqa: BLE001 - fail closed on broker self-report failure
+            return False
+    if isinstance(marker, bool):
+        return marker
+    return True
+
+
 def reconcile_core_holdings(
     *,
     slot_repo,
@@ -535,7 +547,10 @@ def reconcile_core_holdings(
                                     symbol=sym,
                                     quantity=bq,
                                 )
-    if unmanaged_position_store is not None:
+    if (
+        unmanaged_position_store is not None
+        and _broker_balances_are_authoritative_for_unmanaged_resolution(broker)
+    ):
         try:
             for state in unmanaged_position_store.resolve_absent(
                 instance_key=instance_key,
@@ -555,6 +570,11 @@ def reconcile_core_holdings(
                     )
         except Exception:  # noqa: BLE001
             _log.exception("reconcile_core_holdings: unmanaged_position resolve failed")
+    elif unmanaged_position_store is not None:
+        _log.debug(
+            "reconcile_core_holdings: unmanaged_position resolve skipped "
+            "(non-authoritative broker balance view)"
+        )
     # req-0007 / ac-0024: retry pending_resolved for B-case identities no longer mismatched.
     if alarm_emitter is not None and notification_store is not None:
         try:

trex_core/runtime/cli.py

@@ -1016,7 +1016,7 @@ def _cmd_unmanaged_list(args, cfg) -> int:
         symbol=args.symbol,
     )
     payload = {
-        "instance": instance,
+        "instance_key": instance,
         "status": args.status,
         "count": len(states),
         "states": states,