From 7adb2062eb2220e26447f652b0c724285806b071 Mon Sep 17 00:00:00 2001 From: Severus <1529378848@qq.com> Date: Mon, 9 Nov 2026 20:28:09 +0800 Subject: [PATCH] fix(gateway): keep retryable thread inputs ahead of terminal receipts (#7972) (#8008) * fix(gateway): keep retryable thread inputs ahead of terminal receipts (#7972) * fix(gateway): document thread-input ranking policy (#7972) * test(gateway): cover thread-input ranking and receipt starvation (#7972) * test(gateway): format thread-input regression tests with repo ruff config (#7972) --- backend/app/gateway/run_inputs.py | 10 +++- backend/tests/test_run_inputs.py | 77 +++++++++++++++++++++++++++++++ 2 files changed, 86 insertions(+), 1 deletion(-) diff --git a/backend/app/gateway/run_inputs.py b/backend/app/gateway/run_inputs.py index e59812f75f..9db0a04d6b 100644 --- a/backend/app/gateway/run_inputs.py +++ b/backend/app/gateway/run_inputs.py @@ -75,7 +75,15 @@ async def list_thread_inputs(thread_id: str, request: Request): if status in ("pending", "applying") and (row["status"] not in ("pending", "running", "success") or row.get("stop_reason")): status = "not_applied" items.append({key: value for key, value in {**entry, "status": status, "target_run_id": row["run_id"]}.items() if key != "context"}) - return {"items": items[-32:], "active_run_id": await _store(request).active_input_run(thread_id, user_id=user_id)} + # Rank unresolved inputs (pending / not_applied / ...) ahead of terminal + # applied/cancelled receipts, prefer the newest entry within each group, + # cap at 32 records, then restore chronological order. A plain items[-32:] + # slice lets two full batches of receipts hide an older retryable input + # (issue #7972). If more than 32 inputs are unresolved, this deliberately + # returns 32 unresolved rows and no receipts for submission reconciliation; + # that trade-off follows the priority the issue asks for. + keep = sorted(range(len(items)), key=lambda i: (items[i]["status"] in ("applied", "cancelled"), -i))[:32] + return {"items": [items[i] for i in sorted(keep)], "active_run_id": await _store(request).active_input_run(thread_id, user_id=user_id)} async def dispatch_queued_inputs(inputs: RunInputs, *, agent_runs, thread_id: str): diff --git a/backend/tests/test_run_inputs.py b/backend/tests/test_run_inputs.py index 3491b7d2b6..c313c64234 100644 --- a/backend/tests/test_run_inputs.py +++ b/backend/tests/test_run_inputs.py @@ -322,3 +322,80 @@ async def test_http_input_admission_and_status_are_owner_scoped(input_store, mon user = User(id=uuid4(), email="other@example.com", password_hash="unused") assert (await client.post(url, json=body)).status_code == 404 assert (await client.get("/api/threads/thread/inputs")).json() == {"items": [], "active_run_id": None} + + +@pytest.mark.asyncio +async def test_list_thread_inputs_keeps_retryable_inputs_ahead_of_terminal_receipts(monkeypatch): + """Regression for #7972: 32 recent receipts must not hide an older retryable input.""" + from app.gateway import run_inputs as gateway_run_inputs + + def entry(index: int, status: str): + return {"id": f"input-{index:03d}", "status": status, "text": f"instruction {index}", "mode": "queue"} + + # The store returns runs newest-first; list_thread_inputs reverses them. + failed_row = { + "run_id": "run-failed", + "status": "failed", + "stop_reason": "error", + "run_inputs": {"items": [entry(0, "pending")]}, + } + later_rows = [] + for run_index in (1, 2): + items = [entry(run_index * 16 + i, "applied" if i % 16 else "cancelled") for i in range(16)] + later_rows.append({"run_id": f"run-{run_index}", "status": "success", "run_inputs": {"items": items}}) + rows = list(reversed(later_rows)) + [failed_row] + + class _FakeStore: + async def list_input_runs(self, thread_id, *, user_id): + return rows + + async def active_input_run(self, thread_id, *, user_id): + return "run-2" + + async def _current_user(_request): + return USER_ID + + monkeypatch.setattr(gateway_run_inputs, "get_current_user", _current_user) + monkeypatch.setattr(gateway_run_inputs, "_store", lambda _request: _FakeStore()) + + result = await gateway_run_inputs.list_thread_inputs("thread", SimpleNamespace()) + items = result["items"] + assert len(items) == 32 + assert items[0]["id"] == "input-000" and items[0]["status"] == "not_applied" + assert items[-1]["id"] == "input-047" + assert [i["id"] for i in items] == sorted(i["id"] for i in items) + assert "context" not in items[0] + assert result["active_run_id"] == "run-2" + + +@pytest.mark.asyncio +async def test_list_thread_inputs_starves_receipts_when_all_inputs_are_unresolved(monkeypatch): + """Record the deliberate trade-off: >32 unresolved inputs leave no receipt slots.""" + from app.gateway import run_inputs as gateway_run_inputs + + def entry(index: int, status: str): + return {"id": f"input-{index:03d}", "status": status, "text": f"instruction {index}", "mode": "queue"} + + rows = [ + {"run_id": "run-a", "status": "failed", "stop_reason": "error", "run_inputs": {"items": [entry(i, "pending") for i in range(16)]}}, + {"run_id": "run-b", "status": "failed", "stop_reason": "error", "run_inputs": {"items": [entry(16 + i, "pending") for i in range(16)]}}, + {"run_id": "run-c", "status": "success", "run_inputs": {"items": [entry(32, "applied")]}}, + ] + + class _FakeStore: + async def list_input_runs(self, thread_id, *, user_id): + return rows + + async def active_input_run(self, thread_id, *, user_id): + return None + + async def _current_user(_request): + return USER_ID + + monkeypatch.setattr(gateway_run_inputs, "get_current_user", _current_user) + monkeypatch.setattr(gateway_run_inputs, "_store", lambda _request: _FakeStore()) + + items = (await gateway_run_inputs.list_thread_inputs("thread", SimpleNamespace()))["items"] + assert len(items) == 32 + assert all(i["status"] == "not_applied" for i in items) + assert not any(i["id"] == "input-032" for i in items)