Bug/XCom sequence slice endpoint returns wrong rows for crossed-bound slices (negative SQL LIMIT)

### Under which category would you file this issue?

Airflow Core

### Apache Airflow version

main (3.4.0.dev0), reproduced at commit a3f51fcd9c

### What happened and how to reproduce it?

## Issue Description

The mapped XCom sequence-slice endpoint in the Execution API returns incorrect results for **crossed-bound slices**—cases where the effective `stop` index comes before the effective `start` index. According to Python slicing semantics, these slices should always evaluate to an empty list.

**Endpoint**

```text
GET /execution/xcoms/{dag_id}/{run_id}/{task_id}/{key}/slice?start=&stop=&step=
```

**Handler**

```text
airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py
get_mapped_xcom_by_slice()
```

The handler is intended to reproduce Python list slicing over stored XCom rows, but it translates the slice into SQLAlchemy's `Query.slice(a, b)`, which generates:

```sql
OFFSET a
LIMIT (b - a)
```

Unlike Python slicing, SQLAlchemy does **not** clamp negative limits. When `b < a`, the generated `LIMIT` becomes negative, resulting in backend-dependent behavior.

### Backend behavior

- **SQLite**
- A negative `LIMIT` is treated as **"no limit"**.
- The query silently returns rows from `OFFSET` onward.
- **Result:** incorrect data is returned.

- **PostgreSQL / MySQL**
- Negative `LIMIT` values are rejected.
- **Result:** the request fails with a server error.

The endpoint should instead return an empty result (`[]`) for all crossed-bound slices, matching Python semantics.

---

## Examples

Assume six stored XCom values:

```text
["v0", "v1", "v2", "v3", "v4", "v5"]
```

### Case 1

Request:

```text
start=2&stop=-3&step=-1
```

Python:

```python
list(range(6))[2:-3:-1]
# []
```

Current implementation:

```text
stop -> 3
slice(4, 3)
OFFSET 4
LIMIT -1
```

SQLite returns:

```text
["v5", "v4"]
```

Expected:

```text
[]
```

---

### Case 2

Request:

```text
start=4&stop=2
```

Python:

```python
list(range(6))[4:2]
# []
```

Current implementation:

```text
slice(4, 2)
OFFSET 4
LIMIT -2
```

SQLite returns:

```text
["v4", "v5"]
```

Expected:

```text
[]
```

This second case is particularly interesting because it does **not** involve mixed-sign normalization or a `COUNT` query. It simply occurs when `stop < start`.

---

## Existing Test Coverage

The current test suite does not include any crossed-bound slice scenarios, so the negative-`LIMIT` path is never exercised.

---

## Steps to Reproduce

The underlying issue is in the slice-to-SQL translation and can be reproduced without Airflow using only the Python standard library.

```python
import sqlite3

con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE x(map_index INTEGER)")
con.executemany("INSERT INTO x VALUES (?)", [(i,) for i in range(6)])

def q(sql):
return [r[0] for r in con.execute(sql)]

# SQLAlchemy Query.slice(a, b)
# => OFFSET a
# => LIMIT (b - a)

# Case A: start=2, stop=-3, step=-1
# normalized to slice(4, 3)
print(
"A SQLite:",
q("SELECT map_index FROM x ORDER BY map_index ASC LIMIT -1 OFFSET 4"),
)
# [4, 5]

# Case B: start=4, stop=2
# translated to slice(4, 2)
print(
"B SQLite:",
q("SELECT map_index FROM x ORDER BY map_index ASC LIMIT -2 OFFSET 4"),
)
# [4, 5]

print("Python A:", list(range(6))[2:-3:-1])
# []

print("Python B:", list(range(6))[4:2])
# []
```

Output:

```text
A SQLite: [4, 5]
B SQLite: [4, 5]
Python A: []
Python B: []
```

The same behavior can be reproduced end-to-end by storing at least six mapped XCom values and requesting either:

```text
start=4&stop=2
```

or

```text
start=2&stop=-3&step=-1
```

On SQLite the endpoint returns incorrect rows instead of an empty list.

On PostgreSQL/MySQL the request fails due to the negative `LIMIT`.

---

## Root Cause

The crossed-bound branches currently call:

```python
query.slice(start, stop)
```

and

```python
query.slice(stop + 1, start + 1)
```

assuming that `Query.slice()` behaves like Python slicing.

However, SQLAlchemy directly translates the bounds into:

```sql
OFFSET a
LIMIT (b - a)
```

without clamping negative limits.

As a result, crossed-bound slices generate a negative SQL `LIMIT`, producing incorrect results on SQLite and backend errors on PostgreSQL/MySQL.

The affected code paths are currently around:

```text
airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py
```

specifically the branches that call:

```python
query.slice(start, stop)
```

and

```python
query.slice(stop + 1, start + 1)
```
## Proposed Fix

The issue only affects the code paths that call `Query.slice(low, high)`. These are the only branches where the computed bounds can cross, causing SQLAlchemy to generate a negative `LIMIT`.

The branches using `.limit(...)` and `.offset(...)` are already safe because their arguments are guaranteed to be non-negative by their surrounding conditions.

A small helper could centralize the guard and preserve Python slicing semantics before calling `Query.slice()`:

