fix strategy state persistence corrections

tests/adapters/persistence/test_migration_and_drift.py

@@ -28,6 +28,7 @@ _EXPECTED_TABLES = {
     "broker_account_snapshot",  # dcy-041 / GitHub #45
     "ledger_skip",             # GitHub #68 skipped-ledger durable audit
     "reconcile_notification_state",  # req-0007 / ac-0024
+    "slot_strategy_state",  # req-0009 / ac-0027
 }
 
 
@@ -59,6 +60,25 @@ def test_daily_slot_columns_match_expected_contract(pg_clean):
     assert actual == set(EXPECTED_SLOT_FIELDS)
 
 
+def test_strategy_state_schema_matches_instance_map_contract(pg_clean):
+    with psycopg.connect(pg_clean) as conn:
+        with conn.cursor() as cur:
+            cur.execute(
+                "SELECT column_name, data_type, udt_name "
+                "FROM information_schema.columns "
+                "WHERE table_schema = 'public' AND table_name = 'slot_strategy_state'"
+            )
+            columns = {row[0]: (row[1], row[2]) for row in cur.fetchall()}
+            cur.execute(
+                "SELECT a.attname FROM pg_index i "
+                "JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey) "
+                "WHERE i.indrelid = 'slot_strategy_state'::regclass AND i.indisprimary"
+            )
+            primary_key = {row[0] for row in cur.fetchall()}
+    assert columns == {"slot_id": ("uuid", "uuid"), "data": ("jsonb", "jsonb")}
+    assert primary_key == {"slot_id"}
+
+
 def test_drift_detected_when_contract_extended(pg_clean):
     """Negative case: a contract field absent from the table is caught."""
     actual = _columns(pg_clean, "daily_slot")

tests/adapters/persistence/test_postgresql_repository.py

@@ -164,7 +164,7 @@ def test_strategy_state_store_pg(pg_clean):
 
     repo = PostgreSQLSlotRepository(pg_clean)
     store = PgStrategyStateStore(pg_clean)
-    svc = SlotService(repo, instance_key=INST, strategy_state_store=store)
+    svc = SlotService(repo, instance_key=INST)
     repo.reconcile_universe(INST, {"pf-a": "005930"})
     slot = repo.find_slot_by_portfolio(INST, "pf-a")
     svc.transit_slot(slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY,
@@ -189,12 +189,58 @@ def test_strategy_state_store_pg(pg_clean):
         ))
     assert all(write.ok for write in writes)
     assert {"instance-c", "instance-d"} <= set(store.get_for_slots([slot.slot_id])[slot.slot_id])
+    assert store.get_for_slots([slot.slot_id])[slot.slot_id]["instance-a"].version == 1
+
+    # Separate connections race on the same entry/version: only one CAS wins.
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        competing = list(executor.map(
+            lambda data: store.set(
+                slot.slot_id, "instance-a", expected_version=1, data=data, slot_active=True,
+            ),
+            ("winner-one", "winner-two"),
+        ))
+    assert sum(result.ok for result in competing) == 1, competing
+    assert sum(result.reason == ss.REASON_VERSION_CONFLICT for result in competing) == 1
+    winner = next(data for data, result in zip(("winner-one", "winner-two"), competing) if result.ok)
+    assert store.get_for_slots([slot.slot_id])[slot.slot_id]["instance-a"].data == winner
+    assert store.get_for_slots([slot.slot_id])[slot.slot_id]["instance-a"].version == 2
+
+    # A fresh adapter sees the byte-identical persisted entry and can continue CAS.
+    before_restart = store.get_for_slots([slot.slot_id])[slot.slot_id]["instance-a"]
+    after_restart = PgStrategyStateStore(pg_clean).get_for_slots([slot.slot_id])[slot.slot_id]["instance-a"]
+    assert after_restart == before_restart
 
     svc.transit_slot(slot.slot_id, SlotState.HOLDING, SlotState.PENDING_SELL,
                      reason="sell", actor="t")
     svc.transit_slot(slot.slot_id, SlotState.PENDING_SELL, SlotState.INACTIVE,
                      reason="sold", actor="t")
