Restore stale text request invariant

src/main/scala/net/eagle0/eagle/client_text/ClientTextStore.scala

@@ -71,6 +71,12 @@ trait ClientTextStore {
 
   def withMovedBackToUnrequested(id: ClientTextId): ClientTextStore
 
+  /**
+   * Removes generated text rows created after a rewind target. Anything past the retained action count belongs to
+   * history that no longer exists, and deterministic request IDs may be reused if the game re-advances.
+   */
+  def withoutTextsAfter(maxHistoryCount: Int): ClientTextStore
+
   /** Adds a pre-existing complete text directly (for pregenerated texts like hero names). */
   def withAddedCompleteText(
       id: ClientTextId,

src/main/scala/net/eagle0/eagle/client_text/ClientTextStoreImpl.scala

@@ -219,6 +219,26 @@ case class ClientTextStoreImpl(
         this
       }
 
+  override def withoutTextsAfter(maxHistoryCount: Int): ClientTextStore = {
+    val removedIds =
+      completeTexts.values.filter(_.requestedAfterHistoryCount > maxHistoryCount).map(_.id).toSet ++
+        incompleteTexts.values.filter(_.requestedAfterHistoryCount > maxHistoryCount).map(_.id).toSet ++
+        unrequestedTexts.values.filter(_.requestedAfterHistoryCount > maxHistoryCount).map(_.id).toSet
+
+    if removedIds.isEmpty then this
+    else
+      copy(
+        completeTexts = completeTexts -- removedIds,
+        incompleteTexts = incompleteTexts -- removedIds,
+        unrequestedTexts = unrequestedTexts -- removedIds,
+        accessibleTo = accessibleTo -- removedIds,
+        hiddenFromClients = hiddenFromClients -- removedIds,
+        incompleteTextsAreSaved = false,
+        accessibleToIsSaved = false,
+        hiddenFromClientsIsSaved = false
+      )
+  }
+
   def withAddedCompleteText(
       id: ClientTextId,
       text: String,

src/main/scala/net/eagle0/eagle/client_text/PostgresClientTextStore.scala

@@ -426,6 +426,80 @@ class PostgresClientTextStore private[client_text] (
     this
   }
 
+  override def withoutTextsAfter(maxHistoryCount: Int): ClientTextStore = {
+    val selectSql  =
+      """SELECT id FROM client_texts
+        |WHERE requested_after_history_count > ?""".stripMargin
+    val selectStmt = connection.prepareStatement(selectSql)
+    val ids        = JfrEvents.postgresQuery(
+      databaseName,
+      "PostgresClientTextStore.withoutTextsAfter.selectIds",
+      selectSql
+    ) { event =>
+      try {
+        selectStmt.setInt(1, maxHistoryCount)
+        val rs     = selectStmt.executeQuery()
+        val result = scala.collection.mutable.ArrayBuffer[ClientTextId]()
+        while rs.next() do result += rs.getString("id")
+        event.rows = result.size
+        result.toVector
+      } finally selectStmt.close()
+    }
+
+    if ids.isEmpty then return this
+
+    println(
+      s"Removing ${ids.size} generated text row(s) after rewound history count $maxHistoryCount: ${ids.mkString(", ")}"
+    )
+
+    var idsArray: java.sql.Array = null
+    try {
+      idsArray = connection.createArrayOf("text", ids.toArray)
+
+      val deleteVisibilitySql  = "DELETE FROM client_text_visibility WHERE text_id = ANY(?)"
+      val deleteVisibilityStmt = connection.prepareStatement(deleteVisibilitySql)
+      JfrEvents.postgresQuery(
+        databaseName,
+        "PostgresClientTextStore.withoutTextsAfter.deleteVisibility",
+        deleteVisibilitySql
+      ) { event =>
+        try {
+          deleteVisibilityStmt.setArray(1, idsArray)
+          event.rows = deleteVisibilityStmt.executeUpdate()
+        } finally deleteVisibilityStmt.close()
+      }
+
+      val deleteHiddenSql  = "DELETE FROM client_text_hidden WHERE text_id = ANY(?)"
+      val deleteHiddenStmt = connection.prepareStatement(deleteHiddenSql)
+      JfrEvents.postgresQuery(
+        databaseName,
+        "PostgresClientTextStore.withoutTextsAfter.deleteHidden",
+        deleteHiddenSql
+      ) { event =>
+        try {
+          deleteHiddenStmt.setArray(1, idsArray)
+          event.rows = deleteHiddenStmt.executeUpdate()
+        } finally deleteHiddenStmt.close()
+      }
+
+      val deleteTextsSql  = "DELETE FROM client_texts WHERE id = ANY(?)"
+      val deleteTextsStmt = connection.prepareStatement(deleteTextsSql)
+      JfrEvents.postgresQuery(
+        databaseName,
+        "PostgresClientTextStore.withoutTextsAfter.deleteTexts",
+        deleteTextsSql
+      ) { event =>
+        try {
+          deleteTextsStmt.setArray(1, idsArray)
+          event.rows = deleteTextsStmt.executeUpdate()
+        } finally deleteTextsStmt.close()
+      }
+    } finally if idsArray != null then idsArray.free()
+    end try
+
+    this
+  }
+
   override def withAddedCompleteText(
       id: ClientTextId,
       text: String,

src/main/scala/net/eagle0/eagle/client_text/SqliteClientTextStore.scala

@@ -506,6 +506,49 @@ class SqliteClientTextStore private[client_text] (
     this
   }
 
+  override def withoutTextsAfter(maxHistoryCount: Int): ClientTextStore = {
+    val idsStmt = connection.prepareStatement(
+      """SELECT id FROM texts
+        |WHERE requested_after_history_count > ?""".stripMargin
+    )
+    val ids     =
+      try {
+        idsStmt.setInt(1, maxHistoryCount)
+        val rs     = idsStmt.executeQuery()
+        val result = scala.collection.mutable.ArrayBuffer[ClientTextId]()
+        while rs.next() do result += rs.getString("id")
+        result.toVector
+      } finally idsStmt.close()
+
+    if ids.isEmpty then return this
+
+    println(
+      s"Removing ${ids.size} generated text row(s) after rewound history count $maxHistoryCount: ${ids.mkString(", ")}"
+    )
+
+    ids.foreach { id =>
+      val visStmt = connection.prepareStatement("DELETE FROM visibility WHERE text_id = ?")
+      try {
+        visStmt.setString(1, id)
+        visStmt.executeUpdate()
+      } finally visStmt.close()
+
+      val hiddenStmt = connection.prepareStatement("DELETE FROM hidden_texts WHERE text_id = ?")
+      try {
+        hiddenStmt.setString(1, id)
+        hiddenStmt.executeUpdate()
+      } finally hiddenStmt.close()
+
+      val textStmt = connection.prepareStatement("DELETE FROM texts WHERE id = ?")
+      try {
+        textStmt.setString(1, id)
+        textStmt.executeUpdate()
+      } finally textStmt.close()
+    }
+
+    this
+  }
+
   override def withAddedCompleteText(
       id: ClientTextId,
       text: String,

src/main/scala/net/eagle0/eagle/service/UnrequestedTextHandler.scala

@@ -26,6 +26,18 @@ class UnrequestedTextHandler(llmResolver: LlmResolver, heroNameCache: HeroNameCa
   // 3 minutes in milliseconds
   private val StalledThresholdMillis: Long = 3 * 60 * 1000
 
+  private def requireTextRequestStillValid(
+      id: String,
+      requestedAfterHistoryCount: Int,
+      currentHistoryCount: Int
+  ): Unit =
+    if requestedAfterHistoryCount > currentHistoryCount then
+      throw new EagleInternalException(
+        s"Stale pending text request $id: requestedAfterHistoryCount=$requestedAfterHistoryCount " +
+          s"is after current history count $currentHistoryCount. This usually means the game was rewound " +
+          s"without pruning its text store."
+      )
+
   private def withTransaction[T](store: ClientTextStore)(f: => T): T = {
     store.beginTransaction()
     try {
@@ -56,8 +68,10 @@ class UnrequestedTextHandler(llmResolver: LlmResolver, heroNameCache: HeroNameCa
         stuckEntries: Vector[StuckUnrequested]
     )
 
+    val currentHistoryCount                     = gameHistory.count
     val (generationRequests, fixedNameRequests) =
       clientTextStore.unrequestedTexts.values.partition { ut =>
+        requireTextRequestStillValid(ut.id, ut.requestedAfterHistoryCount, currentHistoryCount)
         ut.llmRequest match {
           case _: FixedHeroName => false
           case _                => true
@@ -262,28 +276,31 @@ class UnrequestedTextHandler(llmResolver: LlmResolver, heroNameCache: HeroNameCa
     }
 
     // Load game states, using cached versions when available
+    val currentHistoryCount                = gameHistory.count
     val (llmRequests, ctsWithCachedStates) =
-      clientTextStore.incompleteTexts.values.foldLeft(
-        (Vector.empty[LlmResolver.LlmRequestWithGameState], clientTextStore)
-      ) {
-        case ((requests, cts), ict) =>
-          val (gameState, updatedCts) = ict.cachedGameState match {
-            case Some(cached) =>
-              (cached, cts)
-            case None         =>
-              val loaded       = gameHistory.stateAfter(ict.requestedAfterHistoryCount)
-              val updatedEntry = ict.withCachedGameState(loaded)
-              val newCts       = cts.withCachedIncompleteGameState(ict.id, updatedEntry)
-              (loaded, newCts)
-          }
+      clientTextStore.incompleteTexts.values
+        .foldLeft(
+          (Vector.empty[LlmResolver.LlmRequestWithGameState], clientTextStore)
+        ) {
+          case ((requests, cts), ict) =>
+            requireTextRequestStillValid(ict.id, ict.requestedAfterHistoryCount, currentHistoryCount)
+            val (gameState, updatedCts) = ict.cachedGameState match {
+              case Some(cached) =>
+                (cached, cts)
+              case None         =>
+                val loaded       = gameHistory.stateAfter(ict.requestedAfterHistoryCount)
+                val updatedEntry = ict.withCachedGameState(loaded)
+                val newCts       = cts.withCachedIncompleteGameState(ict.id, updatedEntry)
+                (loaded, newCts)
+            }
 
-          val request = LlmResolver.LlmRequestWithGameState(
-            llmRequest = ict.llmRequest,
-            gameState = gameState,
-            partialCompletion = Some(ict.partialText)
-          )
-          (requests :+ request, updatedCts)
-      }
+            val request = LlmResolver.LlmRequestWithGameState(
+              llmRequest = ict.llmRequest,
+              gameState = gameState,
+              partialCompletion = Some(ict.partialText)
+            )
+            (requests :+ request, updatedCts)
+        }
 
     val results = llmResolver
       .resolveLlmRequests(

src/main/scala/net/eagle0/eagle/service/controller/GameController.scala

@@ -1249,7 +1249,8 @@ final case class GameController(
     val resyncedCount = humanClients.length
 
     // Rewind the engine
-    val rewoundEngine = engine.rewindTo(targetActionCount)
+    val rewoundEngine          = engine.rewindTo(targetActionCount)
+    val rewoundClientTextStore = clientTextStore.withoutTextsAfter(targetActionCount)
 
     // Resync all connected human clients with the new state
     val resyncedClients = humanClients.map { client =>
@@ -1259,7 +1260,7 @@ final case class GameController(
         engine = rewoundEngine,
         factionId = client.factionId,
         otherFactionIds = rewoundEngine.factionIds.filterNot(_ == client.factionId),
-        clientTextStore = clientTextStore
+        clientTextStore = rewoundClientTextStore
       )
       client.resyncAfterRewind(
         history = rewoundEngine.history,
@@ -1281,7 +1282,7 @@ final case class GameController(
           rsClient.nextRandom
         )
       },
-      clientTextStore = this.clientTextStore // Keep text store as-is for now
+      clientTextStore = rewoundClientTextStore
     ).copyWithAtomicSave
 
     (newController, resyncedCount)

src/test/scala/net/eagle0/eagle/client_text/ClientTextStoreImplTest.scala

@@ -764,4 +764,44 @@ class ClientTextStoreImplTest extends AnyFlatSpec with MockFactory with Matchers
       CompleteClientText("unknown", "unknown text", requestedAfterHistoryCount)
     )
   }
+
+  "withoutTextsAfter" should "remove generated text rows after the rewind target" in {
+    val store = ClientTextStoreImpl(
+      pregenerated = pregeneratedClientTextStore,
+      persister = mockPersister,
+      completeTexts = Map(
+        "complete"      -> CompleteClientText("complete", "already done", 400),
+        "staleComplete" -> CompleteClientText("staleComplete", "future text", 500)
+      ),
+      incompleteTexts = Map(
+        "oldIncomplete"   -> IncompleteClientText("oldIncomplete", "", 300, genericLlmRequest, 1L, 1L),
+        "staleIncomplete" -> IncompleteClientText("staleIncomplete", "", 500, genericLlmRequest, 1L, 1L)
+      ),
+      unrequestedTexts = Map(
+        "oldUnrequested"   -> UnrequestedClientText("oldUnrequested", 300, genericLlmRequest),
+        "staleUnrequested" -> UnrequestedClientText("staleUnrequested", 500, genericLlmRequest)
+      ),
+      accessibleTo = Map(
+        "complete"         -> Vector(1),
+        "staleComplete"    -> Vector(1),
+        "oldIncomplete"    -> Vector(1),
+        "staleIncomplete"  -> Vector(1),
+        "oldUnrequested"   -> Vector(1),
+        "staleUnrequested" -> Vector(1)
+      ),
+      savedCompleteCount = 0,
+      incompleteTextsAreSaved = true,
+      accessibleToIsSaved = true,
+      hiddenFromClients = Set("staleComplete", "staleIncomplete", "complete"),
+      hiddenFromClientsIsSaved = true
+    )
+
+    val updated = store.withoutTextsAfter(400)
+
+    updated.completeTexts.keySet shouldBe Set("complete")
+    updated.incompleteTexts.keySet shouldBe Set("oldIncomplete")
+    updated.unrequestedTexts.keySet shouldBe Set("oldUnrequested")
+    updated.accessibleTo.keySet shouldBe Set("complete", "oldIncomplete", "oldUnrequested")
+    updated.hiddenFromClients shouldBe Set("complete")
+  }
 }

src/test/scala/net/eagle0/eagle/service/UnrequestedTextHandlerTest.scala

@@ -88,6 +88,7 @@ class UnrequestedTextHandlerTest extends AnyFlatSpec with BeforeAndAfterEach wit
     (() => mockCts.beginTransaction()).expects().anyNumberOfTimes(): Unit
     (() => mockCts.commitTransaction()).expects().anyNumberOfTimes(): Unit
     (() => mockCts.rollbackTransaction()).expects().anyNumberOfTimes(): Unit
+    (() => mockHistory.count).expects().returning(100000).anyNumberOfTimes(): Unit
   }
 
   behavior of "UnrequestedTextHandlerTest"
@@ -175,6 +176,44 @@ class UnrequestedTextHandlerTest extends AnyFlatSpec with BeforeAndAfterEach wit
     result.newlyCompletedTexts shouldBe Vector.empty
   }
 
+  it should "throw on stale unrequested text requests left behind by a rewind" in {
+    (() => mockCts.beginTransaction()).expects().anyNumberOfTimes(): Unit
+    (() => mockCts.commitTransaction()).expects().anyNumberOfTimes(): Unit
+    (() => mockCts.rollbackTransaction()).expects().anyNumberOfTimes(): Unit
+
+    val staleRequest = UnrequestedClientText(
+      id = "staleRequest",
+      requestedAfterHistoryCount = 8550,
+      llmRequest = llmRequest1
+    )
+
+    (() => mockHistory.count).expects().returning(8545).anyNumberOfTimes(): Unit
+    (() => mockCts.unrequestedTexts)
+      .expects()
+      .returning(Map("staleRequest" -> staleRequest)): Unit
+    mockHistory.stateAfter.expects(8550).never(): Unit
+
+    (mockLlmResolver
+      .resolveLlmRequests(
+        _: Vector[LlmRequestWithGameState],
+        _: ClientTextStore,
+        _: FunctionalRandom
+      ))
+      .expects(*, *, *)
+      .never(): Unit
+
+    val thrown = intercept[EagleInternalException] {
+      handler.clientTextStoreWithHandledUnrequestedTexts(
+        clientTextStore = mockCts,
+        gameHistory = mockHistory,
+        randomLong = 1
+      )
+    }
+    thrown.getMessage shouldBe
+      "Stale pending text request staleRequest: requestedAfterHistoryCount=8550 is after current history count 8545. " +
+      "This usually means the game was rewound without pruning its text store."
+  }
+
   it should "throw an exception if there's a dependency circle" in {
     expectTransactionCalls()
     val unrequested1 = UnrequestedClientText(

src/test/scala/net/eagle0/eagle/service/controller/GameControllerTest.scala

@@ -455,7 +455,8 @@ class GameControllerTest extends AnyFlatSpec with MockFactory with Matchers with
       logWriter = mockLogWriter
     )
 
-    val mockRewoundEngine = mock[Engine]
+    val mockRewoundEngine         = mock[Engine]
+    val mockPrunedClientTextStore = mock[ClientTextStore]
 
     // Setup mocks for the rewound engine
     (() => mockRewoundEngine.factionIds)
@@ -544,9 +545,19 @@ class GameControllerTest extends AnyFlatSpec with MockFactory with Matchers with
       .returns(mockRewoundEngine)
       .once(): Unit
 
-    (() => mockClientTextStore.saved)
+    mockClientTextStore.withoutTextsAfter
+      .expects(3)
+      .returns(mockPrunedClientTextStore)
+      .once(): Unit
+
+    mockPrunedClientTextStore.hasIncompleteTextsAccessibleTo
+      .expects(*)
+      .returning(false)
+      .anyNumberOfTimes(): Unit
+
+    (() => mockPrunedClientTextStore.saved)
       .expects()
-      .returns(mockClientTextStore)
+      .returns(mockPrunedClientTextStore)
       .anyNumberOfTimes(): Unit
 
     val controller = GameController(
@@ -567,5 +578,6 @@ class GameControllerTest extends AnyFlatSpec with MockFactory with Matchers with
     newController.humanClients.foreach { client =>
       client.unfilteredKnownHistoryCount shouldBe 1 // Rewound history has 1 result
     }
+    newController.clientTextStore shouldBe mockPrunedClientTextStore
   }
 }