fix(broker): enforce operator broker truth boundary

trex_core/adapters/broker/kis.py

@@ -18,6 +18,7 @@ import httpx
 
 from trex_core.application.broker_port import (
     BalanceItem,
+    BrokerBalanceError,
     FillStatus,
     OrderAck,
     OrderRequest,
@@ -285,8 +286,9 @@ class KISBroker:
         """KIS 주식잔고조회 (`TTTC8434R`/`VTTC8434R`).
 
         Aggregates `output1[*].pdno` + `hldg_qty`, drops zero-qty rows
-        (D-2 leftover, Design Decision 7). Network/credential errors map to
-        an empty list (caller treats as no broker positions).
+        (D-2 leftover, Design Decision 7). Only a validated `rt_cd=0` response
+        may return an empty list; token/HTTP/shape/KIS failures raise a sanitized
+        :class:`BrokerBalanceError` so reconciliation cannot infer false zero.
         """
         cano, product = self._account()
         tr_id = "VTTC8434R" if self._mock else "TTTC8434R"
@@ -298,21 +300,44 @@ class KISBroker:
             "PRCS_DVSN": "01",
             "CTX_AREA_FK100": "", "CTX_AREA_NK100": "",
         }
+        try:
+            headers = self._headers(tr_id)
+        except (httpx.HTTPError, KeyError, TypeError, ValueError):
+            # Token issuance/header construction is distinct from a rejected
+            # balance endpoint response and must never look like an empty book.
+            raise BrokerBalanceError("token_issuance_failed") from None
         try:
             resp = httpx.get(
                 f"{self._base_url}/uapi/domestic-stock/v1/trading/inquire-balance",
-                headers=self._headers(tr_id), params=params, timeout=self._timeout,
+                headers=headers, params=params, timeout=self._timeout,
             )
+            resp.raise_for_status()
             payload = resp.json()
-        except (httpx.HTTPError, ValueError):
-            return []
+        except httpx.HTTPStatusError as exc:
+            status = exc.response.status_code if exc.response is not None else None
+            code = "authentication_failed" if status == 401 else (
+                "authorization_failed" if status == 403 else "transport_failed"
+            )
+            raise BrokerBalanceError(code) from None
+        except httpx.HTTPError:
+            raise BrokerBalanceError("transport_failed") from None
+        except (ValueError, TypeError):
+            raise BrokerBalanceError("malformed_response") from None
+        if not isinstance(payload, dict):
+            raise BrokerBalanceError("malformed_response")
         if str(payload.get("rt_cd")) != "0":
-            return []
-        rows = payload.get("output1") or []
+            raise BrokerBalanceError("broker_rejected")
+        rows = payload.get("output1")
+        if not isinstance(rows, list):
+            raise BrokerBalanceError("malformed_response")
         out: list[BalanceItem] = []
         for row in rows:
+            if not isinstance(row, dict):
+                raise BrokerBalanceError("malformed_response")
             symbol = str(row.get("pdno") or "").strip()
-            qty = _safe_int(row.get("hldg_qty")) or 0
+            qty = _safe_int(row.get("hldg_qty"))
+            if qty is None:
+                raise BrokerBalanceError("malformed_response")
             if not symbol or qty <= 0:
                 continue
             out.append(BalanceItem(symbol=symbol, hldg_qty=qty))

trex_core/adapters/persistence/postgresql.py

@@ -134,6 +134,18 @@ class PostgreSQLSlotRepository:
                 )
                 return [self._row_to_slot(r) for r in cur.fetchall()]
 