-    assert store.get_for_slots([slot.slot_id])[slot.slot_id]["instance-a"].version == 1
+    assert store.get_for_slots([slot.slot_id])[slot.slot_id]["instance-a"].version == 2
     reused = svc.transit_slot(slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY,
-                              reason="buy", actor="t")
+                               reason="buy", actor="t")
     assert reused.occupancy_seq == 2
+    rehydrated = PgStrategyStateStore(pg_clean)
+    persisted = rehydrated.get_for_slots([slot.slot_id])[slot.slot_id]["instance-a"]
+    assert persisted == before_restart
+    assert rehydrated.set(
+        slot.slot_id, "instance-a", expected_version=2, data="continued", slot_active=True,
+    ).version == 3
+
+
+def test_strategy_state_write_cannot_follow_close(pg_clean):
+    """The store's locked lifecycle read rejects a write after close wins."""
+    from trex_core.adapters.persistence.postgresql import PgStrategyStateStore
+
+    repo = PostgreSQLSlotRepository(pg_clean)
+    svc = SlotService(repo, instance_key=INST)
+    store = PgStrategyStateStore(pg_clean)
+    repo.reconcile_universe(INST, {"pf-close": "005930"})
+    slot = repo.find_slot_by_portfolio(INST, "pf-close")
+    svc.transit_slot(slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, reason="buy", actor="t")
+    svc.transit_slot(slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, reason="fill", actor="t", qty=1)
+    assert store.set(slot.slot_id, "instance", expected_version=0, data="before", slot_active=True).ok
+
+    svc.transit_slot(slot.slot_id, SlotState.HOLDING, SlotState.PENDING_SELL, reason="sell", actor="t")
+    svc.transit_slot(slot.slot_id, SlotState.PENDING_SELL, SlotState.INACTIVE, reason="sold", actor="t")
+    result = store.set(slot.slot_id, "instance", expected_version=1, data="after", slot_active=True)
+    assert not result.ok and result.reason == "slot_inactive"
+    assert store.get_for_slots([slot.slot_id])[slot.slot_id]["instance"].data == "before"

tests/application/test_slots_handler_quantity_mismatch.py

@@ -332,6 +332,33 @@ def test_ac7_warn_store_failure_fail_closed():
     assert "excluded_slots" in resp.payload
 
 
+def test_ac23_exclusion_precedes_broker_and_strategy_state_readers():
+    repo = InMemorySlotRepository()
+    svc = SlotService(repo)
+    warn_store = InMemoryReconcileWarningStore()
+    slot = _make_holding_slot(repo, svc, portfolio_id="pf-A", symbol=SYM_A, qty=10)
+    warn_store.set_warned(str(slot.slot_id), portfolio_id="pf-A", case_code="C")
+
+    class ExplodingReader:
+        def holdings_for_slots_rpc(self):
+            raise AssertionError("excluded slot reached broker reader")
+
+    class ExplodingStateStore:
+        def get_for_slots(self, _slot_ids):
+            raise AssertionError("excluded slot reached state reader")
+
+    handler = make_slots_rpc_handler(
+        repo,
+        instance_key=INSTANCE,
+        warn_store=warn_store,
+        broker_quantity_reader=ExplodingReader(),
+        strategy_state_store=ExplodingStateStore(),
+    )
+    response = handler(_req(filter="holding")).payload
+    assert response["slots"] == []
+    assert response["excluded_slots"] == {"reconcile_quantity_mismatch": 1}
+
+
 # ---------------------------------------------------------------------------
 # AC-8: intent rejected if C-case warning set after last slots RPC (race)
 # ---------------------------------------------------------------------------

tests/application/test_slots_handler_strategy_state.py

@@ -58,3 +58,14 @@ def test_operator_without_or_with_id_receives_full_map_without_watchdog_update()
         state = _call(handler, payload)["slots"][0]["strategy_state"]
         assert set(state) == {"instance-a", "instance-b"}
     assert watchdog.last_seen(str(slot.slot_id)) is None
