From cfc07bbf3943930bd5de6ec956dae217f857d26a Mon Sep 17 00:00:00 2001 From: yucheng Date: Tue, 29 Sep 2026 23:01:50 +0000 Subject: [PATCH] test(integration): cover previous_response_id history rebuilt from an hourly cold storage object Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../test_s3_v2_partition_granularity.py | 58 +++++++++++++++++++ 1 file changed, 58 insertions(+) diff --git a/tests/integration/observability/test_s3_v2_partition_granularity.py b/tests/integration/observability/test_s3_v2_partition_granularity.py index b8f52951296..7730618241d 100644 --- a/tests/integration/observability/test_s3_v2_partition_granularity.py +++ b/tests/integration/observability/test_s3_v2_partition_granularity.py @@ -468,6 +468,64 @@ def test_s3_v2_hour_cold_storage_key_names_the_uploaded_object_and_reads_back(ga assert prompts[1] not in json.dumps(missing.json()["response"]), missing.text +def test_s3_v2_hour_cold_storage_rebuilds_previous_response_id_history_from_the_hour_object( + gateway: Gateway, tmp_path: Path +) -> None: + marker: Final = "s3hsess" + uuid.uuid4().hex[:8] + upstream: Final = CountingUpstream() + histories: Final[list[str]] = [] # mutable-ok: appended per upstream request by the scripted provider thread + reads: Final[list[str]] = [] # mutable-ok: appended per sink GET by the recording sink thread + sink: Final = RecordingS3Sink(delay_seconds=0.05) + + def provider_reply(request: Request) -> Reply: + histories.append(request.body.decode()) + return upstream.respond(request) + + def bucket_reply(request: Request) -> Reply: + if request.method == "GET": + reads.append(unquote(request.target)) + return sink.respond(request) + + with ( + wire_server(provider_reply) as provider, + wire_server(bucket_reply) as bucket, + _s3_proxy(gateway, tmp_path, bucket.url, HOUR, {"cold_storage_custom_logger": "s3_v2"}) as owned, + owned.gateway.scenario() as scenario, + ): + _, anthropic_model, key = _models(scenario, provider.url) + first: Final = owned.gateway.request( + "POST", "/v1/responses", {"model": anthropic_model, "input": f"{marker}-first"}, key=key + ) + assert first.status_code == 200, first.text + rows: Final = eventually( + lambda: read_rows( + 'SELECT request_id, metadata FROM "LiteLLM_SpendLogs" WHERE model_group=%s', (anthropic_model,) + ), + lambda values: len(values) == 1, + seconds=60, + ) + metadata: Final = rows[0]["metadata"] + cold_key: Final = str( + object_value(json.loads(metadata) if isinstance(metadata, str) else metadata)["cold_storage_object_key"] + ) + eventually(sink.objects, lambda objects: f"/{BUCKET}/{quote(cold_key, safe='/')}" in objects, seconds=30) + second: Final = owned.gateway.request( + "POST", + "/v1/responses", + {"model": anthropic_model, "input": f"{marker}-second", "previous_response_id": first.json()["id"]}, + key=key, + ) + objects: Final = sink.objects() + assert second.status_code == 200, second.text + assert second.json()["id"] != first.json()["id"], second.text + assert re.fullmatch(rf"{re.escape(PREFIX)}/\d{{4}}-\d{{2}}-\d{{2}}/\d{{2}}/time-[^/]+\.json", cold_key), cold_key + assert _outside_layout(objects, "hour") == () + assert f"/{BUCKET}/{cold_key}" in reads, reads + assert len(histories) == 2, histories + assert f"{marker}-first" in histories[0] and f"{marker}-second" not in histories[0], histories[0] + assert f"{marker}-first" in histories[1] and f"{marker}-second" in histories[1], histories[1] + + def test_s3_v2_audit_logs_follow_the_audit_params_granularity_not_the_request_logs( gateway: Gateway, tmp_path: Path ) -> None: