Fix CPR AI true-up coordination and MCP retries

Nifty Multi Strategy Front Test - Master File.py

@@ -334,17 +334,30 @@
 # DeprecationWarning onto the operator's console for every feed connection.
 # The SDK version is policy-pinned (DEPS-001 in requirements.txt), so until a
 # deliberate bump moves past that call we silence EXACTLY that message from
-# EXACTLY that module: deprecation warnings raised by our own code, or by any
+# EXACTLY those modules: deprecation warnings raised by our own code, or by any
 # other library, still reach the console.  The warning fires per tick at
 # runtime (never at import), so installing the filter here -- at module load,
-# long before any feed thread starts -- covers every code path; and because
-# `filterwarnings` PREPENDS, it also wins over any blanket -W /
-# PYTHONWARNINGS setting on the host.
+# long before any feed thread starts -- covers every runtime code path.
+#
+# `dhanhq/__init__.py` imports BOTH `marketfeed` and `fulldepth`, and each ships
+# the same `utc_time` helper (marketfeed.py:523, fulldepth.py:391).  The runner
+# only subscribes MarketFeed, so the fulldepth call site should never fire; it is
+# covered anyway because the cost is one regex branch and the alternative is a
+# surprise on the day something reaches for full-depth data.
+#
+# This filter does NOT cover pytest.  Pytest wraps every test in
+# `catch_warnings()` + `simplefilter("always")`, which RESETS `warnings.filters`
+# and discards anything a module installed at import time; only pytest's own
+# `-W` / `[tool.pytest.ini_options] filterwarnings` entries are re-applied
+# inside that context.  The matching ini entries live in `pyproject.toml` and
+# `Tests/Dependencies/test_repository_policy.py` asserts the two stay in step --
+# without them this exact warning reappears in every pytest run even though the
+# runner itself is silent.
 warnings.filterwarnings(
     "ignore",
     message=r"datetime\.datetime\.utcfromtimestamp\(\) is deprecated",
     category=DeprecationWarning,
-    module=r"dhanhq\.marketfeed",
+    module=r"dhanhq\.(marketfeed|fulldepth)",
 )
 
 
@@ -1891,6 +1904,8 @@ class MarketSnapshot:
     - `source_candle_ts` : timestamp of the latest candle in `frame`.
     - `candle_signature` : lightweight fingerprint of the latest row's state.
     - `fetched_at`       : wall-clock time when the fetch completed.
+    - `official_candle_ts`: newest row supplied by the REST/official source, or
+      ``None`` when the publisher has not proved any official coverage.
 
     Why we keep both `source_candle_ts` and `candle_signature`:
     - During a live 1-minute candle, the timestamp does not change but the
@@ -1906,6 +1921,10 @@ class MarketSnapshot:
     source_candle_ts: pd.Timestamp | None
     candle_signature: tuple | None
     fetched_at: datetime
+    # Websocket frames can contain a mix of official REST history and newer
+    # tick-built rows.  This watermark tells a five-minute consumer exactly how
+    # far the official portion reaches without changing the canonical OHLC data.
+    official_candle_ts: pd.Timestamp | None = None
 
 
 PRICE_QUALITY_BROKER_FILL = "BROKER_FILL"
@@ -2203,12 +2222,22 @@ def __init__(self) -> None:
     # ------------------------------------------------------------------
     # OHLC pool
     # ------------------------------------------------------------------
-    def update(self, timeframe: str, frame: pd.DataFrame) -> MarketSnapshot:
+    def update(
+        self,
+        timeframe: str,
+        frame: pd.DataFrame,
+        *,
+        official_candle_ts: pd.Timestamp | datetime | None = None,
+    ) -> MarketSnapshot:
         """
         Atomically replace the stored snapshot for `timeframe`.
 
         The new snapshot is built BEFORE the lock is taken; only the swap
         happens under lock so the critical section stays small.
+
+        ``official_candle_ts`` is optional because a pure tick publisher has no
+        official coverage to claim.  REST-backed publishers pass their newest
+        source timestamp so conservative consumers can wait for official data.
         """
         validated = validate_ohlc_frame(frame)
         source_candle_ts = pd.to_datetime(validated.iloc[-1]["timestamp"])
