diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
index 0df8db9..907fdfd 100644
--- a/.github/workflows/ci.yaml
+++ b/.github/workflows/ci.yaml
@@ -77,7 +77,7 @@ jobs:
       - name: Setup Python
         uses: actions/setup-python@v5
         with:
-          python-version: '3.11'
+          python-version: '3.12'
       - name: Install ext_proc and test dependencies
         working-directory: kaito-guard-ext-proc
         run: python -m pip install -e '.[dev]'
diff --git a/docker/Dockerfile.kaito-guard-ext-proc b/docker/Dockerfile.kaito-guard-ext-proc
index 36cc12b..117d462 100644
--- a/docker/Dockerfile.kaito-guard-ext-proc
+++ b/docker/Dockerfile.kaito-guard-ext-proc
@@ -1,10 +1,11 @@
-FROM python:3.11-slim
+FROM python:3.12.13-slim-bookworm@sha256:4766d8b510c428e595d74b9cc5bbb2fae8e26316fffb4adc89908d79aacd58a2
 
 WORKDIR /app
 
 COPY kaito-guard-ext-proc/setup.py .
 COPY kaito-guard-ext-proc/ext_proc/ ./ext_proc/
 RUN pip install --no-cache-dir .
+RUN python -c "from guardrail_core.streaming_window import StreamingBufferWindow, WindowScanner, WindowScanResult, WindowEmitResult, WindowScanError; from ext_proc.streaming_window import StreamingResponseWindow"
 
 EXPOSE 9000
 
