fix(rag): load deployments oldest first, reject file_id on stores that cannot take one, and show ingest errors in the dashboard

This commit is contained in:
mateo-berri 2026-10-14 11:17:01 +00:00
parent 154d784a0e
commit 480e596374
6 changed files with 100 additions and 10 deletions

View file

@ -8307,7 +8307,7 @@ class ProxyConfig:
try:
new_models: Final[Sequence[_ProxyModelRow]] = await ModelRepository(
WriterPinnedClient(prisma_client.db)
).table.find_many()
).table.find_many(order=[{"created_at": "asc"}, {"model_id": "asc"}])
self.auto_router_db_catalog = build_auto_router_catalog(new_models)
return new_models
except Exception as e:

View file

@ -322,13 +322,18 @@ class BaseRAGIngestion(ABC):
file_id=file_id,
)
try:
if existing_file_id and not self.supports_existing_file_id:
raise ValueError(
f"{self.__class__.__name__} does not support ingesting an existing file_id. "
"Upload file data or provide file_url instead."
)
if existing_file_id and not self.supports_existing_file_id:
return RAGIngestResponse(
id=self.ingest_id,
status="failed",
vector_store_id="",
file_id=None,
error=f"{self.__class__.__name__} does not support ingesting an existing file_id. "
"Upload file data or provide file_url instead.",
error_status_code=400,
)
try:
# Step 2: OCR (optional)
extracted_text: Final = await self.ocr(
file_content=file_content,

View file

@ -399,7 +399,7 @@ def _provider_environment(upstream: str) -> Mapping[str, str]:
def _deployment(
gateway: Gateway, scenario: Scenario, *, name: str, model: str, api_key: str, api_base: str, team_id: str | None
) -> None:
) -> str:
created: Final = gateway.post(
"/model/new",
{
@ -408,7 +408,9 @@ def _deployment(
"model_info": {} if team_id is None else {"team_id": team_id},
},
)
scenario.cleanups.callback(scenario.delete_model, string_value(object_value(created["model_info"])["id"]))
model_id: Final = string_value(object_value(created["model_info"])["id"])
scenario.cleanups.callback(scenario.delete_model, model_id)
return model_id
def _delete_credential(gateway: Gateway, name: str) -> None:
@ -1271,3 +1273,54 @@ async def test_worker_sigkill_mid_query_burst_leaves_the_sibling_on_team_credent
calls: Final = tuple(_call(request) for request in upstream.drain())
assert {(call.prefix, call.authorization) for call in calls} == {("team", f"Bearer {TEAM_API_KEY}")}, calls
assert sum(1 for call in calls if call.route == "/v1/chat/completions") == held_by[survivor_pid] + 1, calls
def test_a_team_with_two_openai_deployments_ingests_through_the_older_one_after_it_is_edited(tmp_path: Path) -> None:
with (
gateway_from_environment() as admin_gateway,
admin_gateway.scenario() as scenario,
wire_server(_responder(threading.Event())) as upstream,
):
older_model: Final = f"integration-rag-older-{uuid.uuid4().hex}"
newer_model: Final = f"integration-rag-newer-{uuid.uuid4().hex}"
team_id: Final = scenario.team(models=[older_model, newer_model])
older_id: Final = _deployment(
admin_gateway,
scenario,
name=older_model,
model="openai/gpt-4o-mini",
api_key=TEAM_API_KEY,
api_base=f"{upstream.url}/team/v1",
team_id=team_id,
)
_deployment(
admin_gateway,
scenario,
name=newer_model,
model="openai/gpt-4o-mini",
api_key=WILDCARD_API_KEY,
api_base=f"{upstream.url}/wildcard/v1",
team_id=team_id,
)
edited: Final = admin_gateway.request("PATCH", f"/model/{older_id}/update", {"litellm_params": {"timeout": 30}})
assert edited.status_code == 200, edited.text
key: Final = scenario.key(team_id=team_id)
with (
owned_proxy(
admin_gateway,
tmp_path,
_provider_environment(upstream.url),
remove_environment=("GOOGLE_API_KEY",),
workers=2,
) as gateway,
gateway.scenario() as owned_scenario,
):
upstream.drain()
response: Final = _ingest(gateway, key)
assert response.status_code == 200, response.text
store_id: Final = string_value(object_value(response.json())["vector_store_id"])
owned_scenario.cleanups.callback(gateway.post, "/vector_store/delete", {"vector_store_id": store_id})
assert tuple(_call(request).signature for request in upstream.drain()) == _ingest_signatures(
"team", TEAM_API_KEY, store_id
)

View file

@ -140,3 +140,11 @@ async def test_ingest_response_preserves_integer_exception_status_code(
)
assert response["error_status_code"] == expected_status_code
async def test_an_existing_file_id_on_a_store_that_cannot_take_one_fails_as_a_bad_request() -> None:
response: Final = await _FailingStoreIngestion(RuntimeError("store must not run")).ingest(file_id="file-abc")
assert response["status"] == "failed"
assert response["error_status_code"] == 400
assert "does not support ingesting an existing file_id" in str(response["error"])

View file

@ -937,3 +937,27 @@ describe("modelCostMap", () => {
expect(mockFetch.mock.calls[1][0]).toMatch(/\/public\/litellm_model_cost_map$/);
});
});
describe("ragIngestCall", () => {
let currentFetch: typeof global.fetch;
beforeEach(() => {
currentFetch = global.fetch;
});
afterEach(() => {
global.fetch = currentFetch;
});
it("rejects with the proxy's error text when the detail is an object", async () => {
const reason = "Error code: 404 - No vector store found with id 'vs_missing'";
global.fetch = vi.fn().mockResolvedValue({
ok: false,
json: async () => ({ detail: { error: reason } }),
} as any) as any;
await expect(
Networking.ragIngestCall("sk-test", new File(["text"], "doc.txt"), "openai", "vs_missing"),
).rejects.toThrow(reason);
});
});

View file

@ -5712,7 +5712,7 @@ export const ragIngestCall = async (
if (!response.ok) {
const error = await response.json();
throw new Error(error.error?.message || error.detail || "Failed to ingest document");
throw new Error(deriveErrorMessage(error));
}
return await response.json();