fix(proxy): warn when a deferred background policy was matched through a request tag

Retrieval re-matches only the key, team, and model scopes, so a post_call
policy that reached a pending background response through a request-body
tag does not govern the completed response. Log that at submit, next to the
deferral, and cover the retrieval re-match with tag-scoped tests.

litellm/proxy/utils.py

@@ -613,9 +613,31 @@ def _defer_post_call_pipelines(
         response.status,
         ", ".join(policy_name for policy_name, _pipeline in deferred),
     )
+    tag_matched: Final = _tag_matched_deferrals(data, deferred)
+    if tag_matched:
+        verbose_proxy_logger.warning(
+            "Policy engine: background response %s matched post_call policies through a request tag at submit; "
+            "retrieval re-matches only the key, team, and model scopes, so a tag carried in the request body "
+            "does not govern the completed response: %s",
+            response.id,
+            ", ".join(f"{policy_name} ({source})" for policy_name, source in tag_matched),
+        )
     _withdraw_deferred_claims(data, deferred)
 
 
+def _tag_matched_deferrals(
+    data: Mapping[str, object], deferred: Sequence[tuple[str, "GuardrailPipeline"]]
+) -> tuple[tuple[str, str], ...]:
+    sources: Final = _policy_state_metadata(data).get("policy_sources")
+    if not isinstance(sources, dict):
+        return ()
+    return tuple(
+        (policy_name, str(sources[policy_name]))
+        for policy_name, _pipeline in deferred
+        if policy_name in sources and "tag:" in str(sources[policy_name])
+    )
+
+
 def _pipeline_is_streamable(policy_name: str, pipeline: "GuardrailPipeline") -> bool:
     unsupported: Final = tuple(
         dict.fromkeys(

tests/test_litellm/proxy/policy_engine/test_response_retrieval.py

@@ -68,6 +68,7 @@ def policy_engine():
             "response-governance": _pipeline_policy("output-word-filter"),
             "input-governance": _pipeline_policy("input-word-filter", mode="pre_call"),
             "team-governance": _pipeline_policy("team-word-filter"),
+            "tag-governance": _pipeline_policy("tag-word-filter"),
         }
     )
     attachment_registry.load_attachments(
@@ -75,6 +76,7 @@ def policy_engine():
             {"policy": "response-governance", "models": [GOVERNED_MODEL_GROUP]},
             {"policy": "input-governance", "models": [GOVERNED_MODEL_GROUP]},
             {"policy": "team-governance", "teams": ["governed-team"]},
+            {"policy": "tag-governance", "tags": ["governed"]},
         ]
     )
     yield
@@ -119,6 +121,26 @@ def test_key_and_team_context_also_governs_retrieval(policy_engine):
     assert _attached_pipelines(data) == (("team-governance", "team-word-filter"),)
 
 
+def test_tag_attached_policy_is_not_re_matched_when_the_retrieval_carries_no_tag(policy_engine):
+    data = _retrieval_data(GOVERNED_MODEL_ID)
+
+    attach_post_call_pipelines_to_retrieval(data=data, user_api_key_dict=UserAPIKeyAuth(), llm_router=_router())
+
+    assert _attached_pipelines(data) == (("response-governance", "output-word-filter"),)
+
+
+def test_tag_attached_policy_governs_a_retrieval_whose_metadata_carries_the_tag(policy_engine):
+    data: dict[str, object] = {
+        "response_id": _encoded_response_id(UNGOVERNED_MODEL_ID),
+        "litellm_metadata": {"tags": ["governed"]},
+    }
+
+    attach_post_call_pipelines_to_retrieval(data=data, user_api_key_dict=UserAPIKeyAuth(), llm_router=_router())
+
+    assert _attached_pipelines(data) == (("tag-governance", "tag-word-filter"),)
+    assert data["litellm_metadata"]["policy_sources"] == {"tag-governance": "tag:governed"}
+
+
 def test_retrieval_of_an_ungoverned_model_attaches_nothing(policy_engine):
     data = _retrieval_data(UNGOVERNED_MODEL_ID)
 

