chunk-invariance

chunk-invariance

ストリーミングフィルター（レダクター、ガードレール、書き換え器）が正しく動作するのは、入力をどのようにチャンクに分割しても、ストリーミング出力を連結した結果が、入力全体を一度にフィルタリングした出力と一致する場合に限られます。このパッケージを使えば、このルールをひとつのアサーションで確認できます。

from chunk_invariance import assert_chunk_invariant

def redact(text):                 # your filter on the whole input
    return text.replace("sk-AbCd1234", "[KEY]")

def redact_stream(chunks):        # your filter on a stream: takes chunks, yields chunks
    for chunk in chunks:
        yield redact(chunk)       # redacts each chunk alone

def test_streaming_redaction():
    assert_chunk_invariant(redact_stream, redact, "use sk-AbCd1234 now")

このテストは失敗し、問題が起きる最小の分割位置を示します。

chunk_invariance.core.ChunkInvarianceError: stream filter is not chunk-invariant
  input:    'use sk-AbCd1234 now'
  split:    ['use s', 'k-AbCd1234 now'] (cut at 5)
  whole:    'use [KEY] now'
  streamed: 'use sk-AbCd1234 now'
  split #6 in the order tried; generated splits run fewest parts first

pip install chunk-invariance
pip install "chunk-invariance[hypothesis]"   # optional Hypothesis strategies
pip install "chunk-invariance[mcp]"          # optional MCP server for AI assistants

検出できる問題

• 各チャンクを個別に評価するフィルター。値が2つのチャンクにまたがって分割されると、どちらの部分にも一致しなくなります。
• 前のチャンクから一定量を持ち越して先頭に付けるフィルター。一致はしますが、値の先頭はすでに前のチャンクとともに出力されています。
• 保留する長さがパターンより短い、ストリーム終端でのフラッシュがない、出力が漏えいではなく破損しているといった問題（入力全体を処理した場合と少しでも異なれば失敗します）。

chunk_invariance.examplesには、同じパターンに対する正しいフィルター（hold_back）と、誤った設計（design_a、design_b）の2つが比較例として用意されています。

検出できない問題

このテストの対象は、プロセス内で呼び出せるフィルターです。上流のモデル、ネットワーク、SSE パーサーによってチャンクが決まる、実行中のゲートウェイはテストしません。そのようなケースには、フラグメント化したレスポンスをゲートウェイ経由で送信し、クライアントに届く内容を確認するpii-leak-benchmarkを使用してください。

テストするのは、指定した分割方法です。デフォルトでは、入力全体と、すべての1か所での分割を試します。これにより、値が2つに分割されたケースを検出できます。splits="all"は短い入力（最大16文字）のすべての分割を試し、splits=3は最大3つのパートに分割するすべての方法を試します。チャンクのリストを明示的に指定すると、その分割だけを実行します。指定していない入力に対してフィルターが正しいことまでは証明できません。

AI アシスタントから使用する

Claude Code、Cursor、Claude Desktopをお使いですか？このパッケージにはオプションの MCP サーバーがあります。これを使うと、アシスタントに独自のストリーミング関数で全分割チェックを実行させ、そのテストをリポジトリに追加させることができます。

pip install "chunk-invariance[mcp]"      # Python 3.10 or later
claude mcp add chunk-invariance -- chunk-invariance-mcp

Claude Desktopの場合は、次の内容をclaude_desktop_config.jsonに追加してください（Claude Desktopがコマンドを PATH から見つけられない場合は、コマンドのフルパスを指定してください）。

{
  "mcpServers": {
    "chunk-invariance": { "command": "chunk-invariance-mcp" }
  }
}

次のように依頼できます。「myapp/filters.pyのredact_streamが、チャンク間で分割された値を漏えいするか確認し、回帰テストを追加してください。」

ツールは3つあります。check_stream_filterは関数（myapp.filters:redact_stream）をインポートし、各サンプルを2分割のすべての方法でストリーミングして、どの分割で問題が起きたかを伝えます。たとえば、入力が'use s'と'k-AbCd1234 now'に分割されたとき、'sk-AbCd1234'が伏せられずに出力された、といった内容です。write_regression_testはassert_chunk_invariantを使ったpytestファイルを返すため、CIでもチェックを継続できます。explainは、チャンク単位のフィルタリングで値が漏えいする理由を1段落で説明します。

デフォルトのサンプル（メールアドレス、SSN、カード番号、いくつかの形式のAPIキー）はすべて架空のものです。フィルターが伏せる形式の、自分で作成したサンプルを渡してください。たとえば、プロバイダーの形式に合わせて自分で入力したキーなどです。フィルターが変更しないサンプルでは何もテストできず、その場合はツールがそのことを通知します。

サンプルに実際の秘密情報や個人データを絶対に使用しないでください。サンプルと、フィルターから漏えいした値はツールの結果に含まれ、アシスタントは会話の他の部分と同様に、その結果をモデルプロバイダーに送信します。コードが例外を発生させた場合は、例外メッセージも送信されますが、コードが出力する内容は送信されません。同じサンプルは生成されるテストファイルにも含まれます。

関数をチェックすると、あなたのコードが実行されます。サーバーはあなたのマシン上でモジュールを別プロセスとしてインポートし、タイムアウト後にそのプロセスを停止します（デフォルトは60秒です）。プロジェクトに.venvがある場合は、そのPythonを使用します。サーバー自体はネットワークに接続せず、テレメトリーを送信せず、ファイルも書き込みません。テストファイルの保存は、あなたとアシスタントが行います。

API

名前	説明
assert_chunk_invariant(stream_filter, whole_filter, text, *, splits="all-two-part")	最初に失敗した分割で ChunkInvarianceError（AssertionError のサブクラス）を送出します。分割数の少ないものから確認します。確認した分割数を返します。str と bytes に対応しています。
assert_chunk_invariant_async(...)	非同期ジェネレーターフィルター向けの同じ関数です。非同期テスト内で使用します。
two_part_splits(text)	空でない2つの部分へのすべての分割です。
all_splits(text, max_parts=None)	空でない部分への、最大 max_parts 個までのすべての分割です。入力全体が最初に含まれます。
per_chunk(process, *, state=dict, flush=None)	process(chunk, state) -> output と flush(state) -> output を、ストリームごとに新しい状態を使う形に適合させます。
from_async(async_stream_filter)	同期アサーション用に、非同期ジェネレーターフィルターを適合させます。
strategies.splits_of(text, max_parts=None)	Hypothesis のストラテジーです。分割数が少なくなる方向に縮小される、text のランダムな分割を生成します。
strategies.text_and_splits(texts, max_parts=None)	(text, chunks) のペアを生成する Hypothesis のストラテジーです。texts はリストではなく、st.sampled_from([...]) のようなストラテジーそのものです。

stream_filter は分割ごとに1回呼び出されるため、呼び出しのたびに新しい状態から開始する必要があります。

背景

この不変条件と、誤った2つの設計については、Phalak, N. (2026), Split-Boundary Leaks in Streaming Guardrails, Zenodo, 10.5281/zenodo.22909585 で説明しています。

標準ライブラリのみを使用します（mcp extra を追加すると、サーバー専用の MCP SDK が加わります）。Apache-2.0。API が同じ TypeScript 版は npm で chunk-invariance として公開されています。