+    def find_managed_slot_by_symbol(self, instance_key: str, symbol: str) -> Optional[Slot]:
+        with self._psycopg.connect(self._dsn) as conn:
+            with conn.cursor() as cur:
+                cur.execute(
+                    f"SELECT {_SLOT_COLUMNS} FROM daily_slot "
+                    "WHERE instance_key = %s AND symbol = %s "
+                    "ORDER BY updated_at DESC NULLS LAST, slot_id ASC LIMIT 1",
+                    (instance_key, symbol),
+                )
+                row = cur.fetchone()
+                return self._row_to_slot(row) if row else None
+
     def reconcile_universe(
         self, instance_key: str, universe: dict[str, str],
         *, allow_retire: bool = True,

trex_core/application/broker_port.py

@@ -13,6 +13,21 @@ from decimal import Decimal
 from typing import Any, Optional, Protocol
 
 
+class BrokerBalanceError(RuntimeError):
+    """Sanitized failure distinct from a validated empty broker balance."""
+
+    _CODES = frozenset({
+        "token_issuance_failed", "authentication_failed", "authorization_failed",
+        "transport_failed", "malformed_response", "broker_rejected",
+    })
+
+    def __init__(self, code: str) -> None:
+        if code not in self._CODES:
+            raise ValueError(f"unsupported broker balance error code: {code}")
+        self.code = code
+        super().__init__(f"broker balance unavailable ({code})")
+
+
 @dataclass(frozen=True)
 class OrderRequest:
     slot_id: str          # originating slot (audit/correlation)

trex_core/application/reconciliation.py

@@ -58,7 +58,8 @@ def reconcile_core_holdings(
                      → set by reconcile_positions
 
     warn_store: Optional — when None, WARN-log-only (backward compat, REQ-103 AC-5).
-    Returns {key: 'error'|'qty_warning'|'ok'|'unmanaged_warning'} for telemetry.
+    Returns {key: 'error'|'qty_warning'|'ok'|'unmanaged_position'|
+    'managed_non_holding_divergence'} for telemetry.
     """
     broker_qty: dict[str, int] = {
         item.symbol: int(item.hldg_qty) for item in broker.get_balance()
@@ -114,13 +115,23 @@ def reconcile_core_holdings(
                     )
         else:
             decisions[str(slot.slot_id)] = "ok"
-    for sym, bq in broker_qty.items():  # unmanaged broker position
+    for sym, bq in broker_qty.items():  # unmatched broker position
         if bq > 0 and sym not in matched:
-            decisions[f"broker:{sym}"] = "unmanaged_warning"
-            _log.warning(
-                "broker holds %s qty=%d with no core holding slot (unmanaged position)",
-                sym, bq,
-            )
+            managed = slot_repo.find_managed_slot_by_symbol(instance_key, sym)
+            if managed is not None:
+                decisions[f"broker:{sym}"] = "managed_non_holding_divergence"
+                _log.warning(
+                    "managed non-holding divergence: slot_id=%s portfolio_id=%s "
+                    "symbol=%s status=%s instance_key=%s",
+                    managed.slot_id, managed.portfolio_id, sym,
+                    managed.status.value, instance_key,
+                )
+            else:
+                decisions[f"broker:{sym}"] = "unmanaged_position"
+                _log.warning(
+                    "unmanaged position: symbol=%s qty=%d instance_key=%s",
+                    sym, bq, instance_key,
+                )
     return decisions
 
 

trex_core/application/snapshot_scheduler.py

@@ -51,6 +51,7 @@ class SnapshotScheduler:
         warn_store: Optional[ReconcileWarningStore] = None,
         credential_fetch_enabled: bool = False,
         kis_reinit_callback: Optional[Callable[[BrokerCredentialPayload], None]] = None,
+        credential_refresh: Optional[Callable[[], object]] = None,
     ) -> None:
         self._read = read_port
         self._store = store
@@ -77,6 +78,8 @@ class SnapshotScheduler:
         # KISBroker with the fetched day-start credential. None = no reinit (paper
         # broker or credential_fetch_enabled=False). Set by daemon.py assembly.
         self._kis_reinit_callback: Optional[Callable[[BrokerCredentialPayload], None]] = kis_reinit_callback
+        # REQ-119: provider-owned refresh supersedes the legacy raw fetch hook.
+        self._credential_refresh = credential_refresh
         # REQ-113: detect break-glass at init time (env var set by operator).
         self._break_glass: bool = (
             os.environ.get("TREX_KIS_BROKER_BREAK_GLASS", "").lower()
@@ -250,7 +253,17 @@ class SnapshotScheduler:
         # Must run before portfolio snapshot so entry_allowed is set before
         # the first BuyIntent can arrive from the strategy daemon.
         if self._credential_fetch_enabled:
-            self._maybe_fetch_credential()
+            if self._credential_refresh is None:
+                self._maybe_fetch_credential()
+            else:
+                try:
+                    self._credential_refresh()
+                    self._entry_allowed = True
+                except Exception as exc:  # provider exposes only sanitized failure
+                    _log.error("DAY_START_LOAD_FAILED: broker credential refresh failed: %s (instance=%s)",
+                               exc.__class__.__name__, self._instance_key)
+                    self._entry_allowed = False
+                    self._emit_day_start_alarm(reason="broker_credential_unavailable")
 
         if self._daemon_id is None:
             try:

trex_core/domain/slot/in_memory_repository.py

@@ -89,6 +89,19 @@ class InMemorySlotRepository:
             if s.instance_key == instance_key and s.status is SlotState.HOLDING
         ]
 
+    def find_managed_slot_by_symbol(self, instance_key: str, symbol: str) -> Optional[Slot]:
+        candidates = [
+            s for s in self._slots.values()
+            if s.instance_key == instance_key and s.symbol == symbol
+        ]
+        # PostgreSQL parity: updated_at DESC NULLS LAST, slot_id ASC.
+        candidates.sort(key=lambda s: (
+            s.updated_at is None,
+            -s.updated_at.timestamp() if s.updated_at is not None else 0,
+            str(s.slot_id),
+        ))
+        return candidates[0] if candidates else None
+
     def reconcile_universe(
         self, instance_key: str, universe: dict[str, str],
         *, allow_retire: bool = True,

trex_core/domain/slot/repository.py

@@ -72,6 +72,13 @@ class SlotRepository(Protocol):
         reconcile; safety-sell / kill-switch sweeps)."""
         ...
 
+    def find_managed_slot_by_symbol(self, instance_key: str, symbol: str) -> Optional[Slot]:
+        """Return one deterministic managed slot for exact instance+symbol.
+
+        Read-only reconciliation triage; this must not imply remediation.
+        """
+        ...
+
     def reconcile_universe(
         self, instance_key: str, universe: dict[str, str],
         *, allow_retire: bool = True,

trex_core/runtime/broker_provider.py

@@ -0,0 +1,147 @@
+"""Common broker credential resolution and lifecycle boundary (REQ-119/120)."""
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from datetime import date, datetime, timedelta, timezone
+from typing import Literal, Protocol
+
+from trex_core.adapters.broker.paper import PaperBroker
+from trex_core.application.broker_port import BrokerPort
+from trex_core.application.portfolio import PortfolioHubUnavailableError
+from trex_core.runtime.config import DaemonConfig
+
+_KST = timezone(timedelta(hours=9))
+
+
+@dataclass(frozen=True)
+class BrokerSource:
+    """Non-secret credential-source metadata safe for CLI/log output."""
+    mode: Literal["env", "fetched", "break_glass"]
+    identity_label: str
+
+
+class BrokerCredentialUnavailable(RuntimeError):
+    _CODES = frozenset({"portfolio_hub_unavailable", "broker_absent", "trading_disabled", "malformed_credential", "broker_initialization_failed"})
+
+    def __init__(self, code: str) -> None:
+        if code not in self._CODES:
+            raise ValueError(f"unsupported broker credential error code: {code}")
+        self.code = code
+        super().__init__("broker credential unavailable")
+
+
+class BrokerProvider(Protocol):
+    def get_broker_for_read(self) -> tuple[BrokerPort, BrokerSource]: ...
+    def refresh_day_start_credential(self) -> BrokerSource: ...
+
+
+def _break_glass_enabled() -> bool:
+    return os.environ.get("TREX_KIS_BROKER_BREAK_GLASS", "").lower() in {"1", "true", "yes"}
+
+
+def _make_env_broker(config: DaemonConfig) -> BrokerPort:
+    if config.broker != "kis":
+        seq_store = None
+        if config.paper.redis_url:
+            from trex_core.adapters.broker.seq_store import RedisSeqStore
+            seq_store = RedisSeqStore(config.paper.redis_url, key=f"trex:broker:paper:{config.instance_key}:order-seq")
+        return PaperBroker(fill_after_polls=config.paper.fill_after_polls, qty_per_poll=config.paper.qty_per_poll,
+                           reject_symbols=set(config.paper.reject_symbols), seq_store=seq_store)
+    from trex_core.adapters.broker.kis import KISBroker
+    k = config.kis
+    if k is None:
+        raise BrokerCredentialUnavailable("broker_initialization_failed")
+    token_cache = None
+    if k.redis_url:
+        from trex_core.adapters.broker.token_cache import RedisTokenCache
+        token_cache = RedisTokenCache(k.redis_url)
+    return KISBroker(k.account_ref, app_key=k.app_key, app_secret=k.app_secret, base_url=k.base_url,
+                     timeout=k.timeout, allow_live=k.allow_live, token_cache=token_cache)
+
+
+class ConfigBrokerProvider:
+    """One provider owns one process/command broker and fetched-source refresh."""
+    def __init__(self, config: DaemonConfig, *, read_port=None, clock=None) -> None:
+        self._config = config
+        self._read = read_port
+        self._clock = clock or (lambda: datetime.now(tz=timezone.utc))
+        self._broker: BrokerPort | None = None
+        self._source: BrokerSource | None = None
+        self._credential_date: date | None = None
+
+    def _read_port(self):
+        if self._read is None:
+            if self._config.portfolio_hub is None:
+                raise BrokerCredentialUnavailable("portfolio_hub_unavailable")
+            from trex_core.adapters.portfolio.hub_read import PortfolioHubReadAdapter
+            p = self._config.portfolio_hub
+            self._read = PortfolioHubReadAdapter(p.base_url, token=p.token or None, timeout=p.timeout)
+        return self._read
+
+    def _fetch(self) -> BrokerSource:
+        try:
+            ctx = self._read_port().fetch_broker_credential()
+        except PortfolioHubUnavailableError:
+            raise BrokerCredentialUnavailable("portfolio_hub_unavailable") from None
+        except BrokerCredentialUnavailable:
+            raise
+        except Exception:
+            raise BrokerCredentialUnavailable("portfolio_hub_unavailable") from None
+        if ctx is None or getattr(ctx, "broker", None) is None:
+            raise BrokerCredentialUnavailable("broker_absent")
+        if not getattr(ctx, "trading_enabled", False):
+            raise BrokerCredentialUnavailable("trading_disabled")
+        credential = ctx.broker
+        if not all(isinstance(getattr(credential, x, None), str) and getattr(credential, x).strip()
+                   for x in ("account_ref", "app_key", "app_secret")):
+            raise BrokerCredentialUnavailable("malformed_credential")
+        try:
+            if self._broker is None:
+                broker = _make_env_broker(self._config)
+                reinitialize = getattr(broker, "reinitialize", None)
+                if not callable(reinitialize):
+                    raise RuntimeError("KIS broker lacks credential reload")
+                reinitialize(credential)
+                self._broker = broker
+            else:
+                self._broker.reinitialize(credential)  # type: ignore[attr-defined]
+        except Exception:
+            raise BrokerCredentialUnavailable("broker_initialization_failed") from None
+        self._credential_date = self._clock().astimezone(_KST).date()
+        # Version is designated non-secret by the portfolio contract; no account/key material.
+        return BrokerSource("fetched", f"credential-version:{getattr(credential, 'credential_version', 'unknown')}")
+
+    def get_broker_for_read(self) -> tuple[BrokerPort, BrokerSource]:
+        if self._config.broker != "kis":
+            if self._broker is None:
+                self._broker = _make_env_broker(self._config)
+            self._source = BrokerSource("env", "paper-config")
+            return self._broker, self._source
+        if not self._config.credential_fetch_enabled:
+            if self._broker is None:
+                self._broker = _make_env_broker(self._config)
+            self._source = BrokerSource("env", "kis-config")
+            return self._broker, self._source
+        if _break_glass_enabled():
+            if self._broker is None:
+                self._broker = _make_env_broker(self._config)
+            self._source = BrokerSource("break_glass", "operator-break-glass")
+            return self._broker, self._source
+        source = self._fetch()
+        self._source = source
+        return self._broker, source  # type: ignore[return-value]
+
+    def refresh_day_start_credential(self) -> BrokerSource:
+        # Paper/env/break-glass paths deliberately do no portfolio-hub call.
+        if self._config.broker != "kis" or not self._config.credential_fetch_enabled or _break_glass_enabled():
+            return self.get_broker_for_read()[1]
+        today = self._clock().astimezone(_KST).date()
+        if self._broker is not None and self._credential_date == today and self._source is not None:
+            return self._source
+        _, source = self.get_broker_for_read()
+        return source
+
+
+def build_broker_provider(config: DaemonConfig, *, read_port=None) -> ConfigBrokerProvider:
+    return ConfigBrokerProvider(config, read_port=read_port)

trex_core/runtime/cli.py

@@ -329,6 +329,29 @@ def _fail(message: str, as_json: bool) -> None:
         print(f"error: {message}", file=sys.stderr)
 
 
+def _broker_read_failure(exc, as_json: bool) -> int:
+    """REQ-120 command boundary: expose code only, never upstream payloads."""
+    code = getattr(exc, "code", "broker_unavailable")
+    payload = {"error": {"code": code, "message": "broker balance unavailable"}}
+    if as_json:
+        print(_json.dumps(payload), file=sys.stderr)
+    else:
+        print(f"broker balance unavailable ({code})", file=sys.stderr)
+    return 1
+
+
+class _CachedBalanceBroker:
+    """One validated balance read per operator invocation; delegates everything else."""
+    def __init__(self, broker, balance) -> None:
+        self._broker, self._balance = broker, balance
+
+    def get_balance(self):
+        return self._balance
+
+    def __getattr__(self, name):
+        return getattr(self._broker, name)
+
+
 # --------------------------------------------------------------------------- #
 # arg parsing
 # --------------------------------------------------------------------------- #
@@ -446,7 +469,6 @@ def _build_parser() -> argparse.ArgumentParser:
 
 def _cmd_sync(args, cfg, confirm) -> int:
     """REQ-105 (dcy-034, BAT-051): sync slot qty to broker ground truth."""
-    import psycopg
 
     from trex_core.adapters.portfolio.reconcile_warning_store import (
         PostgreSQLReconcileWarningStore,
@@ -476,11 +498,18 @@ def _cmd_sync(args, cfg, confirm) -> int:
                 print("aborted")
                 return 1
 
-    warn_store = PostgreSQLReconcileWarningStore(cfg.dsn)
+    # REQ-119/120: resolve and validate broker truth before opening/enumerating DB.
+    from trex_core.runtime.broker_provider import (
+        BrokerCredentialUnavailable, build_broker_provider,
+    )
+    from trex_core.application.broker_port import BrokerBalanceError
+    try:
+        broker_port, _source = build_broker_provider(cfg).get_broker_for_read()
+        broker_port = _CachedBalanceBroker(broker_port, broker_port.get_balance())
+    except (BrokerCredentialUnavailable, BrokerBalanceError) as exc:
+        return _broker_read_failure(exc, args.json)
 
-    # Build broker port
-    from trex_core.runtime.daemon import _build_broker
-    broker_port = _build_broker(cfg)
+    warn_store = PostgreSQLReconcileWarningStore(cfg.dsn)
 
     # Build slot_repo + slot_service
     from trex_core.adapters.persistence.postgresql import PostgreSQLSlotRepository
@@ -498,6 +527,7 @@ def _cmd_sync(args, cfg, confirm) -> int:
         )
 
     # Determine target slot_ids
+    import psycopg
     if args.portfolio_id:
         with psycopg.connect(cfg.dsn) as conn:
             with conn.cursor() as cur:
@@ -564,7 +594,7 @@ def _cmd_sync(args, cfg, confirm) -> int:
                 print(f"         after:  status={after.get('status')} "
                       f"qty={after.get('qty')} avg_price={after.get('avg_price')}")
                 if cleared:
-                    print(f"         reconcile-warning CLEARED")
+                    print("         reconcile-warning CLEARED")
 
     return exit_code
 
@@ -579,12 +609,20 @@ def _cmd_verify(args, cfg) -> int:
     from trex_core.application.reconciliation import verify_slot_reconcile
 
     instance = _resolve_instance(args)
+    # REQ-120 no-write/no-enumeration barrier also applies to verify.
+    from trex_core.runtime.broker_provider import (
+        BrokerCredentialUnavailable, build_broker_provider,
+    )
+    from trex_core.application.broker_port import BrokerBalanceError
+    try:
+        broker_port, _source = build_broker_provider(cfg).get_broker_for_read()
+        broker_port = _CachedBalanceBroker(broker_port, broker_port.get_balance())
+    except (BrokerCredentialUnavailable, BrokerBalanceError) as exc:
+        return _broker_read_failure(exc, args.json)
+
     warn_store = PostgreSQLReconcileWarningStore(cfg.dsn)
     slot_repo = PostgreSQLSlotRepository(cfg.dsn)
 
-    from trex_core.runtime.daemon import _build_broker
-    broker_port = _build_broker(cfg)
-
     # Build a minimal snapshot store (we only need asset_quantity for C-case check).
     # For verify we use a fresh SnapshotStore populated from warn_store metadata.
     # The hub_qty is read from the snapshot; if snapshot is empty we fall back to 0
@@ -787,7 +825,13 @@ def main(argv: list[str] | None = None, *, rpc=None, confirm=None) -> int:
         else None
     )
     _load_dotenv(_toml, getattr(args, "env_file", None))
-    cfg = DaemonConfig.from_env()
+    cfg = (
+        DaemonConfig.from_config(
+            resolve_config_toml(args.config, args.config_path), env_file=args.env_file
+        )
+        if args.config
+        else DaemonConfig.from_env()
+    )
     confirm = confirm or _default_confirm
 
     # REQ-088 (dcy-029): operator report — direct-DB point-in-time aggregate.

trex_core/runtime/daemon.py

@@ -74,6 +74,7 @@ _log = logging.getLogger(__name__)
 
 
 def _build_broker(config: DaemonConfig) -> BrokerPort:
+    """Backward-compatible env-only factory; new callers use broker_provider."""
     if config.broker == "kis":
         from trex_core.adapters.broker.kis import KISBroker
 
@@ -180,7 +181,7 @@ class Daemon:
             instance_key=config.instance_key,
             strategy_state_store=self.strategy_state_store,
         )
-        self.broker = _build_broker(config)
+
         self.outbox = PostgreSQLLedgerOutbox(config.dsn)
         # BAT-008: portfolio-hub read + snapshot store + durable pause + scheduler.
         # (Created before the ledger writer so option-D write-back can target it.)
@@ -193,6 +194,11 @@ class Daemon:
         # BAT-012: day-scoped kill-switch exclusion store + service.
         self.killswitch_store = PostgreSQLKillSwitchStore(config.dsn)
         self.portfolio_read = _build_portfolio_read(config)
+        # REQ-119: one provider owns source resolution and the long-lived broker.
+        # Fetched mode never starts from an implicit env broker.
+        from trex_core.runtime.broker_provider import build_broker_provider
+        self.broker_provider = build_broker_provider(config, read_port=self.portfolio_read)
+        self.broker, self.broker_source = self.broker_provider.get_broker_for_read()
 
         # Event publisher transport — one dedicated instance for the
         # execution-loop thread (pika is not thread-safe).
@@ -237,6 +243,7 @@ class Daemon:
             alarm_emitter=self.alarm_emitter,  # dcy-033 / REQ-100
             warn_store=self.warn_store,  # BAT-050 / REQ-103: durable reconcile-warning
             credential_fetch_enabled=config.credential_fetch_enabled,  # dcy-036 / REQ-113
+            credential_refresh=self.broker_provider.refresh_day_start_credential,
             # dcy-036 / REQ-113 AC-2 / REQ-115 AC-1: KISBroker reinit callback.
             # None when broker!=kis or credential_fetch_enabled=False (backward compat).
             kis_reinit_callback=_build_kis_reinit_callback(config, self.broker),