+
+
+@pytest.mark.parametrize("spy", ["false", "true", 1, [], {}])
+def test_non_boolean_spy_is_rejected_without_visibility_or_watchdog_update(spy):
+    slot, _, watchdog, handler = _wire()
+    payload = _call(handler, {"spy": spy})
+    assert payload == {
+        "slots": [], "available_slots": None, "degraded": False,
+        "reason": "bad_request:spy", "message": "spy must be a boolean",
+    }
+    assert watchdog.last_seen(str(slot.slot_id)) is None

tests/conftest.py

@@ -27,6 +27,7 @@ _PG_TABLES = (
     "daily_symbol_entries",
     "daily_slot",
     "reconcile_notification_state",
+    "slot_strategy_state",
 )
 
 

tests/domain/slot/test_occupancy_seq_and_close_delete.py

@@ -8,7 +8,7 @@ from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository
 
 def test_close_preserves_state_and_reopen_continues_version():
     repo, store = InMemorySlotRepository(), InMemoryStrategyStateStore()
-    service = SlotService(repo, strategy_state_store=store)
+    service = SlotService(repo)
     repo.reconcile_universe("default", {"pf": "SYM"})
     slot = repo.list_slots("default")[0]
     service.transit_slot(slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, reason="buy", actor="t")

trex_core/adapters/persistence/postgresql.py

@@ -716,53 +716,53 @@ class PgStrategyStateStore:
         data: str,
         slot_active: bool,
     ) -> SetResult:
-        if not slot_active:
-            return SetResult(ok=False, reason="slot_inactive")
         if len(data.encode("utf-8")) > _STATE_CAP_BYTES:
             return SetResult(ok=False, reason="state_too_large")
         now = datetime.now(tz=timezone.utc)
         with self._psycopg.connect(self._dsn, autocommit=False) as conn:
             try:
                 with conn.cursor() as cur:
+                    # Lock the lifecycle row before changing state.  A close UPDATE
+                    # and this write therefore have one serial order: a writer that
+                    # follows a close observes INACTIVE and cannot recreate state.
                     cur.execute(
-                        "INSERT INTO slot_strategy_state (slot_id, data) "
+                        "SELECT status FROM daily_slot WHERE slot_id = %s::uuid FOR UPDATE",
-                        "SELECT %s, jsonb_build_object(%s, jsonb_build_object("
+                        (slot_id,),
-                        "'version', 1, 'data', %s, 'updated_at', %s)) WHERE %s = 0 "
-                        "ON CONFLICT (slot_id) DO UPDATE SET data = jsonb_set("
-                        "slot_strategy_state.data, ARRAY[%s], jsonb_build_object("
-                        "'version', %s + 1, 'data', %s, 'updated_at', %s), true) "
-                        "WHERE COALESCE((slot_strategy_state.data -> %s ->> 'version')::bigint, 0) = %s "
-                        "RETURNING data -> %s ->> 'version'",
-                        (
-                            slot_id,
-                            strategy_instance_id,
-                            data,
-                            now,
-                            expected_version,
-                            strategy_instance_id,
-                            expected_version,
-                            data,
-                            now,
-                            strategy_instance_id,
-                            expected_version,
-                            strategy_instance_id,
-                        ),
                     )