tests/test_litellm/proxy/utils/proxy_logging/test_guardrail_pipeline.py

@@ -1489,7 +1489,9 @@ def _output_passing_callbacks() -> list[CustomGuardrail]:
     return [OutputPassingGuardrail(guardrail_name="gr-post", event_hook=GuardrailEventHooks.post_call, default_on=False)]
 
 
-def _claimed_post_call_pipeline_data(*policy_names: str, extra_guardrails: dict[str, list[str]] | None = None):
+def _claimed_post_call_pipeline_data(
+    *policy_names: str, extra_guardrails: dict[str, list[str]] | None = None, policy_source: str = "model:m"
+):
     from litellm.proxy.policy_engine.policy_registry import get_policy_registry
 
     step = {"guardrail": "gr-post", "on_pass": "allow", "on_fail": "block"}
@@ -1511,7 +1513,7 @@ def _claimed_post_call_pipeline_data(*policy_names: str, extra_guardrails: dict[
             "_pipeline_managed_guardrails": {"gr-post"},
             "applied_policies": list(policy_names),
             "applied_guardrails": ["gr-post", *(g for gs in (extra_guardrails or {}).values() for g in gs)],
-            "policy_sources": {policy_name: "model:m" for policy_name in policy_names},
+            "policy_sources": {policy_name: policy_source for policy_name in policy_names},
         },
     }
 
@@ -1542,6 +1544,47 @@ async def test_pending_background_response_withdraws_the_deferred_policy_claims(
     assert "applied_guardrails" not in data["metadata"]
 
 
+@pytest.mark.asyncio
+async def test_pending_background_response_warns_when_the_deferred_policy_was_matched_through_a_tag(
+    proxy_logging, make_user_api_key_auth, monkeypatch, clear_policy_registry, caplog
+):
+    monkeypatch.setattr(litellm, "callbacks", _output_blocking_callbacks({}))
+    monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", None, raising=False)
+    data = _claimed_post_call_pipeline_data("response-governance", policy_source="tag:governed+model:m")
+
+    with caplog.at_level(logging.WARNING, logger="LiteLLM Proxy"):
+        out = await proxy_logging.post_call_success_hook(
+            data=data, response=_background_response("queued"), user_api_key_dict=make_user_api_key_auth()
+        )
+
+    assert out.status == "queued"
+    assert "policy_sources" not in data["metadata"]
+    assert [
+        message for message in _warnings(caplog) if "response-governance (tag:governed+model:m)" in message
+    ] == [
+        "Policy engine: background response resp_bg matched post_call policies through a request tag at submit; "
+        "retrieval re-matches only the key, team, and model scopes, so a tag carried in the request body "
+        "does not govern the completed response: response-governance (tag:governed+model:m)"
+    ]
+
+
+@pytest.mark.asyncio
+async def test_pending_background_response_matched_through_its_model_does_not_warn(
+    proxy_logging, make_user_api_key_auth, monkeypatch, clear_policy_registry, caplog
+):
+    monkeypatch.setattr(litellm, "callbacks", _output_blocking_callbacks({}))
+    monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", None, raising=False)
+
+    with caplog.at_level(logging.WARNING, logger="LiteLLM Proxy"):
+        await proxy_logging.post_call_success_hook(
+            data=_claimed_post_call_pipeline_data("response-governance"),
+            response=_background_response("queued"),
+            user_api_key_dict=make_user_api_key_auth(),
+        )
+
+    assert _warnings(caplog) == []
+
+
 @pytest.mark.asyncio
 async def test_pending_background_response_keeps_the_claim_of_a_policy_that_runs_outside_its_pipeline(
     proxy_logging, make_user_api_key_auth, monkeypatch, clear_policy_registry