test: cover targeted and expired ledger claims

Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>

tests/adapters/persistence/test_postgresql_ledger_outbox.py

@@ -208,6 +208,29 @@ def test_stale_claim_cannot_overwrite_new_owner_outcome(pg_clean):
     assert stored.last_outcome is OutboxOutcome.COMPLETED
 
 
+def test_original_owner_can_finalize_expired_unreclaimed_claim(pg_clean):
+    now = {"t": datetime(2026, 6, 4, tzinfo=UTC)}
+    outbox = PostgreSQLLedgerOutbox(
+        pg_clean,
+        instance_key="inst-a",
+        claim_ttl_sec=5.0,
+        worker_id="worker-a",
+        clock=lambda: now["t"],
+    )
+    row = outbox.enqueue(_trade(execution_id="pg-expired"))
+
+    [claim] = outbox.claim_pending()
+    now["t"] = datetime(2026, 6, 4, 0, 0, 6, tzinfo=UTC)
+    stored = outbox.mark_outcome(
+        row.correlation_id,
+        OutboxOutcome.COMPLETED,
+        expected_claim_expires_at=claim.claim_expires_at,
+    )
+
+    assert stored.status is WorkItemStatus.COMPLETED
+    assert stored.last_outcome is OutboxOutcome.COMPLETED
+
+
 def test_writer_records_retryable_403_without_cross_instance_delivery(pg_clean):
     class _ForbiddenLedger:
         def __init__(self):

tests/application/test_ledger_outbox.py

@@ -77,14 +77,14 @@ def test_process_correlation_id_only_drains_requested_row():
     first = outbox.enqueue(_trade(execution_id="e-1"))
     second = outbox.enqueue(_trade(execution_id="e-2"))
 
-    outcome = writer.process_correlation_id(first.correlation_id)
+    outcome = writer.process_correlation_id(second.correlation_id)
 
     assert outcome is OutboxOutcome.COMPLETED
     assert ledger.count() == 1
     assert writer.counters["claim:claimed"] == 1
     assert writer.counters["outcome:completed"] == 1
-    assert outbox.get(first.correlation_id).status is WorkItemStatus.COMPLETED
-    assert outbox.get(second.correlation_id).status is WorkItemStatus.PENDING
+    assert outbox.get(first.correlation_id).status is WorkItemStatus.PENDING
+    assert outbox.get(second.correlation_id).status is WorkItemStatus.COMPLETED
 
 
 def test_process_correlation_id_skips_foreign_instance_row():

trex_core/adapters/persistence/postgresql.py

@@ -747,8 +747,11 @@ class PostgreSQLLedgerOutbox:
                 if expected_claim_expires_at is None:
                     where += "AND claim_owner IS NULL AND claim_expires_at IS NULL"
                 else:
-                    where += "AND claim_owner=%s AND claim_expires_at=%s"
-                    params.extend([self._worker_id, expected_claim_expires_at])
+                    where += (
+                        "AND claim_owner=%s "
+                        "AND (claim_expires_at=%s OR (status='pending' AND claim_expires_at<=%s))"
+                    )
+                    params.extend([self._worker_id, expected_claim_expires_at, self._now()])
                 cur.execute(
                     "UPDATE ledger_outbox SET status=%s, attempts=attempts+%s, "
                     "last_outcome=%s, last_error=%s, next_retry_at=%s, updated_at=%s, "

trex_core/application/ledger_outbox.py

@@ -522,10 +522,16 @@ class InMemoryLedgerOutbox:
         *,
         correlation_id: str | None = None,
     ) -> list[LedgerWorkItem]:
-        pending = self.list_pending(limit if correlation_id is None else max(limit, 1))
         if correlation_id is None:
-            return pending[:limit]
-        return [row for row in pending if row.correlation_id == correlation_id][:1]
+            return self.list_pending(limit)
+        now = self._now()
+        with self._lock:
+            row = self._rows.get(correlation_id)
+            if row is None or row.status is not WorkItemStatus.PENDING:
+                return []
+            if row.next_retry_at is not None and row.next_retry_at > now:
+                return []
+            return [row]
 
     def pending_cash_delta(self) -> dict[str, float]:
         with self._lock: