diff --git a/AGENTS.md b/AGENTS.md
index e955f8b36a..49149dfa81 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -68,6 +68,8 @@ The materialization contract is also covered by [`docs/doctoring/exact-artifact-
 
 ## Verification discipline
 
+- producer가 안전한 로그 필드를 추가하면 exact revision 쌍으로 consumer sanitizer를 통과시켜 allowlist의 누락을 확인한다. producer 단위 테스트 성공만으로 CI artifact 보존을 주장하지 않으며, 연결 검증에서도 raw 본문 비출력을 유지한다.
+
 Many agent sessions work this organization concurrently under the same standing
 brief. Silence is not evidence: "I have not touched X" describes one session's
 history, never the organization's actual state.
diff --git a/scripts/ci/sanitize_contextual_orchestrator_sidecar_stream.py b/scripts/ci/sanitize_contextual_orchestrator_sidecar_stream.py
index e055513954..9df99e1b70 100644
--- a/scripts/ci/sanitize_contextual_orchestrator_sidecar_stream.py
+++ b/scripts/ci/sanitize_contextual_orchestrator_sidecar_stream.py
@@ -8,8 +8,12 @@
 
 
 _REQUEST_FAILED = re.compile(
+    r"^(?:(?:[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2},[0-9]{3} )?"
+    r"(?:DEBUG|INFO|WARNING|ERROR)[: ]contextual_orchestrator\.server[: ])?"
     r"request_failed status=(?P<status>[1-5][0-9]{2}) "
     r"code=(?P<code>[A-Za-z0-9_.-]{1,64})"
+    r"(?= |$)(?: request_id=(?P<request_id>[0-9a-f]{32}|<omitted>)(?= |$))?"
+    r"(?! request_id=)"
 )
 _PROVIDER_DISCOVERY_FAILED = re.compile(
     r"provider_discovery_failed provider=(?P<provider>[a-z][a-z0-9_]{0,63}) "
@@ -38,7 +42,8 @@
     for pattern in (
         rf"^provider_attempt agent_id={_AGENT_ID} model={_MODEL_ID} attempt=\d+/\d+$",
         rf"^provider_attempt_failed agent_id={_AGENT_ID} model={_MODEL_ID} attempt=\d+ "
-        rf"error_type={_ERROR_TYPE} transient=(?:True|False)(?= error_message=)",
+        rf"error_type={_ERROR_TYPE} transient=(?:True|False)"
+        rf"(?: provider_status=(?:[1-5][0-9]{{2}}|None))?(?= error_message=)",
         rf"^provider_backoff agent_id={_AGENT_ID} attempt=\d+ delay_seconds={_NUMBER}$",
         rf"^provider_exhausted agent_id={_AGENT_ID} model={_MODEL_ID} attempts=\d+ "
         rf"final_error_type={_ERROR_TYPE}$",
@@ -122,12 +127,18 @@ def _sanitize_orchestrator_event(stripped: str) -> str | None:
 def sanitize_line(line: str) -> str | None:
     """Return one allowlisted diagnostic summary or ``None`` for raw content."""
     stripped = line.strip()
-    request_failed = _REQUEST_FAILED.search(stripped)
+    if "\n" in stripped or "\r" in stripped:
+        return None
+    request_failed = _REQUEST_FAILED.match(stripped)
     if request_failed is not None:
-        return (
+        summary = (
             f"request_failed status={request_failed.group('status')} "
             f"code={request_failed.group('code')}"
         )
+        request_id = request_failed.group("request_id")
+        if request_id is not None:
+            summary += f" request_id={request_id}"
+        return summary
     provider_discovery_failed = _PROVIDER_DISCOVERY_FAILED.search(stripped)
     if provider_discovery_failed is not None:
         return (
diff --git a/tests/test_contextual_orchestrator_review_runtime_preflight.py b/tests/test_contextual_orchestrator_review_runtime_preflight.py
index 7f92162097..73899f10c9 100644
--- a/tests/test_contextual_orchestrator_review_runtime_preflight.py
+++ b/tests/test_contextual_orchestrator_review_runtime_preflight.py
@@ -1800,6 +1800,61 @@ def test_sidecar_stream_sanitizer_admits_orchestrator_route_events() -> None:
     assert sanitize_line(f"DEBUG:contextual_orchestrator.orchestrator:{secret}") is None
 
 
+@pytest.mark.parametrize("status", [None, "100", "429", "599", "None"])
+def test_sidecar_stream_provider_status_compatibility(status) -> None:
+    """Legacy and typed producer diagnostics survive without upstream text."""
+    prefix = "provider_attempt_failed agent_id=fixture model=m/x attempt=1 error_type=HTTPError transient=True"
+    fields = "" if status is None else f" provider_status={status}"
+    assert _load_sanitizer()["sanitize_line"](
+        prefix + fields + " error_message=Bearer sk-secret"
+    ) == prefix + fields + " error_message=<omitted>"
+
+
+@pytest.mark.parametrize("request_id", [None, "a1" * 16, "<omitted>"])
+def test_sidecar_stream_request_id_compatibility(request_id) -> None:
+    """Keep safe correlation identifiers, including the producer omission marker."""
+    message = "request_failed status=500 code=internal_error"
+    if request_id is not None:
+        message += f" request_id={request_id}"
+    assert _load_sanitizer()["sanitize_line"](message) == message
+
+
+@pytest.mark.parametrize("status", ["099", "600", "4290", "429secret", "-1", "True", "none", "４２９"])
+def test_sidecar_stream_rejects_invalid_provider_status(status) -> None:
+    """Invalid typed fields must not downgrade to an accepted legacy prefix."""
+    assert _load_sanitizer()["sanitize_line"](
+        "provider_attempt_failed agent_id=fixture model=m/x attempt=1 "
+        f"error_type=HTTPError transient=True provider_status={status} error_message=sk-secret"
+    ) is None
+
+
+@pytest.mark.parametrize("request_id", ["a" * 31, "a" * 33, "A" * 32, "g" * 32,
+                                      "<omitted>secret", "a" * 32 + "-secret", "", "a" * 32 + "\nsecret"])
+def test_sidecar_stream_rejects_invalid_request_id(request_id) -> None:
+    """Do not preserve a partial identifier or fall back to the legacy record."""
+    assert _load_sanitizer()["sanitize_line"](
+        f"request_failed status=500 code=internal_error request_id={request_id}"
+    ) is None
+
+
+@pytest.mark.parametrize("status,code", [("5000", "internal_error"), ("600", "internal_error"),
+                                       ("500", "x" * 65), ("500", "internal_error/secret")])
+def test_sidecar_stream_rejects_partial_request_fields(status, code) -> None:
+    """Status and code validation consumes complete tokens, never safe prefixes."""
+    assert _load_sanitizer()["sanitize_line"](f"request_failed status={status} code={code}") is None
+
+
+@pytest.mark.parametrize("prefix,allowed", [
+    ("", True), ("WARNING:contextual_orchestrator.server:", True),
+    ("2026-09-05 21:40:00,123 WARNING contextual_orchestrator.server ", True),
+    ("provider text ", False), ("WARNING:provider.raw:", False),
+])
+def test_sidecar_stream_request_event_boundary(prefix, allowed) -> None:
+    """Only bare events or the server logger envelope may carry request IDs."""
+    event = "request_failed status=500 code=internal_error request_id=" + "a" * 32
+    assert _load_sanitizer()["sanitize_line"](prefix + event) == (event if allowed else None)
+
+
 def test_sidecar_stream_sanitizer_matches_real_formatter_output() -> None:
     """Fixtures typed from a template miss runtime value types; render the real records.