```python
def _sliced_or_empty(query, low: int, high: int):
"""Apply ``.slice(low, high)`` but return no rows when the bounds are crossed.

SQLAlchemy's ``Query.slice(low, high)`` compiles to:

OFFSET low
LIMIT (high - low)

It does not clamp a negative LIMIT. A crossed slice (``high <= low``),
which Python evaluates as an empty sequence, therefore becomes backend-
dependent:

* SQLite interprets a negative LIMIT as "no limit", returning incorrect rows.
* PostgreSQL/MySQL reject the query because LIMIT cannot be negative.
"""
if high <= low:
return query.limit(0)

return query.slice(low, high)
```

The four `Query.slice()` call sites in `get_mapped_xcom_by_slice()` could then become:

```python
# start >= 0, stop given

# step >= 0
query = _sliced_or_empty(query, start, stop)

# step < 0
query = _sliced_or_empty(query, stop + 1, start + 1)

# start < 0, stop given

# step > 0
query = _sliced_or_empty(query, -1 - start, -1 - stop)

# step < 0
query = _sliced_or_empty(query, -stop, -start)
```

### Why `high <= low`?

By the time these branches execute:

- the query ordering has already been normalized (ascending or descending),
- the direction of iteration is handled separately via the final `values[::step]`.

Each `Query.slice(low, high)` therefore always represents an **ascending window** over the already-normalized ordering.

A crossed window (`high <= low`) is exactly the case that Python evaluates as:

```python
[]
```

Returning:

```python
query.limit(0)
```

preserves Python semantics while producing consistent SQL across SQLite, PostgreSQL, and MySQL.

### Why only guard `Query.slice()`?

The remaining branches use expressions such as:

```python
query.limit(stop)
query.limit(-stop - 1)
query.limit(start + 1)
query.limit(-start)
```

In every case, the surrounding branch conditions already guarantee these values are non-negative.

Only `Query.slice(low, high)` combines two independently computed bounds, making it possible for the generated SQL to contain:

```sql
LIMIT (high - low)
```

where `(high - low) < 0`.

---

## Suggested Tests

Add regression cases to:

```text
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py
```

specifically to the slice endpoint test suite.

Suggested parameterized cases include:

- `start=4, stop=2`
- Simple crossed positive bounds.
- No mixed-sign normalization.
- No `COUNT` query.

- `start=2, stop=-3, step=-1`
- Exercises the negative-stop normalization path.

- Descending equivalents where `start < 0`
- Covers the branches using:
- `slice(-1 - start, -1 - stop)`
- `slice(-stop, -start)`

Each of these cases should currently fail on the existing implementation:

- **SQLite:** returns incorrect rows instead of an empty list.
- **PostgreSQL/MySQL:** fails due to a negative `LIMIT`.

With the proposed guard in place, all cases should consistently return an empty result, matching Python slicing semantics.

---

> **Note**
>
> This is one possible implementation approach intended to facilitate discussion.
> An alternative would be to clamp the computed bounds individually at each call site, but centralizing the behavior in a helper keeps the four `Query.slice()` branches consistent and makes the intent explicit.
>
> If this approach is acceptable, I'd be happy to open a PR with the regression tests and the corresponding fix.

### What you think should happen instead?

## What do I think should happen instead?

The slice endpoint should match Python list-slicing semantics.

For crossed-bound slices, where the effective `stop` index is before the effective `start` index, the endpoint should return an empty list:

```python
list(range(6))[4:2] == []
list(range(6))[2:-3:-1] == []
```

The API should return:

```json
[]
```

consistently across SQLite, PostgreSQL, and MySQL.

It should not generate a negative SQL `LIMIT`.

---

## What do I think went wrong?

The handler translates Python-style slice bounds into SQLAlchemy `Query.slice(low, high)`.

However, `Query.slice(low, high)` does not behave exactly like Python slicing. It compiles to:

```sql
OFFSET low
LIMIT (high - low)
```

When `high <= low`, the computed `LIMIT` becomes zero or negative.

Python treats that as an empty slice, but SQL backends do not:

- SQLite treats negative `LIMIT` as unlimited, so it can return incorrect rows.
- PostgreSQL/MySQL reject negative `LIMIT`, causing the request to fail.

So the bug comes from assuming `Query.slice()` clamps crossed bounds the same way Python slicing does. It does not.

### Operating System

macOS 26.5.1 (arm64)

### Deployment

Virtualenv installation

### Apache Airflow Provider(s)

_No response_

### Versions of Apache Airflow Providers

Not Applicable

### Official Helm Chart version

Not Applicable

### Kubernetes Version

Not Applicable"

### Helm Chart configuration

Not Applicable

### Docker Image customizations

Not Applicable. The bug is in the core Execution API logic.

### Anything else?

The bug is hidden here, Exact line numbers in airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py:

get_mapped_xcom_by_slice() — starts at line 185
The four crossed-capable .slice() calls:
line 238 — query = query.slice(start, stop)
line 240 — query = query.slice(stop + 1, start + 1)
line 253 — query = query.slice(-1 - start, -1 - stop)
line 255 — query = query.slice(-stop, -start)

### Are you willing to submit PR?

- [x] Yes I am willing to submit a PR!

### Code of Conduct

- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

Read get_mapped_xcom_by_slice() in airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py, focusing on the four Query.slice() calls around lines 238-255. Run the slice endpoint tests in airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py, then add regression cases for the crossed bounds described in the issue. Done means those cases return [] consistently without generating a negative SQL LIMIT.

Written by the indexing model from the issue text.