diff --git a/kaito-guard-ext-proc/README.md b/kaito-guard-ext-proc/README.md
index 157aac5..598a0fd 100644
--- a/kaito-guard-ext-proc/README.md
+++ b/kaito-guard-ext-proc/README.md
@@ -3,17 +3,17 @@
 This package is the Envoy gateway/streaming adapter, not a guardrail
 implementation. Guardrail logic belongs to the shared KAITO guardrail core,
 as described in [kaito-project/production-stack#226](https://github.com/kaito-project/production-stack/issues/226).
-Core integration is deferred until its dependency and API are pinned; no local
-scanners or guard engine are provided here.
+The response-window adapter uses KAITO's shared streaming window; no local
+scanners, holdback algorithm, or guard engine are provided here.
 
 The server registers Envoy's `envoy.service.ext_proc.v3.ExternalProcessor`
 service and listens on `[::]:9000`. `Process` returns gRPC `UNIMPLEMENTED`
-until the shared core is integrated. This skeleton does not inspect or protect
+until the adapter is wired into the service. This skeleton does not inspect or protect
 traffic and must not be used as an active guardrail.
 
 ## Run and test
 
-Requires Python 3.11 or newer. From the repository root:
+Requires Python 3.12 or newer, matching the shared core's minimum. From the repository root:
 
 ```sh
 cd kaito-guard-ext-proc
@@ -28,10 +28,71 @@ python -m pytest
 Runtime dependencies are defined only in `setup.py`. `xds-protos`, published
 by the gRPC project, supplies the generated Envoy protobuf messages and service.
 
+## Shared response window
+
+`guardrail-core` is installed from the authoritative
+[KAITO commit `dda1b10e881ed69228978964a2e89ecf241757f9`](https://github.com/kaito-project/kaito/commit/dda1b10e881ed69228978964a2e89ecf241757f9),
+using its immutable source archive and `guardrail_core` subdirectory. It does not
+follow `main`. The installed package exposes `StreamingBufferWindow`,
+`WindowScanner`, `WindowScanResult`, `WindowEmitResult`, and `WindowScanError`
+from `guardrail_core.streaming_window`.
+
+[`StreamingResponseWindow`](ext_proc/streaming_window.py) only manages lifecycle.
+Create one instance per response, supplying a scanner and a holdback length.
+Holdback counts Python string Unicode characters, not UTF-8 bytes. Scanner
+construction, policy ordering, and sufficient holdback remain caller responsibilities;
+stateful scanners must also be scoped to the response.
+
+```python
+from guardrail_core.streaming_window import WindowScanner
+
+from ext_proc.streaming_window import StreamingResponseWindow
+
+
+def checked_chunks(scanner: WindowScanner, holdback_len: int, upstream_text):
+    response = StreamingResponseWindow(scanner, holdback_len=holdback_len)
+    try:
+        for text in upstream_text:
+            result = response.feed(text)
+            if result.blocked:
+                raise RuntimeError("Response blocked")
+            yield from result.chunks
+        result = response.finish()
+        if result.blocked:
+            raise RuntimeError("Response blocked")
+        yield from result.chunks
+    finally:
+        response.abort()
+```
+
+Only an upstream that verifies complete, successful termination may reach
+`finish()`. Parse errors, truncation, and cancellation must instead call `abort()`
+without flushing. An iterator that silently treats truncation as EOF is not a
+suitable input. The example illustrates text-window usage, not an SSE parser
+or Envoy output adapter.
+
+- `feed()` and `finish()` return the shared `WindowEmitResult` unchanged.
+- `blocked` and `redacted` expose shared-window state; `closed` is lifecycle state.
+- A block result is terminal and must enter the caller's explicit rejection path.
+- Scanner errors propagate as `WindowScanError` with their original cause;
+  exceptions and cancellation make the response terminal and must enter the
+  caller's failure path. Unchecked buffered text is never flushed on failure.
+- `finish()` invokes the shared final `flush()` exactly once. Further `feed()`
+  or `finish()` calls after any terminal state raise `RuntimeError`.
+- `abort()` is idempotent and never emits text. Discard the response instance
+  after termination. Neither this adapter nor its error messages log raw text.
+
+This change does **not** integrate `Process`, migrate an SSE/OpenAI parser,
+provide policy/scanner configuration, or establish Envoy end-to-end protection.
+Tests validate the lifecycle adapter against direct calls to the pinned shared
+window, not live gateway traffic.
+
 ## CI
 
 The `kaito-guard-ext-proc` job in the repository's CI workflow installs
-`.[dev]` with Python 3.11, runs pytest, and builds the container image.
+`.[dev]` with Python 3.12, runs pytest, and builds the container image.
+The image build also imports the shared window, its interfaces/results, and the
+adapter to verify that the runtime installation contains them.
 The image is not published, and this job does not test real Envoy integration.
 
 ## Container
diff --git a/kaito-guard-ext-proc/ext_proc/streaming_window.py b/kaito-guard-ext-proc/ext_proc/streaming_window.py
new file mode 100644
index 0000000..f18aa38
--- /dev/null
+++ b/kaito-guard-ext-proc/ext_proc/streaming_window.py
@@ -0,0 +1,66 @@
+# Copyright (c) KAITO authors.
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Response lifecycle only; buffering and safe release belong to KAITO core."""
+
+from guardrail_core.streaming_window import (
+    StreamingBufferWindow,
+    WindowEmitResult,
+    WindowScanner,
+)
+
+
+class StreamingResponseWindow:
+    """Own one shared window, with terminal finish, abort, block, and failure."""
+
+    def __init__(self, scanner: WindowScanner, *, holdback_len: int) -> None:
+        self._window = StreamingBufferWindow(scanner, holdback_len=holdback_len)
+        self._closed = False
+
+    @property
+    def closed(self) -> bool:
+        return self._closed
+
+    @property
+    def blocked(self) -> bool:
+        return self._window.blocked
+
+    @property
+    def redacted(self) -> bool:
+        return self._window.redacted
+
+    def feed(self, text: str) -> WindowEmitResult:
+        self._require_open()
+        try:
+            result = self._window.feed(text)
+        except BaseException:
+            # Cancellation must be terminal too; preserve the original failure.
+            self.abort()
+            raise
+        if result.blocked:
+            self.abort()
+        return result
+
+    def finish(self) -> WindowEmitResult:
+        """Flush once, only after successful completion of the upstream stream."""
+        self._require_open()
+        self._closed = True
+        return self._window.flush()
+
+    def abort(self) -> None:
+        """Terminate without scanning or releasing the retained tail."""
+        self._closed = True
+
+    def _require_open(self) -> None:
+        if self._closed:
+            raise RuntimeError("Streaming response window is closed.")
diff --git a/kaito-guard-ext-proc/setup.py b/kaito-guard-ext-proc/setup.py
index 2c2bc65..3a0d652 100644
--- a/kaito-guard-ext-proc/setup.py
+++ b/kaito-guard-ext-proc/setup.py
@@ -6,8 +6,10 @@
     description="Python ext_proc for KAITO guardrailing in Envoy",
     author="KAITO Team",
     packages=find_packages(include=["ext_proc", "ext_proc.*"]),
-    python_requires=">=3.11",
+    python_requires=">=3.12",
     install_requires=[
+        "guardrail-core @ https://github.com/kaito-project/kaito/archive/"
+        "dda1b10e881ed69228978964a2e89ecf241757f9.tar.gz#subdirectory=guardrail_core",
         "grpcio==1.84.0",
         "xds-protos==1.84.0",
     ],
diff --git a/kaito-guard-ext-proc/tests/test_streaming_window.py b/kaito-guard-ext-proc/tests/test_streaming_window.py
new file mode 100644
index 0000000..8829e1d
--- /dev/null
+++ b/kaito-guard-ext-proc/tests/test_streaming_window.py
@@ -0,0 +1,255 @@
+# Copyright (c) KAITO authors.
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import asyncio
+from collections.abc import Iterator
+from itertools import product
+
+import pytest
+from guardrail_core.streaming_window import (
+    StreamingBufferWindow,
+    WindowEmitResult,
+    WindowScanError,
+    WindowScanResult,
+)
+
+from ext_proc.streaming_window import StreamingResponseWindow
+
+
+class TestScanner:
+    __test__ = False
+
+    def __init__(
+        self,
+        *,
+        block: bool = False,
+        error: BaseException | None = None,
+        fail_on_flush: bool = False,
+    ) -> None:
+        self.block = block
+        self.error = error
+        self.fail_on_flush = fail_on_flush
+        self.calls: list[tuple[str, bool, str]] = []
+
+    def scan(
+        self, text: str, *, flush: bool = False, preceding_char: str = ""
+    ) -> WindowScanResult:
+        self.calls.append((text, flush, preceding_char))
+        if self.error is not None and (flush or not self.fail_on_flush):
+            raise self.error
+        if "secret" in text:
+            if self.block:
+                return WindowScanResult(blocked=True)
+            return WindowScanResult(sanitized_text=text.replace("secret", "[X]"))
+        return WindowScanResult()
+
+
+def partitions(text: str) -> Iterator[tuple[str, ...]]:
+    for cuts in product((False, True), repeat=len(text) - 1):
+        start = 0
+        parts = []
+        for index, cut in enumerate(cuts, start=1):
+            if cut:
+                parts.append(text[start:index])
+                start = index
+        parts.append(text[start:])
+        yield tuple(parts)
+
+
+@pytest.mark.parametrize("parts", list(partitions("secret")))
+@pytest.mark.parametrize("block", [False, True])
+def test_cross_feed_match_and_shared_window_equivalence(parts, block):
+    scanner = TestScanner(block=block)
+    direct_scanner = TestScanner(block=block)
+    response = StreamingResponseWindow(scanner, holdback_len=6)
+    direct = StreamingBufferWindow(direct_scanner, holdback_len=6)
+    emitted = []
+    for text in ("ok ", *parts, "!"):
+        result = response.feed(text)
+        assert result == direct.feed(text)
+        assert isinstance(result, WindowEmitResult)
+        emitted.extend(result.chunks)
+        assert response.redacted == direct.redacted
+        assert response.blocked == direct.blocked
+        if result.blocked:
+            assert response.closed
+            break
+    else:
+        result = response.finish()
+        assert result == direct.flush()
+        emitted.extend(result.chunks)
+
+    assert scanner.calls == direct_scanner.calls
+    if block:
+        assert "secret" not in "".join(emitted)
+        assert "ok ".startswith("".join(emitted))
+    else:
+        assert "".join(emitted) == "ok [X]!"
+        assert response.redacted
+    assert response.closed
+
+
+@pytest.mark.parametrize("holdback", [0, 1, 5, 100])
+@pytest.mark.parametrize("parts", [("",), ("a", "bc", "def"), ("你", "好", "世界")])
+def test_normal_and_unicode_input_match_direct_window(parts, holdback):
+    scanner = TestScanner()
+    direct_scanner = TestScanner()
+    response = StreamingResponseWindow(scanner, holdback_len=holdback)
+    direct = StreamingBufferWindow(direct_scanner, holdback_len=holdback)
+    emitted = []
+    for text in parts:
+        result = response.feed(text)
+        assert result == direct.feed(text)
+        emitted.extend(result.chunks)
+    result = response.finish()
+    assert result == direct.flush()
+    emitted.extend(result.chunks)
+    assert "".join(emitted) == "".join(parts)
+    assert scanner.calls == direct_scanner.calls
+    assert response.closed
+    assert not response.redacted
+    assert not response.blocked
+
+
+def test_preceding_character_is_managed_by_shared_window():
+    scanner = TestScanner()
+    response = StreamingResponseWindow(scanner, holdback_len=2)
+    assert response.feed("abc").chunks == ("a",)
+    assert response.feed("de").chunks == ("bc",)
+    assert response.finish().chunks == ("de",)
+    assert scanner.calls == [
+        ("abc", False, ""),
+        ("bcde", False, "a"),
+        ("de", True, "c"),
+    ]
+
+
+@pytest.mark.parametrize("block", [False, True])
+def test_final_scan_can_redact_or_block(block):
+    scanner = TestScanner(block=block)
+    response = StreamingResponseWindow(scanner, holdback_len=100)
+    assert response.feed("secret").chunks == ()
+    assert scanner.calls == []
+    result = response.finish()
+    assert result == WindowEmitResult(chunks=() if block else ("[X]",), blocked=block)
+    assert response.blocked == block
+    assert response.redacted == (not block)
+    assert response.closed
+    assert scanner.calls == [("secret", True, "")]
+
+
+def test_finish_is_terminal_and_never_flushes_twice():
+    scanner = TestScanner()
+    response = StreamingResponseWindow(scanner, holdback_len=100)
+    response.feed("retained")
+    assert response.finish().chunks == ("retained",)
+    with pytest.raises(RuntimeError, match="closed"):
+        response.finish()
+    with pytest.raises(RuntimeError, match="closed"):
+        response.feed("later")
+    response.abort()
+    response.abort()
+    assert scanner.calls == [("retained", True, "")]
+
+
+@pytest.mark.parametrize(
+    "upstream_error", [ValueError("parse error"), EOFError(), asyncio.CancelledError()]
+)
+def test_upstream_failure_aborts_without_scanning_or_releasing_tail(upstream_error):
+    scanner = TestScanner()
+    response = StreamingResponseWindow(scanner, holdback_len=100)
+    assert response.feed("unchecked tail").chunks == ()
+    with pytest.raises(type(upstream_error)):
+        try:
+            raise upstream_error
+        finally:
+            response.abort()
+    response.abort()
+    assert response.closed
+    with pytest.raises(RuntimeError, match="closed"):
+        response.feed("later")
+    with pytest.raises(RuntimeError, match="closed"):
+        response.finish()
+    assert scanner.calls == []
+
+
+@pytest.mark.parametrize("flush", [False, True])
+@pytest.mark.parametrize("cancel", [False, True])
+def test_scanner_failure_or_cancellation_is_terminal(flush, cancel, caplog):
+    error = asyncio.CancelledError() if cancel else ValueError("private scanner text")
+    scanner = TestScanner(error=error, fail_on_flush=flush)
+    response = StreamingResponseWindow(scanner, holdback_len=4)
+    assert response.feed("tail").chunks == ()
+    expected = asyncio.CancelledError if cancel else WindowScanError
+    with pytest.raises(expected) as caught:
+        if flush:
+            response.finish()
+        else:
+            response.feed("!")
+    if cancel:
+        assert caught.value is error
+    else:
+        assert caught.value.__cause__ is error
+        assert response.blocked
+        assert "private scanner text" not in str(caught.value)
+    assert response.closed
+    calls = list(scanner.calls)
+    with pytest.raises(RuntimeError, match="closed"):
+        response.feed("later")
+    with pytest.raises(RuntimeError, match="closed"):
+        response.finish()
+    assert scanner.calls == calls
+    assert not caplog.records
+
+
+def test_block_prevents_further_input_or_flush():
+    scanner = TestScanner(block=True)
+    response = StreamingResponseWindow(scanner, holdback_len=6)
+    assert response.feed("secret").chunks == ()
+    assert response.feed("!").blocked
+    assert response.blocked
+    assert response.closed
+    with pytest.raises(RuntimeError, match="closed"):
+        response.feed("later")
+    with pytest.raises(RuntimeError, match="closed"):
+        response.finish()
+    assert scanner.calls == [("secret!", False, "")]
+
+
+@pytest.mark.parametrize("termination", ["finish", "abort", "block", "error"])
+def test_response_state_is_not_shared(termination):
+    first_scanner = TestScanner(block=termination == "block")
+    first = StreamingResponseWindow(first_scanner, holdback_len=6)
+    second_scanner = TestScanner()
+    second = StreamingResponseWindow(second_scanner, holdback_len=6)
+    first.feed("sec")
+    second.feed("ret")
+    if termination == "finish":
+        assert first.finish().chunks == ("sec",)
+    elif termination == "abort":
+        first.abort()
+    elif termination == "block":
+        assert first.feed("ret!").blocked
+    else:
+        first_scanner.error = ValueError("scan failure")
+        with pytest.raises(WindowScanError):
+            first.feed("ret!")
+    assert second.finish().chunks == ("ret",)
+    assert not second.blocked
+    assert not second.redacted
+    assert second_scanner.calls == [("ret", True, "")]
+
+
+def test_invalid_holdback_uses_shared_validation():
+    with pytest.raises(ValueError, match="non-negative"):
+        StreamingResponseWindow(TestScanner(), holdback_len=-1)