args = ()
kwargs = {'account_nick': 'cuixia', 'action': 'option_open', 'date': datetime.datetime(2026, 8, 14, 23, 21, 12, 277850, tzinfo=zoneinfo.ZoneInfo(key='America/New_York')), 'dispatch': 'manual-option-schedule', ...}
_sanitize_request_by = <function add_trading_plan.<locals>._sanitize_request_by at 0x13c874e00>
_symbol = 'OPTION', _action = 'option_open'
_VALID_ACTIONS = {'buy', 'cover', 'hold', 'option_open', 'reap_call', 'reap_put', ...}
_shares = 1, _account = 'cuixia', _ZI = <class 'zoneinfo.ZoneInfo'>
_ET = zoneinfo.ZoneInfo(key='America/New_York')
date_value = datetime.datetime(2026, 8, 17, 0, 0, tzinfo=zoneinfo.ZoneInfo(key='America/New_York'))
_sched = <module 'rtrader.scheduler' from '/Users/cao/robinhood-pm-93df26abc3d5/rtrader/scheduler.py'>

    @logger_wraps()
    def add_trading_plan(*args, **kwargs):
        def _sanitize_request_by(value: Any) -> Optional[str]:
            if value is None:
                return None
            try:
                text = str(value).strip()
            except Exception:  # swallow-ok: defensive coercion of arbitrary input; None is the sentinel
                return None
            if not text:
                return None
            return text[:120]
    
        # ── Type guards — catch bad parameters early instead of corrupting MongoDB ──
        _symbol = kwargs.get("symbol")
        if not _symbol or not isinstance(_symbol, str):
            raise RuntimeError(f"symbol must be a non-empty string, got {type(_symbol)}: {_symbol!r}")
    
        _action = kwargs.get("action")
        if not _action or not isinstance(_action, str):
            raise RuntimeError(f"action must be a non-empty string, got {type(_action)}: {_action!r}")
        _VALID_ACTIONS = {"buy", "sell", "hold", "cover", "short", "snipe_call", "snipe_put", "reap_call", "reap_put", "roll_call", "roll_put", "option_open"}
        if _action.lower() not in _VALID_ACTIONS:
            raise RuntimeError(f"action must be one of {_VALID_ACTIONS}, got {_action!r}")
    
        _shares = kwargs.get("shares", 0)
        if not isinstance(_shares, (int, float)):
            raise RuntimeError(f"shares must be numeric, got {type(_shares)}: {_shares!r}")
    
        _account = kwargs.get("account_nick")
        if not _account or not isinstance(_account, str):
            raise RuntimeError(f"account_nick must be a non-empty string, got {type(_account)}: {_account!r}")
    
        # Parse and validate date — MUST be tz-aware datetime for MongoDB comparisons.
        # Naked date objects and naive datetimes cause tz mismatch crashes downstream.
        from zoneinfo import ZoneInfo as _ZI
        _ET = _ZI("America/New_York")
        date_value = kwargs.get("date")
        if date_value is None:
            raise RuntimeError("date is required for add_trading_plan")
        if isinstance(date_value, str):
            try:
                date_value = datetime.fromisoformat(date_value)
            except ValueError:
                date_value = datetime.strptime(date_value, "%Y-%m-%d")
            # String-parsed datetimes are naive — attach ET timezone
            if date_value.tzinfo is None:
                date_value = date_value.replace(tzinfo=_ET)
        elif isinstance(date_value, date) and not isinstance(date_value, datetime):
            # date object → convert to datetime with tzinfo=ET
            date_value = datetime.combine(date_value, datetime.min.time(), tzinfo=_ET)  # tz-ok
        elif isinstance(date_value, datetime) and date_value.tzinfo is None:
            # Naive datetime → attach ET
            date_value = date_value.replace(tzinfo=_ET)
        if not isinstance(date_value, datetime):
            raise RuntimeError(f"date must be datetime/date/str, got {type(date_value)}: {date_value!r}")
    
        # Auto-advance plan date when today's session has CLOSED (post-close
        # creation can't fire today). Plans created BEFORE today's open should
        # stay dated today — `is_after_today_trading_hour()` distinguishes the
        # two; the old `not is_market_open()` conflated them and incorrectly
        # advanced before-open plans by one day.
        from rtrader import scheduler as _sched
        _now_et = datetime.now(_ET)
        _plan_date_et = date_value.astimezone(_ET).date()
        _today_et = _now_et.date()
        if _plan_date_et == _today_et and _sched.is_after_today_trading_hour(_now_et):
            # Use get_next_plan_session for holiday-aware advance.
            # Function takes no args — derives "now" internally via datetime.now(ET).
            _next_td = _sched.get_next_plan_session()
            date_value = datetime.combine(_next_td, datetime.min.time(), tzinfo=_ET)  # tz-ok
            logger.warning(
                "[add_trading_plan] Auto-advanced plan date {} → {} "
                "(today's session closed, next plan session is {}; "
                "set strict_date=True to disable)",
                _today_et, _next_td, _next_td,
            )
    
        raw_metadata = kwargs.get("metadata")
    
        with DBAccounting() as mongo:
            rr = mongo.collection("TradingPlan")
            alpaca_mode = normalize_alpaca_mode(kwargs.get("alpaca_mode", "live"))
            # Sanitize action window values (ensure non-null, sensible bounds)
            raw_action_start = kwargs.get("action_start")
            try:
                action_start_value = int(raw_action_start)
            except (TypeError, ValueError):
                action_start_value = 1
            if action_start_value is None or action_start_value <= 0:
                action_start_value = 1
            if action_start_value > 390:
                action_start_value = 390
    
            raw_action_end = kwargs.get("action_end")
            try:
                action_end_value = int(raw_action_end) if raw_action_end is not None else 390
            except (TypeError, ValueError):
                action_end_value = 390
            if action_end_value is None or action_end_value < action_start_value:
                action_end_value = 390
    
            record = {
                "symbol": kwargs.get("symbol", "").upper(),
                "action": kwargs.get("action", "").lower(),
                "date": date_value,
                "shares": kwargs.get("shares", 0),
                "amount": kwargs.get("amount", 0),
                "price": kwargs.get("price", 0.0),
                "limit_price": kwargs.get("limit_price", None),
                "notes": kwargs.get("notes", ""),
                "status": kwargs.get("status", "review"),
                "action_start": action_start_value,
                "action_end": action_end_value,
                "created_at": datetime.now(),
                "updated_at": datetime.now(),
                "trigger": kwargs.get("trigger", ""),
                "price_operator": kwargs.get("price_operator", ""),
                "price_limit": kwargs.get("price_limit", 0.0),
                "level_operator": kwargs.get("level_operator", ""),
                "level_limit": kwargs.get("level_limit", 0.0),
                "log": kwargs.get("log", ""),
                "alpaca_mode": alpaca_mode,
                "account_nick": kwargs.get("account_nick") or "",
                "broker_backend": kwargs.get("broker_backend") or "",
            }
    
            # Dispatch tag — identifies which service/workflow created the plan.
            # Used for categorization (eod vs scalp vs squeeze), lock bypass,
            # and preventing duplicate plan creation.
            #
            # Dispatch format conventions:
            #   eod-buy-{account}-{SYMBOL}                    — EOD top-pick buy (buy_trader_executor)
            #   eod-stop-refresh-{account}-{SYMBOL}           — daily stop-loss refresh (portfolio_risk_monitor)
            #   verdict-{action}-{symbol}-{account}           — LLM verdict sell/trim/exit/add (dispatch_llm_verdicts)
            #   verdict-ah-add-{symbol}-{account}             — after-hours institutional override ADD
            #   verdict-profit-protect-{symbol}-{account}     — profit protection stop-loss
            #   scalp-pdt-exit-{job_id}-{type}                — scalp PDT next-day exit (eod/stop/signal/open)
            #   squeeze-short-vol-{symbol}-{account}          — short squeeze entry (short_volume_cli)
            dispatch = kwargs.get("dispatch")
            if not dispatch:
                raise RuntimeError(
                    f"dispatch is required for add_trading_plan (symbol={kwargs.get('symbol')}, "
                    f"action={kwargs.get('action')}, account={kwargs.get('account_nick')})"
                )
            record["dispatch"] = str(dispatch)
    
            flow_bias = kwargs.get("flow_bias")
            if flow_bias:
                record["flow_bias"] = str(flow_bias)
    
            signal_bias = kwargs.get("signal_bias")
            if signal_bias:
                record["signal_bias"] = str(signal_bias)
    
            instrument_type = kwargs.get("instrument_type")
            if instrument_type:
                record["instrument_type"] = str(instrument_type).lower()
    
            mechanism_type = kwargs.get("mechanism_type")
            if mechanism_type:
                record["mechanism_type"] = str(mechanism_type)
    
            mechanism_mode = kwargs.get("mechanism_mode")
            if mechanism_mode:
                record["mechanism_mode"] = str(mechanism_mode)
    
            contract_symbol = kwargs.get("contract_symbol")
            if contract_symbol:
                record["contract_symbol"] = str(contract_symbol).upper()
    
            contract_mode = kwargs.get("contract_mode")
            if contract_mode:
                record["contract_mode"] = str(contract_mode)
    
            metadata = dict(raw_metadata) if isinstance(raw_metadata, dict) else raw_metadata
            sanitized_metadata = None
            if metadata is not None:
                if isinstance(metadata, dict):
                    if "leg" in metadata and isinstance(metadata["leg"], str):
                        metadata["leg"] = metadata["leg"].strip()
                    if "account_id" in metadata and isinstance(metadata["account_id"], str):
                        metadata["account_id"] = metadata["account_id"].strip()
                sanitized_metadata = _metadata_json_safe(metadata)
            if isinstance(sanitized_metadata, dict):
                record["metadata"] = sanitized_metadata
            elif sanitized_metadata not in (None, "", [], (), {}):
                record["metadata"] = sanitized_metadata
    
            # Option-intent plan (bug 93df26abc3d5): a plan that carries an OPTION
            # INTENT, not a contract symbol. The executor resolves the actual contract
            # at trigger time against the live quote. Allowed fields are
            # underlying / side / contracts / limit_price / expected_expiry_days; the
            # executor validates them at trigger — keeping the create-time surface
            # narrow lets us evolve the schema later without breaking existing plans.
            raw_option_intent = kwargs.get("option_intent")
            if isinstance(raw_option_intent, dict) and raw_option_intent:
                from rtrader.option_intent import normalize_option_intent
    
                try:
                    normalized_intent = normalize_option_intent(raw_option_intent)
                except ValueError as exc:
                    raise RuntimeError(f"option_intent invalid: {exc}") from exc
                metadata_dict = record.get("metadata")
                if not isinstance(metadata_dict, dict):
                    metadata_dict = {}
                    record["metadata"] = metadata_dict
                metadata_dict["option_intent"] = normalized_intent
    
            # Trade-attribute ledger stash (2026-07-02): the "why we entered" —
            # conviction score/climb/tier, TIDE regime, snapshot features, verdict
            # reason. Carried on the plan so the fill-reconciler can join it to the
            # actual fill by COID and write the TradeAttributeLedger entry record.
            entry_attributes = kwargs.get("entry_attributes")
            if isinstance(entry_attributes, dict) and entry_attributes:
                record["entry_attributes"] = _metadata_json_safe(entry_attributes)
    
            metadata_for_compare = record.get("metadata") or {}
            normalized_leg = str(metadata_for_compare.get("leg") or "").strip().lower()
            if not normalized_leg:
                normalized_leg = None
            normalized_account = str(metadata_for_compare.get("account_id") or "").strip()
    
            request_by_value = _sanitize_request_by(
                kwargs.get("request_by")
                or kwargs.get("requested_by")
            )
            if not request_by_value and isinstance(record.get("metadata"), dict):
                metadata_lookup = record["metadata"]
                request_by_value = _sanitize_request_by(
                    metadata_lookup.get("request_by")
                    or metadata_lookup.get("requested_by")
                    or metadata_lookup.get("workflow")
                    or metadata_lookup.get("source")
                )
            if not request_by_value:
                request_by_value = _sanitize_request_by(kwargs.get("trigger"))
            if not request_by_value:
                request_by_value = "system"
    
            record["request_by"] = request_by_value
            if isinstance(record.get("metadata"), dict):
                metadata_map = record["metadata"]
                metadata_map.setdefault("request_by", request_by_value)
                metadata_map.setdefault("requested_by", request_by_value)
    
            flow_bias = kwargs.get("flow_bias")
            record["flow_bias"] = str(flow_bias).lower() if flow_bias else ""
    
            signal_bias = kwargs.get("signal_bias")
            record["signal_bias"] = str(signal_bias).lower() if signal_bias else ""
    
            # TIDE bias — fires the plan as a market order only while the live avg-stock
            # breadth regime favors this direction (long/short). Analog of flow_bias on
            # market-wide breadth. See trading_plan_executor.is_tide_condition_met.
            tide_bias = kwargs.get("tide_bias")
            record["tide_bias"] = str(tide_bias).lower() if tide_bias else ""
    
            price_target = kwargs.get("price_target")
            record["price_target"] = str(price_target).lower() if price_target else ""
    
            # Enforce: seller flow sell plans MUST use market orders.
            # Price is falling under seller pressure — limit orders won't fill.
            if record["flow_bias"] == "seller" and record["action"] == "sell":
                if "metadata" not in record or not isinstance(record.get("metadata"), dict):
                    record["metadata"] = {}
                if record["metadata"].get("order_type") != "market":
                    record["metadata"]["order_type"] = "market"
                    logger.info(
                        "[add_trading_plan] Auto-set order_type=market for seller flow sell plan: {}",
                        record["symbol"],
                    )
    
            flow_window = kwargs.get("flow_window_minutes")
            if flow_window not in (None, "", []):
                try:
                    record["flow_window_minutes"] = max(int(flow_window), 1)
                except (TypeError, ValueError):
                    pass
            # When flow_window_minutes is absent, the executor resolves it
            # dynamically based on SPY intraday volatility (atr_ratio).
    
            flow_delta = kwargs.get("flow_min_delta_pct")
            if flow_delta in (None, "", []):
                record["flow_min_delta_pct"] = float(getattr(settings, "BUYER_SELLER_FLOW_MIN_DELTA_PCT", 0.005))
            else:
                try:
                    delta_value = float(flow_delta)
                    if delta_value > 1:
                        delta_value = delta_value / 100.0
                    record["flow_min_delta_pct"] = max(delta_value, 0.0)
                except (TypeError, ValueError):
                    record["flow_min_delta_pct"] = float(getattr(settings, "BUYER_SELLER_FLOW_MIN_DELTA_PCT", 0.005))
    
            # --- Option gain/loss target fields ---
            entry_price = kwargs.get("entry_price")
            if entry_price is not None:
                try:
                    record["entry_price"] = float(entry_price)
                except (TypeError, ValueError):
                    pass
    
            gain_target_pct = kwargs.get("gain_target_pct")
            if gain_target_pct is not None:
                try:
                    val = float(gain_target_pct)
                    if val > 1:
                        val = val / 100.0
                    record["gain_target_pct"] = val
                except (TypeError, ValueError):
                    pass
    
            stop_loss_pct = kwargs.get("stop_loss_pct")
            if stop_loss_pct is not None:
                try:
                    val = float(stop_loss_pct)
                    if val > 1:
                        val = val / 100.0
                    record["stop_loss_pct"] = val
                except (TypeError, ValueError):
                    pass
    
            valid_until = kwargs.get("valid_until")
            if valid_until is not None:
                if isinstance(valid_until, str):
                    try:
                        valid_until = datetime.strptime(valid_until, "%Y-%m-%d")
                    except ValueError:
                        valid_until = datetime.fromisoformat(valid_until)
                if isinstance(valid_until, date) and not isinstance(valid_until, datetime):
                    valid_until = datetime.combine(valid_until, datetime.max.time())
                record["valid_until"] = valid_until
    
            # Duplicate detection prior to insertion
            # Use tz-aware bounds matching the stored date (date_value is always tz-aware ET)
            lookup_start = datetime.combine(date_value.date(), datetime.min.time(), tzinfo=date_value.tzinfo)  # tz-ok
            lookup_end = lookup_start + timedelta(days=1)  # calendar-ok
            base_query = {
                "symbol": record["symbol"],
                "action": record["action"],
                "date": {"$gte": lookup_start, "$lt": lookup_end},
                "account_nick": record["account_nick"],
            }
            potential = rr.find(base_query, {"metadata": 1, "shares": 1, "dispatch": 1, "status": 1})
            for existing in potential:
                existing_status = str(existing.get("status") or "")
                if existing_status in ("cancelled", "canceled"):
                    continue  # don't match against cancelled plans
                existing_metadata = existing.get("metadata") or {}
                existing_leg = str(existing_metadata.get("leg") or "").strip().lower() or None
                existing_account = str(existing_metadata.get("account_id") or "").strip()
                existing_shares = int(existing.get("shares") or 0)
                # Scalp PDT plans have different dispatch tags (Plan A/B/D/E) for the
                # same symbol — don't dedup them. Only dedup if dispatch also matches.
                existing_dispatch = str(existing.get("dispatch") or "").strip()
                record_dispatch = str(record.get("dispatch") or "").strip()
                # Shares match: required for most plans (different shares = intentional chunks).
                # Exception: verdict buy_back / scalp_accumulate re-runs recalculate shares
                # each time (price changes). For these, same dispatch = same intent regardless
                # of shares. The dispatch tag already encodes symbol+account uniquely.
                shares_match = existing_shares == int(record.get("shares") or 0)
                # Recompute-each-run dispatches: every re-run recalculates shares/stop from the CURRENT
                # position or price, so SAME dispatch = SAME intent regardless of share count. Dedup by
                # dispatch and UPDATE the existing pending plan in place (operator 2026-06-23: eod-stop-
                # refresh + verdict-profit-protect were inserting 100s of duplicate plans/day, bloating
                # the executor queue and delaying real fills).
                _recompute_dispatch = record_dispatch.startswith((
                    "verdict-buy_back-", "verdict-scalp-acc-", "verdict-add-",
                    "eod-stop-refresh-", "verdict-profit-protect-", "profit-protect-stop-",
                ))
                if (
                    existing_leg == normalized_leg
                    and existing_account == normalized_account
                    and (shares_match or _recompute_dispatch)
                    and existing_dispatch == record_dispatch
                ):
                    if _recompute_dispatch and not shares_match:
                        # Refresh the existing pending plan with the latest shares/stop instead of a dup.
                        _upd = {"shares": int(record.get("shares") or 0), "updated_at": datetime.now()}
                        for _k in ("price", "limit_price", "price_limit", "price_operator",
                                   "level_operator", "level_limit", "notes"):
                            if _k in record:
                                _upd[_k] = record[_k]
                        rr.update_one({"_id": existing.get("_id")}, {"$set": _upd})
                    logger.info(
                        "[add_trading_plan] Duplicate {}/{} {} dispatch={} shares={}vs{}; {} plan {}",
                        record["symbol"], record["action"], lookup_start.date(), record_dispatch,
                        existing_shares, int(record.get("shares") or 0),
                        "UPDATED" if (_recompute_dispatch and not shares_match) else "kept",
                        existing.get("_id"),
                    )
                    return existing.get("_id")
    
            # Assert data integrity before insert
            assert record["symbol"], "Symbol cannot be empty"
            assert record["action"] in ["buy", "sell", "hold", "cover", "short", "snipe_call", "snipe_put", "reap_call", "reap_put", "roll_call", "roll_put", "option_open"], f"Invalid action: {record['action']}"
            assert isinstance(record["date"], (datetime, date)), "Date must be datetime"
            assert record["shares"] >= 0, "Shares must be non-negative"
            assert record["action_start"] >= 0, "Action start must be non-negative"
            assert record["action_end"] > record["action_start"], "Action end must be after action start"
    
            metadata_keys = sorted((record.get("metadata") or {}).keys())
            logger.info(
                "[add_trading_plan] Inserting plan | symbol={} action={} shares={} amount={:.2f} price={:.2f} start={} end={} metadata_keys={}",
                record["symbol"],
                record["action"],
                record["shares"],
                float(record.get("amount", 0.0)),
                float(record.get("price", 0.0)),
                record.get("action_start"),
                record.get("action_end"),
                metadata_keys,
            )
    
            try:
                result = rr.insert_one(record)
    
                # Assert insertion succeeded
                assert result.inserted_id, "Failed to insert trading plan"
    
                # Verify the record was actually inserted
                inserted = rr.find_one({"_id": result.inserted_id})