@@ -2220,6 +2249,11 @@ def update(self, timeframe: str, frame: pd.DataFrame) -> MarketSnapshot:
             source_candle_ts=source_candle_ts,
             candle_signature=candle_signature,
             fetched_at=datetime.now(ZoneInfo("Asia/Kolkata")),
+            official_candle_ts=(
+                None
+                if official_candle_ts is None
+                else pd.Timestamp(official_candle_ts)
+            ),
         )
         with self._lock:
             self._snapshots[str(timeframe)] = snapshot
@@ -2243,6 +2277,7 @@ def get(self, timeframe: str) -> MarketSnapshot | None:
             source_candle_ts=snapshot.source_candle_ts,
             candle_signature=snapshot.candle_signature,
             fetched_at=snapshot.fetched_at,
+            official_candle_ts=snapshot.official_candle_ts,
         )
 
     # ------------------------------------------------------------------
@@ -3783,7 +3818,14 @@ def run(self) -> None:
                     break
                 try:
                     frame = self.fetch_ohlc(timeframe)
-                    snapshot = self.store.update(timeframe, frame)
+                    # This producer's entire frame came directly from REST, so
+                    # its newest timestamp is also the official-data watermark.
+                    # Websocket mode publishes the same metadata after true-up.
+                    snapshot = self.store.update(
+                        timeframe,
+                        frame,
+                        official_candle_ts=pd.Timestamp(frame["timestamp"].max()),
+                    )
                     if self.last_logged_candle_ts.get(timeframe) != snapshot.source_candle_ts:
                         self.last_logged_candle_ts[timeframe] = snapshot.source_candle_ts
                         self.log.info(
@@ -3879,6 +3921,10 @@ def __init__(
         # Latest REST history: warmup seed, then refreshed by every true-up.
         # Supervisor-owned; the pump never touches it.
         self.official_frame: pd.DataFrame = pd.DataFrame()
+        # The frame may later be merged with newer tick-built rows.  Keep the
+        # newest timestamp that came from REST as a separate watermark so a
+        # strategy can prove its completed bucket has been officially trued up.
+        self._official_candle_ts: pd.Timestamp | None = None
 
         # Connection state shared between pump and supervisor.
         self._feed_lock = threading.Lock()
@@ -3974,8 +4020,13 @@ def _warmup_official_history(self) -> bool:
                     exchange_segment=NIFTY_INDEX_EXCHANGE_SEGMENT,
                     instrument_type=NIFTY_INDEX_INSTRUMENT_TYPE,
                 )
-                self.store.update("1", frame)
                 self.official_frame = frame
+                self._official_candle_ts = pd.Timestamp(frame["timestamp"].max())
+                self.store.update(
+                    "1",
+                    frame,
+                    official_candle_ts=self._official_candle_ts,
+                )
                 self.log.info("Warmup history loaded | Rows=%s", len(frame))
                 return True
             except Exception as exc:
@@ -4102,7 +4153,11 @@ def _publish_frame_locked(self, force: bool) -> None:
         if frame.empty:
             return
         try:
-            snapshot = self.store.update("1", frame)
+            snapshot = self.store.update(
+                "1",
+                frame,
+                official_candle_ts=self._official_candle_ts,
+            )
             self._ohlc_ok = True
             if self.last_logged_candle_ts != snapshot.source_candle_ts:
                 self.last_logged_candle_ts = snapshot.source_candle_ts
@@ -4178,13 +4233,19 @@ def _run_true_up(self, reason: str, now_ist: datetime | None = None) -> None:
             self.aggregator.tick_bars_frame(),
             forming_minute=pd.Timestamp(now_ist).floor("min"),
         )
