Output filters see each streamed chunk alone, so a value split across two SSE events passes output_redactor #1036

#!/usr/bin/env bash
# Output filter vs a value split across two SSE events.
# Run from demos/filter_chains/model_listener_filter/ in a checkout of katanemo/plano.
# Needs: docker, python3 with fastapi + uvicorn. No API keys.
set -u
IMAGE="${PLANO_IMAGE:-katanemo/plano:0.4.36}"
WORK="$(mktemp -d)"
cp output_filter.py "$WORK/"

cat > "$WORK/split_upstream.py" <<'PY'
import asyncio, json, time
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse

app = FastAPI()
HEAD, TAIL = "The token is SECRET_", "TOKEN."

def ev(content):
    return "data: " + json.dumps({"id": "x", "object": "chat.completion.chunk", "created": int(time.time()),
        "model": "gpt-4o-mini", "choices": [{"index": 0, "delta": {"content": content}, "finish_reason": None}]}) + "\n\n"

@app.post("/v1/chat/completions")
async def completions(req: Request):
    body = await req.json()
    split = body["messages"][-1]["content"] == "split"
    async def gen():
        if split:
            yield ev(HEAD); await asyncio.sleep(0.5)
            yield ev(TAIL); await asyncio.sleep(0.5)
        else:
            yield ev(HEAD + TAIL); await asyncio.sleep(0.5)
        yield "data: [DONE]\n\n"
    return StreamingResponse(gen(), media_type="text/event-stream")
PY

cat > "$WORK/plano_config.yaml" <<'YAML'
version: v0.3.0
filters:
  - id: output_redactor
    url: http://host.docker.internal:10502
    type: http
model_providers:
  - model: openai/gpt-4o-mini
    access_key: local-demo-key
    base_url: http://host.docker.internal:8765/v1
    default: true
listeners:
  - type: model
    name: llm_gateway
    port: 12000
    output_filters:
      - output_redactor
YAML

(cd "$WORK" && python3 -m uvicorn split_upstream:app --host 0.0.0.0 --port 8765 >/dev/null 2>&1 & echo $! > "$WORK/up.pid")
(cd "$WORK" && python3 -m uvicorn output_filter:app --host 0.0.0.0 --port 10502 >/dev/null 2>&1 & echo $! > "$WORK/filter.pid")
docker run -d --name plano-split-repro -p 12000:12000 \
  -v "$WORK/plano_config.yaml:/app/plano_config.yaml:ro" \
  --add-host host.docker.internal:host-gateway "$IMAGE" >/dev/null
for _ in $(seq 1 60); do curl -s -o /dev/null localhost:12000/v1/models && break; sleep 1; done
sleep 2

ask() {
  curl -sN http://localhost:12000/v1/chat/completions -H 'Content-Type: application/json' \
    -d "{\"model\":\"gpt-4o-mini\",\"messages\":[{\"role\":\"user\",\"content\":\"$1\"}],\"stream\":true}" |
  python3 -c '
import sys, json
text = ""
for line in sys.stdin:
    line = line.strip()
    if line.startswith("data:") and "[DONE]" not in line:
        text += json.loads(line[5:])["choices"][0]["delta"].get("content") or ""
print(text)'
}

echo "marker in one SSE event : $(ask single)"
echo "marker split across two : $(ask split)"

docker rm -f plano-split-repro >/dev/null
kill "$(cat "$WORK/up.pid")" "$(cat "$WORK/filter.pid")" 2>/dev/null