>               assert inserted, f"Failed to verify inserted plan with ID {result.inserted_id}"
                       ^^^^^^^^
E               AssertionError: Failed to verify inserted plan with ID test-id

rtrader/database.py:5579: AssertionError

During handling of the above exception, another exception occurred:

monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x13bdb06b0>

    def test_add_trading_plan_accepts_option_intent_kwarg(
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """The create-time surface accepts an option_intent kwarg and persists
        it into metadata.option_intent. This is the contract the executor leg
        reads at trigger time."""
        from rtrader import database as database_module
    
        # Capture the record handed to Mongo — we don't want to write a real
        # doc to the DB in a unit test.
        captured: Dict[str, Any] = {}
    
        class _FakeCollection:
            def insert_one(self, record):
                captured.update(record)
                return types.SimpleNamespace(inserted_id="test-id")
    
            def find(self, *_a, **_kw):
                # Duplicate-detection lookup — return empty so no duplicate-match.
                return iter(())
    
            def find_one(self, *_a, **_kw):
                return None
    
        class _FakeMongo:
            def collection(self, name):
                assert name == "TradingPlan"
                return _FakeCollection()
    
            def __enter__(self):
                return self
    
            def __exit__(self, exc_type, exc_val, exc_tb):
                return False
    
        monkeypatch.setattr(database_module, "DBAccounting", lambda: _FakeMongo())
    
        from rtrader.option_intent import normalize_option_intent
    
        intent = normalize_option_intent({
            "underlying": "IWM",
            "side": "call",
            "contracts": 1,
        })
    
        from datetime import datetime
        from zoneinfo import ZoneInfo
        _now = datetime.now(ZoneInfo("America/New_York"))
    
>       database_module.add_trading_plan(
            symbol="OPTION",
            action="option_open",
            date=_now,
            shares=1,
            account_nick="cuixia",
            dispatch="manual-option-schedule",
            option_intent=intent,
        )

tests/services/test_option_intent_executor.py:726: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
rtrader/utils/log.py:121: in wrapped
    result = func(*args, **kwargs)
             ^^^^^^^^^^^^^^^^^^^^^
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = ()
kwargs = {'account_nick': 'cuixia', 'action': 'option_open', 'date': datetime.datetime(2026, 8, 14, 23, 21, 12, 277850, tzinfo=zoneinfo.ZoneInfo(key='America/New_York')), 'dispatch': 'manual-option-schedule', ...}
_sanitize_request_by = <function add_trading_plan.<locals>._sanitize_request_by at 0x13c874e00>
_symbol = 'OPTION', _action = 'option_open'
_VALID_ACTIONS = {'buy', 'cover', 'hold', 'option_open', 'reap_call', 'reap_put', ...}
_shares = 1, _account = 'cuixia', _ZI = <class 'zoneinfo.ZoneInfo'>
_ET = zoneinfo.ZoneInfo(key='America/New_York')
date_value = datetime.datetime(2026, 8, 17, 0, 0, tzinfo=zoneinfo.ZoneInfo(key='America/New_York'))
_sched = <module 'rtrader.scheduler' from '/Users/cao/robinhood-pm-93df26abc3d5/rtrader/scheduler.py'>

    @logger_wraps()
    def add_trading_plan(*args, **kwargs):
        def _sanitize_request_by(value: Any) -> Optional[str]:
            if value is None:
                return None
            try:
                text = str(value).strip()
            except Exception:  # swallow-ok: defensive coercion of arbitrary input; None is the sentinel
                return None
            if not text:
                return None
            return text[:120]
    
        # ── Type guards — catch bad parameters early instead of corrupting MongoDB ──
        _symbol = kwargs.get("symbol")
        if not _symbol or not isinstance(_symbol, str):
            raise RuntimeError(f"symbol must be a non-empty string, got {type(_symbol)}: {_symbol!r}")
    
        _action = kwargs.get("action")
        if not _action or not isinstance(_action, str):
            raise RuntimeError(f"action must be a non-empty string, got {type(_action)}: {_action!r}")
        _VALID_ACTIONS = {"buy", "sell", "hold", "cover", "short", "snipe_call", "snipe_put", "reap_call", "reap_put", "roll_call", "roll_put", "option_open"}
        if _action.lower() not in _VALID_ACTIONS:
            raise RuntimeError(f"action must be one of {_VALID_ACTIONS}, got {_action!r}")
    
        _shares = kwargs.get("shares", 0)
        if not isinstance(_shares, (int, float)):
            raise RuntimeError(f"shares must be numeric, got {type(_shares)}: {_shares!r}")
    
        _account = kwargs.get("account_nick")
        if not _account or not isinstance(_account, str):
            raise RuntimeError(f"account_nick must be a non-empty string, got {type(_account)}: {_account!r}")
    
        # Parse and validate date — MUST be tz-aware datetime for MongoDB comparisons.
        # Naked date objects and naive datetimes cause tz mismatch crashes downstream.
        from zoneinfo import ZoneInfo as _ZI
        _ET = _ZI("America/New_York")
        date_value = kwargs.get("date")
        if date_value is None:
            raise RuntimeError("date is required for add_trading_plan")
        if isinstance(date_value, str):
            try:
                date_value = datetime.fromisoformat(date_value)
            except ValueError:
                date_value = datetime.strptime(date_value, "%Y-%m-%d")
            # String-parsed datetimes are naive — attach ET timezone
            if date_value.tzinfo is None:
                date_value = date_value.replace(tzinfo=_ET)
        elif isinstance(date_value, date) and not isinstance(date_value, datetime):
            # date object → convert to datetime with tzinfo=ET
            date_value = datetime.combine(date_value, datetime.min.time(), tzinfo=_ET)  # tz-ok
        elif isinstance(date_value, datetime) and date_value.tzinfo is None:
            # Naive datetime → attach ET
            date_value = date_value.replace(tzinfo=_ET)
        if not isinstance(date_value, datetime):
            raise RuntimeError(f"date must be datetime/date/str, got {type(date_value)}: {date_value!r}")
    
        # Auto-advance plan date when today's session has CLOSED (post-close
        # creation can't fire today). Plans created BEFORE today's open should
        # stay dated today — `is_after_today_trading_hour()` distinguishes the
        # two; the old `not is_market_open()` conflated them and incorrectly
        # advanced before-open plans by one day.
        from rtrader import scheduler as _sched
        _now_et = datetime.now(_ET)
        _plan_date_et = date_value.astimezone(_ET).date()
        _today_et = _now_et.date()
        if _plan_date_et == _today_et and _sched.is_after_today_trading_hour(_now_et):
            # Use get_next_plan_session for holiday-aware advance.
            # Function takes no args — derives "now" internally via datetime.now(ET).
            _next_td = _sched.get_next_plan_session()
            date_value = datetime.combine(_next_td, datetime.min.time(), tzinfo=_ET)  # tz-ok
            logger.warning(
                "[add_trading_plan] Auto-advanced plan date {} → {} "
                "(today's session closed, next plan session is {}; "
                "set strict_date=True to disable)",
                _today_et, _next_td, _next_td,
            )
    
        raw_metadata = kwargs.get("metadata")
    
        with DBAccounting() as mongo:
            rr = mongo.collection("TradingPlan")
            alpaca_mode = normalize_alpaca_mode(kwargs.get("alpaca_mode", "live"))
            # Sanitize action window values (ensure non-null, sensible bounds)
            raw_action_start = kwargs.get("action_start")
            try:
                action_start_value = int(raw_action_start)
            except (TypeError, ValueError):
                action_start_value = 1
            if action_start_value is None or action_start_value <= 0:
                action_start_value = 1
            if action_start_value > 390:
                action_start_value = 390
    
            raw_action_end = kwargs.get("action_end")
            try:
                action_end_value = int(raw_action_end) if raw_action_end is not None else 390
            except (TypeError, ValueError):
                action_end_value = 390
            if action_end_value is None or action_end_value < action_start_value:
                action_end_value = 390
    
            record = {
                "symbol": kwargs.get("symbol", "").upper(),
                "action": kwargs.get("action", "").lower(),
                "date": date_value,
                "shares": kwargs.get("shares", 0),
                "amount": kwargs.get("amount", 0),
                "price": kwargs.get("price", 0.0),
                "limit_price": kwargs.get("limit_price", None),
                "notes": kwargs.get("notes", ""),
                "status": kwargs.get("status", "review"),
                "action_start": action_start_value,
                "action_end": action_end_value,
                "created_at": datetime.now(),
                "updated_at": datetime.now(),
                "trigger": kwargs.get("trigger", ""),
                "price_operator": kwargs.get("price_operator", ""),
                "price_limit": kwargs.get("price_limit", 0.0),
                "level_operator": kwargs.get("level_operator", ""),
                "level_limit": kwargs.get("level_limit", 0.0),
                "log": kwargs.get("log", ""),
                "alpaca_mode": alpaca_mode,
                "account_nick": kwargs.get("account_nick") or "",
                "broker_backend": kwargs.get("broker_backend") or "",
            }
    
            # Dispatch tag — identifies which service/workflow created the plan.
            # Used for categorization (eod vs scalp vs squeeze), lock bypass,
            # and preventing duplicate plan creation.
            #
            # Dispatch format conventions:
            #   eod-buy-{account}-{SYMBOL}                    — EOD top-pick buy (buy_trader_executor)
            #   eod-stop-refresh-{account}-{SYMBOL}           — daily stop-loss refresh (portfolio_risk_monitor)
            #   verdict-{action}-{symbol}-{account}           — LLM verdict sell/trim/exit/add (dispatch_llm_verdicts)
            #   verdict-ah-add-{symbol}-{account}             — after-hours institutional override ADD
            #   verdict-profit-protect-{symbol}-{account}     — profit protection stop-loss
            #   scalp-pdt-exit-{job_id}-{type}                — scalp PDT next-day exit (eod/stop/signal/open)
            #   squeeze-short-vol-{symbol}-{account}          — short squeeze entry (short_volume_cli)
            dispatch = kwargs.get("dispatch")
            if not dispatch:
                raise RuntimeError(
                    f"dispatch is required for add_trading_plan (symbol={kwargs.get('symbol')}, "
                    f"action={kwargs.get('action')}, account={kwargs.get('account_nick')})"
                )
            record["dispatch"] = str(dispatch)
    
            flow_bias = kwargs.get("flow_bias")
            if flow_bias:
                record["flow_bias"] = str(flow_bias)
    
            signal_bias = kwargs.get("signal_bias")
            if signal_bias:
                record["signal_bias"] = str(signal_bias)
    
            instrument_type = kwargs.get("instrument_type")
            if instrument_type:
                record["instrument_type"] = str(instrument_type).lower()
    
            mechanism_type = kwargs.get("mechanism_type")
            if mechanism_type:
                record["mechanism_type"] = str(mechanism_type)
    
            mechanism_mode = kwargs.get("mechanism_mode")
            if mechanism_mode:
                record["mechanism_mode"] = str(mechanism_mode)
    
            contract_symbol = kwargs.get("contract_symbol")
            if contract_symbol:
                record["contract_symbol"] = str(contract_symbol).upper()
    
            contract_mode = kwargs.get("contract_mode")
            if contract_mode:
                record["contract_mode"] = str(contract_mode)
    
            metadata = dict(raw_metadata) if isinstance(raw_metadata, dict) else raw_metadata
            sanitized_metadata = None
            if metadata is not None:
                if isinstance(metadata, dict):
                    if "leg" in metadata and isinstance(metadata["leg"], str):
                        metadata["leg"] = metadata["leg"].strip()
                    if "account_id" in metadata and isinstance(metadata["account_id"], str):
                        metadata["account_id"] = metadata["account_id"].strip()
                sanitized_metadata = _metadata_json_safe(metadata)
            if isinstance(sanitized_metadata, dict):
                record["metadata"] = sanitized_metadata
            elif sanitized_metadata not in (None, "", [], (), {}):
                record["metadata"] = sanitized_metadata
    
            # Option-intent plan (bug 93df26abc3d5): a plan that carries an OPTION
            # INTENT, not a contract symbol. The executor resolves the actual contract
            # at trigger time against the live quote. Allowed fields are
            # underlying / side / contracts / limit_price / expected_expiry_days; the
            # executor validates them at trigger — keeping the create-time surface
            # narrow lets us evolve the schema later without breaking existing plans.
            raw_option_intent = kwargs.get("option_intent")
            if isinstance(raw_option_intent, dict) and raw_option_intent:
                from rtrader.option_intent import normalize_option_intent
    
                try:
                    normalized_intent = normalize_option_intent(raw_option_intent)
                except ValueError as exc:
                    raise RuntimeError(f"option_intent invalid: {exc}") from exc
                metadata_dict = record.get("metadata")
                if not isinstance(metadata_dict, dict):
                    metadata_dict = {}
                    record["metadata"] = metadata_dict
                metadata_dict["option_intent"] = normalized_intent
    
            # Trade-attribute ledger stash (2026-07-02): the "why we entered" —
            # conviction score/climb/tier, TIDE regime, snapshot features, verdict
            # reason. Carried on the plan so the fill-reconciler can join it to the
            # actual fill by COID and write the TradeAttributeLedger entry record.
            entry_attributes = kwargs.get("entry_attributes")
            if isinstance(entry_attributes, dict) and entry_attributes:
                record["entry_attributes"] = _metadata_json_safe(entry_attributes)
    
            metadata_for_compare = record.get("metadata") or {}
            normalized_leg = str(metadata_for_compare.get("leg") or "").strip().lower()
            if not normalized_leg:
                normalized_leg = None
            normalized_account = str(metadata_for_compare.get("account_id") or "").strip()
    
            request_by_value = _sanitize_request_by(
                kwargs.get("request_by")
                or kwargs.get("requested_by")
            )
            if not request_by_value and isinstance(record.get("metadata"), dict):
                metadata_lookup = record["metadata"]
                request_by_value = _sanitize_request_by(
                    metadata_lookup.get("request_by")
                    or metadata_lookup.get("requested_by")
                    or metadata_lookup.get("workflow")
                    or metadata_lookup.get("source")
                )
            if not request_by_value:
                request_by_value = _sanitize_request_by(kwargs.get("trigger"))
            if not request_by_value:
                request_by_value = "system"
    
            record["request_by"] = request_by_value
            if isinstance(record.get("metadata"), dict):
                metadata_map = record["metadata"]
                metadata_map.setdefault("request_by", request_by_value)
                metadata_map.setdefault("requested_by", request_by_value)
    
            flow_bias = kwargs.get("flow_bias")
            record["flow_bias"] = str(flow_bias).lower() if flow_bias else ""
    
            signal_bias = kwargs.get("signal_bias")
            record["signal_bias"] = str(signal_bias).lower() if signal_bias else ""
    
            # TIDE bias — fires the plan as a market order only while the live avg-stock
            # breadth regime favors this direction (long/short). Analog of flow_bias on
            # market-wide breadth. See trading_plan_executor.is_tide_condition_met.
            tide_bias = kwargs.get("tide_bias")
            record["tide_bias"] = str(tide_bias).lower() if tide_bias else ""
    
            price_target = kwargs.get("price_target")
            record["price_target"] = str(price_target).lower() if price_target else ""
    
            # Enforce: seller flow sell plans MUST use market orders.
            # Price is falling under seller pressure — limit orders won't fill.
            if record["flow_bias"] == "seller" and record["action"] == "sell":
                if "metadata" not in record or not isinstance(record.get("metadata"), dict):
                    record["metadata"] = {}
                if record["metadata"].get("order_type") != "market":
                    record["metadata"]["order_type"] = "market"
                    logger.info(
                        "[add_trading_plan] Auto-set order_type=market for seller flow sell plan: {}",
                        record["symbol"],
                    )
    
            flow_window = kwargs.get("flow_window_minutes")
            if flow_window not in (None, "", []):
                try:
                    record["flow_window_minutes"] = max(int(flow_window), 1)
                except (TypeError, ValueError):
                    pass
            # When flow_window_minutes is absent, the executor resolves it
            # dynamically based on SPY intraday volatility (atr_ratio).
    
            flow_delta = kwargs.get("flow_min_delta_pct")
            if flow_delta in (None, "", []):
                record["flow_min_delta_pct"] = float(getattr(settings, "BUYER_SELLER_FLOW_MIN_DELTA_PCT", 0.005))
            else:
                try:
                    delta_value = float(flow_delta)
                    if delta_value > 1:
                        delta_value = delta_value / 100.0
                    record["flow_min_delta_pct"] = max(delta_value, 0.0)
                except (TypeError, ValueError):
                    record["flow_min_delta_pct"] = float(getattr(settings, "BUYER_SELLER_FLOW_MIN_DELTA_PCT", 0.005))
    
            # --- Option gain/loss target fields ---
            entry_price = kwargs.get("entry_price")
            if entry_price is not None:
                try:
                    record["entry_price"] = float(entry_price)
                except (TypeError, ValueError):
                    pass
    
            gain_target_pct = kwargs.get("gain_target_pct")
            if gain_target_pct is not None:
                try:
                    val = float(gain_target_pct)
                    if val > 1:
                        val = val / 100.0
                    record["gain_target_pct"] = val
                except (TypeError, ValueError):
                    pass
    
            stop_loss_pct = kwargs.get("stop_loss_pct")
            if stop_loss_pct is not None:
                try:
                    val = float(stop_loss_pct)
                    if val > 1:
                        val = val / 100.0
                    record["stop_loss_pct"] = val
                except (TypeError, ValueError):
                    pass
    
            valid_until = kwargs.get("valid_until")
            if valid_until is not None:
                if isinstance(valid_until, str):
                    try:
                        valid_until = datetime.strptime(valid_until, "%Y-%m-%d")
                    except ValueError:
                        valid_until = datetime.fromisoformat(valid_until)
                if isinstance(valid_until, date) and not isinstance(valid_until, datetime):
                    valid_until = datetime.combine(valid_until, datetime.max.time())
                record["valid_until"] = valid_until
    
            # Duplicate detection prior to insertion
            # Use tz-aware bounds matching the stored date (date_value is always tz-aware ET)
            lookup_start = datetime.combine(date_value.date(), datetime.min.time(), tzinfo=date_value.tzinfo)  # tz-ok
            lookup_end = lookup_start + timedelta(days=1)  # calendar-ok
            base_query = {
                "symbol": record["symbol"],
                "action": record["action"],
                "date": {"$gte": lookup_start, "$lt": lookup_end},
                "account_nick": record["account_nick"],
            }
            potential = rr.find(base_query, {"metadata": 1, "shares": 1, "dispatch": 1, "status": 1})
            for existing in potential:
                existing_status = str(existing.get("status") or "")
                if existing_status in ("cancelled", "canceled"):
                    continue  # don't match against cancelled plans
                existing_metadata = existing.get("metadata") or {}
                existing_leg = str(existing_metadata.get("leg") or "").strip().lower() or None
                existing_account = str(existing_metadata.get("account_id") or "").strip()
                existing_shares = int(existing.get("shares") or 0)
                # Scalp PDT plans have different dispatch tags (Plan A/B/D/E) for the
                # same symbol — don't dedup them. Only dedup if dispatch also matches.
                existing_dispatch = str(existing.get("dispatch") or "").strip()
                record_dispatch = str(record.get("dispatch") or "").strip()
                # Shares match: required for most plans (different shares = intentional chunks).
                # Exception: verdict buy_back / scalp_accumulate re-runs recalculate shares
                # each time (price changes). For these, same dispatch = same intent regardless
                # of shares. The dispatch tag already encodes symbol+account uniquely.
                shares_match = existing_shares == int(record.get("shares") or 0)
                # Recompute-each-run dispatches: every re-run recalculates shares/stop from the CURRENT
                # position or price, so SAME dispatch = SAME intent regardless of share count. Dedup by
                # dispatch and UPDATE the existing pending plan in place (operator 2026-06-23: eod-stop-
                # refresh + verdict-profit-protect were inserting 100s of duplicate plans/day, bloating
                # the executor queue and delaying real fills).
                _recompute_dispatch = record_dispatch.startswith((
                    "verdict-buy_back-", "verdict-scalp-acc-", "verdict-add-",
                    "eod-stop-refresh-", "verdict-profit-protect-", "profit-protect-stop-",
                ))
                if (
                    existing_leg == normalized_leg
                    and existing_account == normalized_account
                    and (shares_match or _recompute_dispatch)
                    and existing_dispatch == record_dispatch
                ):
                    if _recompute_dispatch and not shares_match:
                        # Refresh the existing pending plan with the latest shares/stop instead of a dup.
                        _upd = {"shares": int(record.get("shares") or 0), "updated_at": datetime.now()}
                        for _k in ("price", "limit_price", "price_limit", "price_operator",
                                   "level_operator", "level_limit", "notes"):
                            if _k in record:
                                _upd[_k] = record[_k]
                        rr.update_one({"_id": existing.get("_id")}, {"$set": _upd})
                    logger.info(
                        "[add_trading_plan] Duplicate {}/{} {} dispatch={} shares={}vs{}; {} plan {}",
                        record["symbol"], record["action"], lookup_start.date(), record_dispatch,
                        existing_shares, int(record.get("shares") or 0),
                        "UPDATED" if (_recompute_dispatch and not shares_match) else "kept",
                        existing.get("_id"),
                    )
                    return existing.get("_id")
    
            # Assert data integrity before insert
            assert record["symbol"], "Symbol cannot be empty"
            assert record["action"] in ["buy", "sell", "hold", "cover", "short", "snipe_call", "snipe_put", "reap_call", "reap_put", "roll_call", "roll_put", "option_open"], f"Invalid action: {record['action']}"
            assert isinstance(record["date"], (datetime, date)), "Date must be datetime"
            assert record["shares"] >= 0, "Shares must be non-negative"
            assert record["action_start"] >= 0, "Action start must be non-negative"
            assert record["action_end"] > record["action_start"], "Action end must be after action start"
    
            metadata_keys = sorted((record.get("metadata") or {}).keys())
            logger.info(
                "[add_trading_plan] Inserting plan | symbol={} action={} shares={} amount={:.2f} price={:.2f} start={} end={} metadata_keys={}",
                record["symbol"],
                record["action"],
                record["shares"],
                float(record.get("amount", 0.0)),
                float(record.get("price", 0.0)),
                record.get("action_start"),
                record.get("action_end"),
                metadata_keys,
            )
    
            try:
                result = rr.insert_one(record)
    
                # Assert insertion succeeded
                assert result.inserted_id, "Failed to insert trading plan"
    
                # Verify the record was actually inserted
                inserted = rr.find_one({"_id": result.inserted_id})
                assert inserted, f"Failed to verify inserted plan with ID {result.inserted_id}"
    
                logger.info(f"[add_trading_plan] Successfully inserted plan with ID: {result.inserted_id}")
    
                # Publish plan create event for real-time updates
                # Send the complete plan data for the event
                # Format date as YYYY-MM-DD for consistency with frontend
                if isinstance(record["date"], datetime):
                    date_str = record["date"].strftime("%Y-%m-%d")
                elif isinstance(record["date"], date):
                    date_str = record["date"].strftime("%Y-%m-%d")
                else:
                    date_str = str(record["date"])
    
                plan_data = {
                    "_id": str(result.inserted_id),
                    "symbol": record["symbol"],
                    "action": record["action"],
                    "date": date_str,
                    "shares": record["shares"],
                    "amount": record["amount"],
                    "price": record["price"],
                    "notes": record["notes"],
                    "status": record["status"],
                    "action_start": record["action_start"],
                    "action_end": record["action_end"],
                    "trigger": record.get("trigger", ""),
                    "price_operator": record.get("price_operator", ""),
                    "price_limit": record.get("price_limit", 0.0),
                    "level_operator": record.get("level_operator", ""),
                    "level_limit": record.get("level_limit", 0.0),
                    "flow_bias": record.get("flow_bias", ""),
                    "signal_bias": record.get("signal_bias", ""),
                    "tide_bias": record.get("tide_bias", ""),
                    "flow_window_minutes": record.get("flow_window_minutes"),
                    "flow_min_delta_pct": record.get("flow_min_delta_pct"),
                    "alpaca_mode": record.get("alpaca_mode", "live"),
                    "request_by": record.get("request_by"),
                    "requested_by": record.get("request_by"),
                }
                PlanEventPublisher.publish_create(str(result.inserted_id), plan_data)
    
                # Discord notification for plan creation
                _notify_trading_plan(
                    "CREATED", record["symbol"], record["action"],
                    record["shares"], account=record.get("account_nick", ""),
                    status=record["status"], dispatch=record.get("dispatch", ""),
                    notes=record.get("notes", ""), plan_date=date_str,
                )
    
                # Send notification via report.notify
                from rtrader.utils import report
                notification_msg = (
                    f"Trading plan to {record['action']} {record['shares']} {record['symbol']} "
                    f"scheduled for minutes {record['action_start']}-{record['action_end']} "
                    f"(requested by {record.get('request_by')})!"
                )
                report.notify(notification_msg, "Trading Plan", type="success")
    
                return result.inserted_id
    
            except Exception as e:
                import traceback
                logger.error(f"[add_trading_plan] Database operation failed: {str(e)}")
                logger.error(f"[add_trading_plan] Stack trace:\n{traceback.format_exc()}")
>               raise AssertionError(f"Database operation failed: {str(e)}")
E               AssertionError: Database operation failed: Failed to verify inserted plan with ID test-id

rtrader/database.py:5644: AssertionError