-                    row = cur.fetchone()
+                    slot = cur.fetchone()
-                    if row is None:
+                    if slot is None or slot[0] not in {"pending_buy", "holding", "pending_sell"}:
-                        cur.execute(
+                        return SetResult(ok=False, reason="slot_inactive")
-                            "SELECT COALESCE((data -> %s ->> 'version')::bigint, 0) "
+                    cur.execute(
-                            "FROM slot_strategy_state WHERE slot_id = %s",
+                        "SELECT data FROM slot_strategy_state WHERE slot_id = %s::uuid",
-                            (strategy_instance_id, slot_id),
+                        (slot_id,),
-                        )
+                    )
-                        conflict = cur.fetchone()
+                    state_row = cur.fetchone()
+                    state_map = state_row[0] if state_row is not None else {}
+                    entry = state_map.get(strategy_instance_id)
+                    version = int(entry["version"]) if entry is not None else 0
+                    if expected_version != version:
                         return SetResult(
                             ok=False,
                             reason="version_conflict",
-                            version=int(conflict[0]) if conflict is not None else 0,
+                            version=version,
+                        )
+                    entry_json = "jsonb_build_object('version', %s::bigint, " \
+                        "'data', %s::text, 'updated_at', %s::timestamptz)"
+                    if state_row is None:
+                        cur.execute(
+                            "INSERT INTO slot_strategy_state (slot_id, data) VALUES ("
+                            "%s::uuid, jsonb_build_object(%s::text, " + entry_json + "))",
+                            (slot_id, strategy_instance_id, version + 1, data, now),
+                        )
+                    else:
+                        cur.execute(
+                            "UPDATE slot_strategy_state SET data = jsonb_set(data, "
+                            "ARRAY[%s::text]::text[], " + entry_json + ", true) "
+                            "WHERE slot_id = %s::uuid",
+                            (strategy_instance_id, version + 1, data, now, slot_id),
                         )
                 conn.commit()
-                return SetResult(ok=True, version=int(row[0]))
+                return SetResult(ok=True, version=version + 1)
             except BaseException:
                 conn.rollback()
                 raise

trex_core/adapters/persistence/sql/027_replace_slot_strategy_state_with_instance_map.sql

@@ -12,7 +12,7 @@ BEGIN
         DROP TABLE slot_strategy_state;
     END IF;
 END $$;