-        self.official_frame = official
-        self._publish_frame_if_changed(force=True)
+        newest_official = pd.Timestamp(official["timestamp"].max())
+        # Update the official frame, its watermark, and the published merged
+        # snapshot under one lock.  Without this atomic boundary, the supervisor
+        # could publish new official OHLC with an older watermark (or vice versa)
+        # and make a waiting CPR worker observe a mixed generation.
+        with self._publish_lock:
+            self.official_frame = official
+            self._official_candle_ts = newest_official
+            self._publish_frame_locked(force=True)
         # Drop every tick bar the official history now covers. The merge would
         # ignore them anyway (official wins), but keeping them makes the NEXT
         # divergence report re-count this cycle's mismatches forever -- the
         # stats above must describe only the minutes trued-up right now.
-        newest_official = pd.Timestamp(official["timestamp"].max())
         self.aggregator.prune_older_than(newest_official + pd.Timedelta(minutes=1))
         log_fn = (
             self.log.warning
@@ -9519,6 +9580,7 @@ def __init__(
         self._latest_one_minute_frame = pd.DataFrame()
         self._latest_context_as_of: datetime | None = None
         self._last_agent_bar_identity: str | None = None
+        self._waiting_for_official_bar_identity: str | None = None
         self._prior_accepted_regime: str | None = None
         self._cpr_state: CPRAITradeState | None = None
 
@@ -9590,6 +9652,48 @@ def _completed_bar_identity(self, strategy_frame: pd.DataFrame) -> str:
             timestamp = timestamp.tz_convert(IST_TIMEZONE).tz_localize(None)
         return f"{timestamp.date()}|{timestamp.isoformat()}"
 
+    @staticmethod
+    def _naive_ist_timestamp(value: object) -> pd.Timestamp | None:
+        """Normalize a timestamp before comparing producer and strategy clocks.
+
+        Dhan history normally uses naive IST timestamps, while tests or future
+        adapters may provide timezone-aware values.  Converting both shapes to
+        naive IST prevents a harmless representation difference from either
+        raising or incorrectly bypassing the official-data gate.
+        """
+
+        timestamp = pd.Timestamp(value)
+        if pd.isna(timestamp):
+            return None
+        if timestamp.tzinfo is not None:
+            timestamp = timestamp.tz_convert(IST_TIMEZONE).tz_localize(None)
+        return timestamp
+
+    def _official_snapshot_covers_completed_bar(
+        self,
+        snapshot: MarketSnapshot,
+        strategy_frame: pd.DataFrame,
+    ) -> bool:
+        """Return true only when REST covers the bucket's final source minute.
+
+        Five-minute candles are start-stamped.  A 09:55 candle therefore needs
+        the official one-minute source to cover 09:59 before Codex may freeze
+        it.  This condition-based gate naturally tolerates a slow Dhan response:
+        the normal worker poll simply checks the next atomic snapshot instead
+        of guessing how many seconds the REST true-up will take.
+        """
+
+        if strategy_frame.empty or "timestamp" not in strategy_frame.columns:
+            return False
+        official_timestamp = self._naive_ist_timestamp(snapshot.official_candle_ts)
+        bucket_start = self._naive_ist_timestamp(
+            strategy_frame.iloc[-1]["timestamp"]
+        )
+        if official_timestamp is None or bucket_start is None:
+            return False
+        bucket_final_minute = bucket_start + pd.Timedelta(minutes=4)
+        return official_timestamp >= bucket_final_minute
+
     def _position_state_payload(self) -> dict[str, object]:
         """Expose allowlisted premise/risk facts, never execution capabilities.
 
@@ -10527,8 +10631,10 @@ def run(self) -> None:
 
         The worker starts decisions at 09:30 but safety runs even before then.
         Shared one-minute data is resampled with the current IST clock so forming
-        websocket minutes are excluded. Any single poll failure is logged and
-        retried; it does not terminate future hard-stop or square-off handling.
+        websocket minutes are excluded. A clock-complete bucket then waits until
+        the atomic REST watermark covers its fifth source minute. Any single poll
+        failure is logged and retried; it does not terminate future hard-stop or
+        square-off handling.
         """
 
         self.log.info("Starting %s strategy worker.", self.strategy_name)
@@ -10555,6 +10661,21 @@ def run(self) -> None:
                 if completed.empty:
                     self.wait_for_next_poll()
                     continue
+                if not self._official_snapshot_covers_completed_bar(
+                    snapshot,
+                    completed,
+                ):
+                    waiting_identity = self._completed_bar_identity(completed)
+                    if waiting_identity != self._waiting_for_official_bar_identity:
+                        self._waiting_for_official_bar_identity = waiting_identity
+                        self.log.info(
+                            "Completed CPR bucket %s is waiting for its final "
+                            "one-minute REST true-up.",
+                            waiting_identity,
+                        )
+                    self.wait_for_next_poll()
+                    continue
+                self._waiting_for_official_bar_identity = None
                 self.process_strategy_frame(completed)
             except Exception as exc:  # noqa: BLE001 - one turn must not kill safety
                 self.log.exception("CPR AI worker poll failed: %s", exc)

Signal Generators/CPR AI Agent/README.md

@@ -28,8 +28,12 @@ deep-copy views of the same frozen completed-bar context:
   `HOLD`, premise `EXIT`, or the one permitted `SCALE_IN` request.
 
 The snapshot contains no order surface, account, credential, broker, venue, or
-execution object. A missing or failed tool call, a duplicate/unapproved tool,
-or any unexpected agent action invalidates the turn.
+execution object. The immediate turn request repeats the four exact tool names
+in addition to the developer prompt. If the first isolated turn still omits a
+tool, or one tool reports failure, the host permits one retry against the same
+immutable snapshot using only the time left in the original SDK deadline. A
+second incomplete result, a duplicate/unapproved tool, or any unexpected agent
+action invalidates the turn and produces `HOLD`.
 
 ## Decision contract and host gates
 
@@ -74,8 +78,14 @@ host then enforces the selected framework:
 - It polls mechanical safety every five seconds and calls Codex at most once for
   each newly completed five-minute candle. A start-stamped one-minute candle is
   not complete until the next minute begins, and all five exact minute slots
-  must exist once; websocket revisions and later REST true-ups cannot create a
-  second call for the same bucket.
+  must exist once.
+- In websocket mode, clock completeness alone is not enough. Every shared OHLC
+  snapshot carries an atomic `official_candle_ts` watermark. A 09:55 five-minute
+  bucket waits until the REST source covers its final 09:59 one-minute candle,
+  even when Dhan's true-up takes longer than its normal five-second delay. This
+  is a condition check rather than a hard-coded sleep. A later official revision
+  can still invalidate an in-flight result, but it cannot create a second model
+  call for an already-consumed bucket.
 - At 15:00 IST new entries and adds stop; management and exits continue.
 - At 15:15 IST the host square-off closes exposure and stops the worker.
 
@@ -136,7 +146,8 @@ missing row with a warning; the labels remain separate from legacy CPR workers.
 With `CPR_AI_DECISION_LOGGING_ENABLED=true`, the host appends sanitized JSONL to
 `Backtest Outputs/cpr_ai_decisions.jsonl` by default. Each row records the frozen
 context, proposal, accepted regime, validation code/reason, authoritative host
-geometry, execution outcome, latency, token usage, and tool-call evidence.
+geometry, execution outcome, latency, inference-attempt count, aggregate token
+usage across a retry, and final tool-call evidence.
 Credential-like mapping fields are removed recursively before serialization.
 Logging never makes a proposal executable, and an enabled log must succeed
 before an entry or add may be submitted.