-CREATE TABLE slot_strategy_state (
+CREATE TABLE IF NOT EXISTS slot_strategy_state (
     slot_id UUID PRIMARY KEY,
     data JSONB NOT NULL DEFAULT '{}'::jsonb
 );

trex_core/application/messaging_handlers.py

@@ -223,7 +223,16 @@ def make_slots_rpc_handler(
     def handler(req: RpcRequest) -> RpcResponse:
         body = req.payload or {}
         filter_val = str(body.get("filter") or "all").lower()
-        spy = bool(body.get("spy", False))
+        spy_value = body.get("spy", False)
+        if not isinstance(spy_value, bool):
+            return RpcResponse(payload={
+                "slots": [],
+                "available_slots": None,
+                "degraded": False,
+                "reason": "bad_request:spy",
+                "message": "spy must be a boolean",
+            })
+        spy = spy_value
         strategy_instance_id = body.get("strategy_instance_id")
         if not spy and (not isinstance(strategy_instance_id, str) or not strategy_instance_id):
             return RpcResponse(payload={
@@ -268,10 +277,42 @@ def make_slots_rpc_handler(
             for s in slots
         ]
 
+        # AC-0023 exclusion must precede all optional data readers so excluded
+        # slots cannot expose state or make an otherwise valid response fail.
+        excluded_count = 0
+        if warn_store is not None:
+            try:
+                c_warned_ids = warn_store.list_c_case_warned()
+            except Exception:  # noqa: BLE001
+                _log.exception("warn_store.list_c_case_warned() failed — fail-closed")
+                return RpcResponse(payload={
+                    "slots": [],
+                    "available_slots": None,
+                    "degraded": True,
+                    "reason": "warn_store_unavailable",
+                    "message": (
+                        "reconcile warning store unavailable; "
+                        "returning no slots (fail-closed)"
+                    ),
+                    "excluded_slots": {"reconcile_quantity_mismatch": 0},
+                })
+            if c_warned_ids:
+                included = [
+                    (slot, row)
+                    for slot, row in zip(slots, slot_rows, strict=True)
+                    if not (
+                        slot.status.value == "holding"
+                        and str(slot.slot_id) in c_warned_ids
+                    )
+                ]
+                excluded_count = len(slots) - len(included)
+                slots = [slot for slot, _ in included]
+                slot_rows = [row for _, row in included]
+
         broker_qty_by_symbol: dict[str, int] = {}
         broker_qty_available = False
         broker_qty_captured_at: Optional[datetime] = None
-        if broker_quantity_reader is not None:
+        if broker_quantity_reader is not None and slots:
             try:
                 holdings, broker_qty_captured_at = broker_quantity_reader.holdings_for_slots_rpc()
                 broker_qty_by_symbol = {
@@ -361,43 +402,6 @@ def make_slots_rpc_handler(
         if strategy_filter is not None:
             slot_rows = [r for r in slot_rows if r.get("strategy_name") == strategy_filter]
 
-        # AC-0023: exclude holding slots with active case_code='C' reconcile warning.
-        # Fail-closed: if warn_store exists but throws, return degraded=True / no slots.
-        excluded_count = 0
-        if warn_store is not None:
-            try:
-                c_warned_ids = warn_store.list_c_case_warned()
-            except Exception:  # noqa: BLE001
-                _log.exception("warn_store.list_c_case_warned() failed — fail-closed")
-                return RpcResponse(payload={
-                    "slots": [],
-                    "available_slots": available_slots,
-                    "degraded": True,
-                    "reason": "warn_store_unavailable",
-                    "message": (
-                        "reconcile warning store unavailable; "
-                        "returning no slots (fail-closed)"
-                    ),
-                    "excluded_slots": {"reconcile_quantity_mismatch": 0},
-                })
-            if c_warned_ids:
-                # Rebuild slot lookup from the original `slots` list (before strategy
-                # filter reduced slot_rows). slot_rows[i] corresponds to slots[i]
-                # only before filtering; use slot_id stored in each row.
-                slot_by_id = {str(s.slot_id): s for s in slots}
-                filtered_rows: list[dict] = []
-                for row in slot_rows:
-                    slot_obj = slot_by_id.get(str(row.get("slot_id", "")))
-                    if (
-                        slot_obj is not None
-                        and str(slot_obj.slot_id) in c_warned_ids
-                        and slot_obj.status.value == "holding"
-                    ):
-                        excluded_count += 1
-                    else:
-                        filtered_rows.append(row)
-                slot_rows = filtered_rows
-
         base_payload: dict = {
             "slots": slot_rows,
             "available_slots": available_slots,

trex_core/domain/slot/service.py

@@ -44,15 +44,12 @@ class SlotService:
         self,
         repository: SlotRepository,
         instance_key: str = "default",
-        *,
-        strategy_state_store: Any = None,
     ) -> None:
         self._repo = repository
         # dcy-020/REQ-057: advisory lock is scoped per instance so a shared DB
         # does not serialize transitions across instances (NFR-1).
         self._instance_key = instance_key
         self._lock_id = slot_advisory_lock_id(instance_key)
-        self._strategy_state_store = strategy_state_store
 
     @property
     def instance_key(self) -> str:

trex_core/runtime/daemon.py

@@ -182,13 +182,12 @@ class Daemon:
         # Persistence + services
         self.slot_repo = PostgreSQLSlotRepository(config.dsn)
         self.intent_store = PostgreSQLIntentStore(config.dsn)
-        # dcy-028 / REQ-079·081 (BAT-042): opaque strategy-state store. Injected
+        # dcy-028 / REQ-079·081 (BAT-042): opaque strategy-state store persists
-        # into the slot service so close (→INACTIVE) deletes the slot's state.
+        # independently of slot occupancy and is shared by both RPC handlers.
         self.strategy_state_store = PgStrategyStateStore(config.dsn)
         self.slot_service = SlotService(
             self.slot_repo,
             instance_key=config.instance_key,
-            strategy_state_store=self.strategy_state_store,
         )
 
         self.outbox = PostgreSQLLedgerOutbox(config.dsn)
@@ -458,7 +457,7 @@ class Daemon:
             topology.slots_queue(self.config.instance_key), handler, self.stop)
 
     def _strategy_state_set_loop(self) -> None:
-        # dcy-028 / REQ-080: opaque strategy-state write RPC (occupancy_seq+version CAS).
+        # dcy-028 / REQ-080: opaque strategy-state write RPC (entry-version CAS).
         transport = _build_transport(self.config)
         self._transports.append(transport)
         try: