From 398c673885bd94e0c5f047240ae3289a04f72754 Mon Sep 17 00:00:00 2001 From: Attila Laszlo Nagy Date: Wed, 16 Sep 2026 17:01:48 +0200 Subject: [PATCH 1/3] fix: preserve generation progress and reject invalid guard verdicts --- packages/sie_sdk/README.md | 29 ++++ packages/sie_sdk/src/sie_sdk/client/sync.py | 12 +- .../sie_server/adapters/_generation_base.py | 18 +- .../sie_server/adapters/sglang/generation.py | 155 ++++++++---------- .../sie_server/src/sie_server/api/generate.py | 25 ++- .../src/sie_server/processors/streaming.py | 17 +- .../sie_server/processors/tool_call_parser.py | 3 + .../tests/adapters/test_generation_base.py | 15 +- .../tests/adapters/test_sglang_generation.py | 108 ++++++++++-- .../sie_server/tests/api/test_generate.py | 10 ++ .../tests/processors/test_streaming.py | 66 +++++++- packages/sie_ts_sdk/README.md | 15 ++ packages/sie_ts_sdk/src/client.ts | 5 + 13 files changed, 333 insertions(+), 145 deletions(-) diff --git a/packages/sie_sdk/README.md b/packages/sie_sdk/README.md index 304caeaff..f640daa9f 100644 --- a/packages/sie_sdk/README.md +++ b/packages/sie_sdk/README.md @@ -34,6 +34,35 @@ for entry in scores["scores"]: print(entry["item_id"], entry["score"]) ``` +## Generation prompts and guard verdicts + +`generate` and `stream_generate` treat text-only prompts as raw continuation +input. They do not render a chat template, including model settings such as +`enable_thinking` or `guardian_config`. Already-rendered prompts stay unchanged. +Native requests with images render the prompt and images as one user turn. + +Use `chat_completions` or `stream_chat_completions` with messages for chat, +instruction-based structured output (`response_format`), and guard checks: + +```python +answer = client.chat_completions( + "Qwen/Qwen3-4B-Instruct-2507", + [{"role": "user", "content": "Write a haiku about the sea."}], + max_completion_tokens=64, +) +print(answer["choices"][0]["message"]["content"]) +``` + +The worker renders the selected model's template and applies served template +settings with operator configuration taking precedence over request kwargs. +Granite Guardian's shipped risk dimension is `harm`; prose requesting a +different dimension does not change that setting. Its configured threshold +produces `Yes` (unsafe) or `No` (safe). A missing or invalid verdict returns +`invalid_guard_verdict`; never treat an error or an empty response as safe. +Private reasoning is hidden on both input surfaces. If it consumes the entire +generation budget without usable output, the request fails with +`empty_model_output`. + ## Connecting to a managed SIE platform The examples above target a local server. For a managed SIE gateway, diff --git a/packages/sie_sdk/src/sie_sdk/client/sync.py b/packages/sie_sdk/src/sie_sdk/client/sync.py index d661064cd..23613f9bf 100644 --- a/packages/sie_sdk/src/sie_sdk/client/sync.py +++ b/packages/sie_sdk/src/sie_sdk/client/sync.py @@ -2360,10 +2360,11 @@ def generate( ``"Qwen/Qwen3-4B-Instruct-2507"`` are normalized to the gateway's SIE-safe path id ``"Qwen__Qwen3-4B-Instruct-2507"`` for this endpoint. - prompt: Raw prompt string. Chat-template rendering, if any, - is performed by the worker — this surface has no chat-template - helpers in the SDK (use the OpenAI SDK against - ``/v1/chat/completions`` for chat-shaped requests). + prompt: Raw continuation input for text-only requests, passed + unchanged without a chat template. Served template settings + such as ``enable_thinking`` and ``guardian_config`` do not + apply to raw input. Use :meth:`chat_completions` with messages + for chat, instruction-based structured output, and guard checks. max_new_tokens: Hard cap on output tokens. images: Optional native image inputs. When present, the worker renders one user turn containing the images and ``prompt`` @@ -3026,6 +3027,9 @@ def stream_generate( These worker-origin digests are distinct from the catalog weights revision and :attr:`last_model_revision`; gateway SSE omits the ``X-SIE-Model-Revision`` header. + + Text-only prompts are raw continuation input, as in :meth:`generate`. + Use :meth:`stream_chat_completions` to apply the served chat template. """ resolved_grammar = validate_generate_grammar(grammar) if grammar is not None else None pool_name, resolved_gpu = self._resolve_pool_and_gpu(gpu) diff --git a/packages/sie_server/src/sie_server/adapters/_generation_base.py b/packages/sie_server/src/sie_server/adapters/_generation_base.py index 035648889..078c1584a 100644 --- a/packages/sie_server/src/sie_server/adapters/_generation_base.py +++ b/packages/sie_server/src/sie_server/adapters/_generation_base.py @@ -105,6 +105,7 @@ class GenerationDrainingError(GenerationCapacityError): _CLIENT_SAFE_GENERATION_ERROR_MESSAGES = { "inference_error": "internal error during generation", "grammar_compile_failed": "internal error compiling grammar", + "invalid_guard_verdict": "guard model did not produce a valid thresholded verdict", } _CLIENT_SAFE_GENERATION_ERROR_CODES = frozenset( { @@ -122,6 +123,7 @@ class GenerationDrainingError(GenerationCapacityError): "empty_model_output", "grammar_invalid", "invalid_request", + "invalid_guard_verdict", "parallel_tool_calls_violated", "rate_limit_exceeded", "tool_call_parse_error", @@ -716,25 +718,11 @@ async def suppress_thinking_blocks( reason = " (private reasoning consumed the generation budget)" if hid_reasoning else "" rewritten = replace( rewritten, + finish_reason="error", error_code="empty_model_output", error_message=f"model produced no visible output text{reason}", ) - # Reasoning-only engine deltas have no wire-visible information. - # Keep terminals, per-choice finish markers, tool/error chunks, and - # non-streaming candidate aggregates intact. - if ( - not rewritten.text_delta - and not rewritten.done - and rewritten.finish_reason is None - and rewritten.tool_call_delta is None - and rewritten.error_code is None - and rewritten.error_message is None - and rewritten.logprobs is None - and rewritten.candidates is None - ): - continue - if rewritten.done: terminal_outcome_selected = True yield rewritten diff --git a/packages/sie_server/src/sie_server/adapters/sglang/generation.py b/packages/sie_server/src/sie_server/adapters/sglang/generation.py index 426797c9c..16bbe4a63 100644 --- a/packages/sie_server/src/sie_server/adapters/sglang/generation.py +++ b/packages/sie_server/src/sie_server/adapters/sglang/generation.py @@ -1548,20 +1548,8 @@ async def generate( # so we slice off the tail-since-last-event each # round to build per-chunk OpenAI-shape logprobs. logprobs_surfaced = 0 - # Guard verdict buffering (CHECK POLICY) — inert for non-guard - # models. A guard's Yes/No verdict can sit a few token positions - # in, behind a leading whitespace/punctuation/preamble token that - # SGLang spreads across streaming chunks. We accumulate the - # leading chunks' per-token logprob entries (``guard_lp_buffer``) - # and SUPPRESS their text (the guard consumer wants just the - # verdict, not the preamble) until a verdict resolves within the - # first ``_GUARD_VERDICT_SCAN_POSITIONS`` positions — or the - # stream terminates first, in which case we flush the raw buffered - # chunks unchanged (fallback, never drop output). guard_active = bool(self._guard) guard_resolved = False - guard_lp_buffer: list[dict[str, Any]] = [] - guard_pending: list[GenerationChunk] = [] async for raw_line in response.aiter_lines(): line = raw_line.strip() @@ -1608,69 +1596,40 @@ async def generate( cumulative_lp = event_meta.get("output_token_logprobs") if isinstance(cumulative_lp, list): logprobs_surfaced = max(logprobs_surfaced, len(cumulative_lp)) - # Guard verdict thresholding (CHECK POLICY): resolve the - # P(unsafe)>=threshold verdict from the first parseable - # position within ``_GUARD_VERDICT_SCAN_POSITIONS`` and emit a - # single verdict chunk. Inert for non-guard models, which take - # the byte-for-byte unchanged ``else`` path below. - if guard_active and not guard_resolved: - # Accumulate this chunk's forced logprob entries so a - # verdict that lands on a later token position is visible. - if chunk.logprobs: - guard_lp_buffer.extend(chunk.logprobs) - guard_pending.append(chunk) - verdict = _thresholded_verdict(tuple(guard_lp_buffer), self._guard) - if verdict is not None: - guard_resolved = True - # Carry through terminal state if the verdict resolved - # on (or only by) the terminal chunk, so done / - # finish_reason / completion_tokens are preserved. - last = guard_pending[-1] - # Strip the internally-forced logprobs when the client - # did not ask for them (implementation detail). When - # the client did ask, drop the single verdict entry - # that was consumed/rewritten (it described the raw - # sampled token, not the served threshold verdict) and - # keep the rest of the buffered token metadata. - if client_requested_logprobs: - v_idx = _verdict_position(tuple(guard_lp_buffer)) - kept = [e for i, e in enumerate(guard_lp_buffer) if i != v_idx] - remaining = tuple(kept) or None + if guard_active: + if chunk.error_code is not None or chunk.finish_reason in ("error", "cancelled"): + chunk = dataclasses.replace(chunk, text_delta="", is_first=False, logprobs=None) + elif not guard_resolved: + guard_lp_buffer = _guard_verdict_logprobs(event) + verdict = _thresholded_verdict(guard_lp_buffer, self._guard) + if verdict is not None: + guard_resolved = True + if client_requested_logprobs: + v_idx = _verdict_position(tuple(guard_lp_buffer)) + remaining = tuple(e for i, e in enumerate(guard_lp_buffer) if i != v_idx) or None + else: + remaining = None + chunk = dataclasses.replace( + chunk, text_delta=verdict, is_first=True, logprobs=remaining + ) + elif chunk.done: + chunk = dataclasses.replace( + chunk, + text_delta="", + is_first=False, + logprobs=None, + finish_reason="error", + error_code="invalid_guard_verdict", + error_message="guard model did not produce a valid thresholded verdict", + ) else: - remaining = None - verdict_chunk = dataclasses.replace( - last, - text_delta=verdict, - is_first=True, - logprobs=remaining, - ) - stream_timer.mark_yield(has_text=True) - yield verdict_chunk - if verdict_chunk.done: - break - elif chunk.done: - # Terminal reached without a parseable verdict in the - # first N positions: flush the raw buffered chunks - # unchanged so the response is never dropped. Strip the - # forced logprobs only when the client didn't ask. - guard_resolved = True - for buffered in guard_pending: - if not client_requested_logprobs: - buffered = dataclasses.replace(buffered, logprobs=None) - stream_timer.mark_yield(has_text=bool(buffered.text_delta)) - yield buffered - break - # else: keep buffering (suppress this leading chunk's text). - else: - # Guard tail chunks (after the verdict resolved) must still - # honour the M4 logprobs contract: strip the internally - # forced logprobs when the client didn't request them. - if guard_active and not client_requested_logprobs and chunk.logprobs is not None: - chunk = dataclasses.replace(chunk, logprobs=None) - stream_timer.mark_yield(has_text=bool(chunk.text_delta)) - yield chunk - if chunk.done: - break + chunk = dataclasses.replace(chunk, text_delta="", is_first=False, logprobs=None) + else: + chunk = dataclasses.replace(chunk, text_delta="", is_first=False, logprobs=None) + stream_timer.mark_yield(has_text=bool(chunk.text_delta)) + yield chunk + if chunk.done: + break if not terminal_yielded: raise RuntimeError("SGLang stream terminated without terminal event") @@ -1707,13 +1666,36 @@ async def generate( ) +def _guard_verdict_logprobs(event: dict[str, Any]) -> tuple[dict[str, Any], ...]: + meta = event.get("meta_info") + if not isinstance(meta, dict): + return () + tokens, top = meta.get("output_token_logprobs"), meta.get("output_top_logprobs") + if not isinstance(tokens, list) or not isinstance(top, list) or len(top) < len(tokens): + return () + entries: list[dict[str, Any]] = [] + for token, alternatives in zip(tokens[:_GUARD_VERDICT_SCAN_POSITIONS], top, strict=False): + if not isinstance(alternatives, list): + return () + parsed: list[dict[str, Any]] = [] + for raw in [token, *alternatives]: + if not isinstance(raw, (list, tuple)) or len(raw) < 3 or not isinstance(raw[2], str): + return () + value = raw[0] + if not isinstance(value, (int, float)) or isinstance(value, bool) or not math.isfinite(value) or value > 0: + return () + parsed.append({"token": raw[2], "logprob": value, "bytes": list(raw[2].encode("utf-8"))}) + entries.append({**parsed[0], "top_logprobs": parsed[1:]}) + return tuple(entries) + + def _p_unsafe_from_entry(entry: Any) -> float | None: """``P(unsafe)`` from one OpenAI-shape content token's ``top_logprobs``. Renormalises ``exp(lp_yes)/(exp(lp_yes)+exp(lp_no))`` over the ``yes``/``no`` verdict tokens in this single position. ``None`` when neither appears. """ - if not isinstance(entry, dict): + if not isinstance(entry, dict) or str(entry.get("token") or "").strip().lower() not in ("yes", "no"): return None lp_yes: float | None = None lp_no: float | None = None @@ -1722,16 +1704,19 @@ def _p_unsafe_from_entry(entry: Any) -> float | None: continue tok = str(top.get("token") or "").strip().lower() val = top.get("logprob") - if not isinstance(val, (int, float)) or isinstance(val, bool): + if tok not in ("yes", "no"): continue + if not isinstance(val, (int, float)) or isinstance(val, bool) or not math.isfinite(val) or val > 0: + return None if tok == "yes": lp_yes = val if lp_yes is None else max(lp_yes, val) elif tok == "no": lp_no = val if lp_no is None else max(lp_no, val) if lp_yes is None and lp_no is None: return None - ey = math.exp(lp_yes) if lp_yes is not None else 0.0 - en = math.exp(lp_no) if lp_no is not None else 0.0 + offset = max(value for value in (lp_yes, lp_no) if value is not None) + ey = math.exp(lp_yes - offset) if lp_yes is not None else 0.0 + en = math.exp(lp_no - offset) if lp_no is not None else 0.0 return ey / (ey + en) if (ey + en) > 0 else None @@ -1760,7 +1745,7 @@ def _p_unsafe_from_verdict_logprobs( renormalises ``exp(lp_yes)/(exp(lp_yes)+exp(lp_no))`` over those two tokens. Scanning past position 0 keeps a leading whitespace/punctuation/preamble token from hiding the verdict, matching the eval runner's ``content[:3]`` - scan. ``None`` when no verdict token appears in range (caller keeps raw). + scan. ``None`` when no verdict token appears in range. """ if not chunk_logprobs: return None @@ -1776,19 +1761,23 @@ def _thresholded_verdict( guard: dict[str, Any], scan_positions: int = _GUARD_VERDICT_SCAN_POSITIONS, ) -> str | None: - """The guard's thresholded verdict token, or ``None`` to leave output as-is. + """The guard's thresholded verdict token, or ``None`` for an invalid verdict. ``guard`` is ``{"threshold": float, "positive": "Yes", "negative": "No"}`` (positive/negative default to Yes/No). Returns the ``positive`` label iff ``P(unsafe) >= threshold``, else ``negative``; ``None`` when P(unsafe) can't - be computed (no verdict logprobs within ``scan_positions``) so the raw model - token is preserved. + be computed from valid verdict logprobs within ``scan_positions``. """ p_unsafe = _p_unsafe_from_verdict_logprobs(chunk_logprobs, scan_positions) if p_unsafe is None: return None threshold = guard.get("threshold") - if not isinstance(threshold, (int, float)) or isinstance(threshold, bool): + if ( + not isinstance(threshold, (int, float)) + or isinstance(threshold, bool) + or not math.isfinite(threshold) + or not 0 <= threshold <= 1 + ): return None positive = str(guard.get("positive") or "Yes") negative = str(guard.get("negative") or "No") diff --git a/packages/sie_server/src/sie_server/api/generate.py b/packages/sie_server/src/sie_server/api/generate.py index d37ee5114..9102c9d61 100644 --- a/packages/sie_server/src/sie_server/api/generate.py +++ b/packages/sie_server/src/sie_server/api/generate.py @@ -850,19 +850,18 @@ async def _stream_generate_events( yield f"data: {json.dumps(event)}\n\n" seq += 1 break - if chunk.text_delta or chunk.logprobs: - if chunk.text_delta and ttft_ms is None: - ttft_ms = (time.perf_counter() - t0) * 1000.0 - event: dict[str, Any] = { - "request_id": request_id, - "seq": seq, - "text_delta": chunk.text_delta, - "done": False, - } - if chunk.logprobs: - event["logprobs"] = list(chunk.logprobs) - seq += 1 - yield f"data: {json.dumps(event)}\n\n" + if chunk.text_delta and ttft_ms is None: + ttft_ms = (time.perf_counter() - t0) * 1000.0 + event: dict[str, Any] = { + "request_id": request_id, + "seq": seq, + "text_delta": chunk.text_delta, + "done": False, + } + if chunk.logprobs: + event["logprobs"] = list(chunk.logprobs) + seq += 1 + yield f"data: {json.dumps(event)}\n\n" except GenerationError as exc: terminal_outcome_selected = True logger.info("stream_generate refused after preflight: %s", exc) diff --git a/packages/sie_server/src/sie_server/processors/streaming.py b/packages/sie_server/src/sie_server/processors/streaming.py index 0691340d6..20b1dc37f 100644 --- a/packages/sie_server/src/sie_server/processors/streaming.py +++ b/packages/sie_server/src/sie_server/processors/streaming.py @@ -2095,6 +2095,7 @@ async def _stream_generate( first_text_at: float | None = None publish_at = time.monotonic() first_yield_done = False + first_text_published = False terminal_sent = False # H6: separate flag for "the terminal we published was # ``transport_failure``" — even when the terminal lands on the @@ -2115,7 +2116,7 @@ def _record_failure() -> bool: publish_failures.pop(0) return len(publish_failures) >= _PUBLISH_FAIL_THRESHOLD - async def _flush_pending() -> bool: + async def _flush_pending(*, progress: bool = False) -> bool: """Enqueue any pending coalesced text. Returns False on overflow. Uses a bounded await (``_CHUNK_PUT_TIMEOUT_S``) so a brief @@ -2126,17 +2127,18 @@ async def _flush_pending() -> bool: a gap in the wire sequence (the gateway rejects gaps as a stream error). """ - nonlocal seq, pending_count, last_flush_ts - if not pending_text: + nonlocal seq, pending_count, last_flush_ts, first_text_published + if not pending_text and not progress: return True + text = "".join(pending_text) payload = _encode_chunk( kind="chunk", request_id=request_id, attempt_id=attempt_id, seq=seq, - text_delta="".join(pending_text), + text_delta=text, done=False, - is_first=(seq == 0), + is_first=bool(text) and not first_text_published, logprobs=pending_logprobs or None, ) try: @@ -2144,6 +2146,7 @@ async def _flush_pending() -> bool: except (asyncio.QueueFull, TimeoutError): return False seq += 1 + first_text_published = first_text_published or bool(text) pending_text.clear() pending_logprobs.clear() pending_count = 0 @@ -2455,8 +2458,8 @@ async def _next_chunk() -> GenerationChunk: pending_logprobs.extend(chunk.logprobs) now = time.monotonic() - if pending_count >= _FLUSH_MAX_TOKENS or (pending_text and (now - last_flush_ts) >= _FLUSH_INTERVAL_S): - ok = await _flush_pending() + if pending_count >= _FLUSH_MAX_TOKENS or seq == 0 or (now - last_flush_ts) >= _FLUSH_INTERVAL_S: + ok = await _flush_pending(progress=True) if not ok: # Flush failed after a bounded await on a non-terminal # chunk — content was dropped. Publish diff --git a/packages/sie_server/src/sie_server/processors/tool_call_parser.py b/packages/sie_server/src/sie_server/processors/tool_call_parser.py index 6e89696e6..196de35d3 100644 --- a/packages/sie_server/src/sie_server/processors/tool_call_parser.py +++ b/packages/sie_server/src/sie_server/processors/tool_call_parser.py @@ -351,6 +351,9 @@ def _state(idx: int) -> _ChoiceState: return s async for chunk in chunks: + if not chunk.text_delta and not chunk.done and chunk.finish_reason is None and not chunk.logprobs: + yield chunk + continue idx = chunk.choice_index state = _state(idx) incoming = chunk.text_delta diff --git a/packages/sie_server/tests/adapters/test_generation_base.py b/packages/sie_server/tests/adapters/test_generation_base.py index add6a2224..1f94c3bca 100644 --- a/packages/sie_server/tests/adapters/test_generation_base.py +++ b/packages/sie_server/tests/adapters/test_generation_base.py @@ -282,12 +282,17 @@ async def test_suppress_thinking_blocks_hides_truncated_reasoning_and_its_logpro normalized = [chunk async for chunk in suppress_thinking_blocks(source)] - assert len(normalized) == 1 - assert normalized[0].done is True - assert normalized[0].finish_reason == "length" - assert normalized[0].completion_tokens == 2 + assert len(normalized) == 2 + assert normalized[0].done is False assert normalized[0].text_delta == "" + assert normalized[0].is_first is False assert normalized[0].logprobs is None + assert normalized[1].done is True + assert normalized[1].finish_reason == "error" + assert normalized[1].error_code == "empty_model_output" + assert normalized[1].completion_tokens == 2 + assert normalized[1].text_delta == "" + assert normalized[1].logprobs is None @pytest.mark.asyncio @@ -436,7 +441,7 @@ async def test_reasoning_consuming_the_whole_budget_stamps_empty_output_error() assert "".join(chunk.text_delta for chunk in normalized) == "\n\n" terminal = normalized[-1] assert terminal.done is True - assert terminal.finish_reason == "length" + assert terminal.finish_reason == "error" assert terminal.error_code == "empty_model_output" assert terminal.error_message is not None assert "reasoning" in terminal.error_message diff --git a/packages/sie_server/tests/adapters/test_sglang_generation.py b/packages/sie_server/tests/adapters/test_sglang_generation.py index 559c98964..050c1a4ce 100644 --- a/packages/sie_server/tests/adapters/test_sglang_generation.py +++ b/packages/sie_server/tests/adapters/test_sglang_generation.py @@ -1952,8 +1952,7 @@ def test_threshold_high_recall_vs_precision(self) -> None: assert _thresholded_verdict(lp, {"threshold": 0.5}) == "Yes" assert _thresholded_verdict(lp, {"threshold": 0.8}) == "No" - def test_no_threshold_or_no_logprobs_leaves_output(self) -> None: - # Missing/invalid threshold or absent verdict logprobs -> None (raw kept). + def test_no_threshold_or_no_logprobs_is_invalid(self) -> None: assert _thresholded_verdict(self._chunk_logprobs(-0.1, -2.0), {}) is None assert _thresholded_verdict((), {"threshold": 0.8}) is None @@ -2167,10 +2166,8 @@ def test_guard_verdict_in_second_position_applies_threshold(mock_async_client: M @patch("sie_server.adapters.sglang.generation.httpx.AsyncClient") -def test_guard_no_verdict_in_scan_window_falls_back_to_raw(mock_async_client: MagicMock) -> None: - """H2 fallback: no Yes/No anywhere in the first N positions → the raw buffered - output is preserved (never dropped or hung). - """ +def test_guard_no_verdict_in_scan_window_fails_closed(mock_async_client: MagicMock) -> None: + """A missing verdict must never be returned as a successful guard result.""" sse_lines = [ _guard_event("a", [_sglang_token("a", -0.1)], [_guard_top(filler="a")]), _guard_event( @@ -2188,9 +2185,10 @@ def test_guard_no_verdict_in_scan_window_falls_back_to_raw(mock_async_client: Ma ] mock_async_client.return_value = _make_client_with_stream(_FakeStreamingResponse(sse_lines)) chunks = _drive_guard(_guard_adapter(), sse_lines) - # Raw output preserved: the concatenated deltas reproduce the model output. - assert "".join(c.text_delta for c in chunks) == "abc" - assert any(c.done for c in chunks) + assert all(c.text_delta == "" for c in chunks) + assert chunks[-1].done + assert chunks[-1].finish_reason == "error" + assert chunks[-1].error_code == "invalid_guard_verdict" @patch("sie_server.adapters.sglang.generation.httpx.AsyncClient") @@ -2215,10 +2213,8 @@ def test_guard_no_client_logprobs_strips_forced_logprobs_on_success(mock_async_c @patch("sie_server.adapters.sglang.generation.httpx.AsyncClient") -def test_guard_no_client_logprobs_strips_forced_logprobs_on_fallback(mock_async_client: MagicMock) -> None: - """M4: on the fallback path (no verdict parsed) the forced logprobs are still - stripped when the client did not request them. - """ +def test_guard_no_client_logprobs_strips_forced_logprobs_on_error(mock_async_client: MagicMock) -> None: + """Unusable guard output exposes neither raw text nor forced logprobs.""" sse_lines = [ _guard_event("a", [_sglang_token("a", -0.1)], [_guard_top(filler="a")]), _guard_event( @@ -2232,7 +2228,8 @@ def test_guard_no_client_logprobs_strips_forced_logprobs_on_fallback(mock_async_ mock_async_client.return_value = _make_client_with_stream(_FakeStreamingResponse(sse_lines)) chunks = _drive_guard(_guard_adapter(), sse_lines) assert all(c.logprobs is None for c in chunks) - assert "".join(c.text_delta for c in chunks) == "ab" + assert "".join(c.text_delta for c in chunks) == "" + assert chunks[-1].error_code == "invalid_guard_verdict" @patch("sie_server.adapters.sglang.generation.httpx.AsyncClient") @@ -2497,3 +2494,86 @@ async def consume() -> None: with pytest.raises(GenerationError, match="incorrect candidate count"): asyncio.run(consume()) assert chunks == [] + + +@pytest.mark.parametrize("text", ["", "No", "Maybe"]) +@patch("sie_server.adapters.sglang.generation.httpx.AsyncClient") +def test_guard_missing_verdict_distribution_is_typed_error(mock_async_client: MagicMock, text: str) -> None: + lines = [_guard_event(text, [_sglang_token(text, -0.1)], [], terminal=True), "data: [DONE]"] + mock_async_client.return_value = _make_client_with_stream(_FakeStreamingResponse(lines)) + chunks = _drive_guard(_guard_adapter(), lines, logprobs=True) + assert len(chunks) == 1 + terminal = chunks[0] + assert terminal.text_delta == "" + assert terminal.logprobs is None + assert terminal.finish_reason == "error" + assert terminal.error_code == "invalid_guard_verdict" + assert terminal.prompt_tokens == 5 + assert terminal.completion_tokens == 1 + + +@pytest.mark.parametrize("invalid", [float("nan"), float("inf"), float("-inf"), 0.1, True, "-0.1"]) +def test_guard_invalid_verdict_probability_cannot_be_safe(invalid: Any) -> None: + logprobs = ({"token": "No", "top_logprobs": [_lp("Yes", invalid), _lp("No", -0.1)]},) + assert _thresholded_verdict(logprobs, {"threshold": 0.5}) is None + + +@pytest.mark.parametrize("threshold", [float("nan"), float("inf"), -0.1, 1.1, True, "0.5"]) +def test_guard_invalid_threshold_cannot_be_safe(threshold: Any) -> None: + logprobs = ({"token": "Yes", "top_logprobs": [_lp("Yes", -0.5), _lp("No", -0.5)]},) + assert _thresholded_verdict(logprobs, {"threshold": threshold}) is None + + +def test_guard_very_small_probabilities_are_normalized_without_underflow() -> None: + logprobs = ({"token": "Yes", "top_logprobs": [_lp("Yes", -1000), _lp("No", -1001)]},) + assert _thresholded_verdict(logprobs, {"threshold": 0.5}) == "Yes" + + +@patch("sie_server.adapters.sglang.generation.httpx.AsyncClient") +def test_guard_eos_with_verdict_alternatives_is_not_a_verdict(mock_async_client: MagicMock) -> None: + lines = [_guard_event("", [_sglang_token("<|end_of_text|>", -0.1)], [_guard_top(yes=-4, no=-3)], terminal=True)] + mock_async_client.return_value = _make_client_with_stream(_FakeStreamingResponse(lines)) + chunks = _drive_guard(_guard_adapter(), lines) + assert chunks[-1].error_code == "invalid_guard_verdict" + assert chunks[-1].text_delta == "" + + +@patch("sie_server.adapters.sglang.generation.httpx.AsyncClient") +def test_guard_tail_cannot_change_thresholded_verdict(mock_async_client: MagicMock) -> None: + lines = [ + _guard_event("Yes", [_sglang_token("Yes", -0.1)], [_guard_top(yes=-0.1, no=-3)]), + _guard_event( + "Yes because", + [_sglang_token("Yes", -0.1), _sglang_token(" because", -0.2)], + [_guard_top(yes=-0.1, no=-3), _guard_top(filler=" because")], + terminal=True, + ), + ] + mock_async_client.return_value = _make_client_with_stream(_FakeStreamingResponse(lines)) + chunks = _drive_guard(_guard_adapter(), lines) + assert "".join(c.text_delta for c in chunks) == "Yes" + assert chunks[-1].done + assert chunks[-1].completion_tokens == 2 + + +@pytest.mark.parametrize("invalid", [None, "-0.1", True, False, float("nan"), float("inf"), float("-inf"), 0.1]) +@pytest.mark.parametrize("position", ["sampled", "alternative"]) +@patch("sie_server.adapters.sglang.generation.httpx.AsyncClient") +def test_guard_malformed_wire_probability_fails_closed( + mock_async_client: MagicMock, invalid: Any, position: str +) -> None: + sampled = invalid if position == "sampled" else -0.1 + alternative = invalid if position == "alternative" else -0.1 + lines = [ + _guard_event( + "No", + [_sglang_token("No", sampled)], + [[_sglang_token("Yes", alternative), _sglang_token("No", -0.1)]], + terminal=True, + ) + ] + mock_async_client.return_value = _make_client_with_stream(_FakeStreamingResponse(lines)) + chunks = _drive_guard(_guard_adapter(), lines) + assert chunks[-1].error_code == "invalid_guard_verdict" + assert chunks[-1].finish_reason == "error" + assert not any(c.text_delta for c in chunks) diff --git a/packages/sie_server/tests/api/test_generate.py b/packages/sie_server/tests/api/test_generate.py index 59fdd76b0..9d0ae4159 100644 --- a/packages/sie_server/tests/api/test_generate.py +++ b/packages/sie_server/tests/api/test_generate.py @@ -393,17 +393,27 @@ async def render(_config: object, prompt: str, image_count: int) -> str: ] streamed_text = "".join(event.get("text_delta", "") for event in events) assert streamed_text == "Visible answer" + assert events[0]["text_delta"] == "" + assert events[0]["done"] is False assert "private reasoning" not in streamed_text else: assert response.json()["text"] == "Visible answer" @pytest.mark.parametrize("stream", [False, True]) + @pytest.mark.parametrize( + "template_kwargs", [{"enable_thinking": False}, {"guardian_config": {"risk_name": "harm"}}] + ) def test_text_only_generate_preserves_legacy_adapter_call_signature( self, client: TestClient, registry: MagicMock, stream: bool, + template_kwargs: dict[str, object], ) -> None: + config = _make_config() + assert config.tasks.generate is not None + config.tasks.generate.chat_template_kwargs = template_kwargs + registry.get_config.return_value = config legacy_adapter = _LegacyTextGenAdapter() registry.get.return_value = legacy_adapter diff --git a/packages/sie_server/tests/processors/test_streaming.py b/packages/sie_server/tests/processors/test_streaming.py index d37663769..341191db2 100644 --- a/packages/sie_server/tests/processors/test_streaming.py +++ b/packages/sie_server/tests/processors/test_streaming.py @@ -1120,8 +1120,11 @@ async def test_streaming_processor_hides_reasoning_for_every_resolved_profile( @pytest.mark.asyncio -@pytest.mark.parametrize("enable_thinking", [False, True]) -async def test_streaming_processor_renders_chat_template(monkeypatch, enable_thinking: bool) -> None: +@pytest.mark.parametrize( + "template_kwargs", + [{"enable_thinking": False}, {"enable_thinking": True}, {"guardian_config": {"risk_name": "harm"}}], +) +async def test_streaming_processor_renders_chat_template(monkeypatch, template_kwargs: dict[str, Any]) -> None: """``Messages`` shape → adapter receives the rendered template string.""" nc = AsyncMock() script = [ @@ -1144,7 +1147,7 @@ async def _capture_generate(prompt, *, max_new_tokens, temperature=1.0, top_p=1. registry = _make_registry_with_chat_config( adapter, - chat_template_kwargs={"enable_thinking": enable_thinking}, + chat_template_kwargs=template_kwargs, ) # Patch ``load_tokenizer`` (called from a thread) with a stub that @@ -1172,7 +1175,7 @@ def encode(self, text, *, add_special_tokens): assert len(captured_prompts) == 1 assert captured_prompts[0] == "ping" - assert seen_kwargs == {"enable_thinking": enable_thinking} + assert seen_kwargs == template_kwargs decoded = _decode_chunks(nc) visible = "".join(chunk.get("text_delta", "") for chunk in decoded) assert visible == "hi" @@ -4026,3 +4029,58 @@ async def test_image_generation_without_token_counts_does_not_synthesize_usage( terminal = _terminal_chunk(nc) assert terminal["done"] is True assert "usage" not in terminal + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "tools", [None, [{"type": "function", "function": {"name": "lookup", "parameters": {"type": "object"}}}]] +) +async def test_hidden_reasoning_emits_progress_before_visible_answer( + monkeypatch: pytest.MonkeyPatch, tools: Any +) -> None: + nc = AsyncMock() + script = [ + GenerationChunk(text_delta="private", is_first=True, logprobs=({"token": "private"},)), + GenerationChunk(text_delta=" still private"), + GenerationChunk(text_delta="Answer"), + GenerationChunk(text_delta="", done=True, finish_reason="stop", prompt_tokens=2, completion_tokens=8), + ] + adapter = _FakeGenAdapter(script) + registry = _make_registry_with_chat_config(adapter, chat_template_kwargs={"enable_thinking": False}) + proc = StreamingProcessor(nc=nc, registry=registry, worker_id="w1") + monkeypatch.setattr(proc, "_check_context_length", AsyncMock(return_value=None)) + monkeypatch.setattr("sie_server.processors.streaming._FLUSH_INTERVAL_S", 0) + work = _make_work_item() + if tools is not None: + work["generate"]["tools"] = tools + await proc.process(_make_msg(work), "test/model") + decoded = _decode_chunks(nc) + assert [chunk["seq"] for chunk in decoded] == list(range(len(decoded))) + assert decoded[0]["text_delta"] == "" + assert decoded[0]["done"] is False + assert not decoded[0].get("is_first") + assert decoded[1]["text_delta"] == "" + assert "private" not in str(decoded) + visible = next(chunk for chunk in decoded if chunk["text_delta"]) + assert visible["text_delta"] == "Answer" + assert visible["is_first"] is True + assert decoded[-1]["done"] is True + + +@pytest.mark.asyncio +@pytest.mark.parametrize("template_kwargs", [{"enable_thinking": False}, {"guardian_config": {"risk_name": "harm"}}]) +async def test_native_raw_prompt_is_preserved_with_served_template_settings( + monkeypatch: pytest.MonkeyPatch, template_kwargs: dict[str, Any] +) -> None: + nc = AsyncMock() + adapter = _PreflightGenAdapter() + registry = _make_registry_with_chat_config(adapter, chat_template_kwargs=template_kwargs) + proc = StreamingProcessor(nc=nc, registry=registry, worker_id="w1") + monkeypatch.setattr(proc, "_check_context_length", AsyncMock(return_value=None)) + render = AsyncMock() + monkeypatch.setattr(proc, "_render_chat_template", render) + prompt = "Already rendered prefix\n" + await proc.process(_make_msg(_make_work_item(generate={"prompt": prompt, "max_new_tokens": 16})), "test/model") + assert adapter.dispatched_parameters is not None + assert adapter.dispatched_parameters["prompt"] == prompt + render.assert_not_called() diff --git a/packages/sie_ts_sdk/README.md b/packages/sie_ts_sdk/README.md index c62bb192c..ecb110f72 100644 --- a/packages/sie_ts_sdk/README.md +++ b/packages/sie_ts_sdk/README.md @@ -67,6 +67,21 @@ console.log(scored.scores[0].itemId, scored.scores[0].score); ## Generation +Text-only `generate` and `streamGenerate` prompts are raw continuation input: +the worker preserves them without rendering a chat template or applying +`enable_thinking` / `guardian_config`. Native requests with images render one +user turn. For chat, instruction-based structured output (`responseFormat`), +and guard checks, use `chatCompletions` or `streamChatCompletions` with messages +so the worker applies the served template settings. Operator settings take +precedence over request template kwargs. + +Granite Guardian's shipped risk dimension is `harm`; requesting another +dimension in prompt text does not change it. Valid thresholded verdicts are +`Yes` (unsafe) and `No` (safe); missing or invalid verdicts fail with +`invalid_guard_verdict`. Never interpret an error or empty output as safe. +Reasoning remains private; a budget exhausted without usable output fails with +`empty_model_output`. + ```typescript // Aggregated result const gen = await client.generate( diff --git a/packages/sie_ts_sdk/src/client.ts b/packages/sie_ts_sdk/src/client.ts index 64705a51d..53b9f2c2f 100644 --- a/packages/sie_ts_sdk/src/client.ts +++ b/packages/sie_ts_sdk/src/client.ts @@ -1270,6 +1270,11 @@ export class SIEClient { * the gateway aggregates, and the SDK returns the assembled result * plus SIE-native timing metadata (TTFT, TPOT, attempt id). To * consume chunks as they arrive, use {@link streamGenerate} instead. + * Text-only prompts are raw continuation input, passed unchanged without + * the model's chat template or its enable_thinking/guardian_config settings. + * Use {@link chatCompletions} with messages for chat, instruction-based + * structured output, and guard checks. Image-bearing native requests render + * the prompt and images as one user turn through the model's template. * * @example * ```typescript From ba8c3b11e98711b974bef5f07d8ba15a6e5bcbbd Mon Sep 17 00:00:00 2001 From: Attila Laszlo Nagy Date: Wed, 16 Sep 2026 17:10:00 +0200 Subject: [PATCH 2/3] fix: omit logprobs for rewritten guard verdicts --- .../sie_server/adapters/sglang/generation.py | 29 +------------------ .../tests/adapters/test_sglang_generation.py | 19 ++++++------ 2 files changed, 10 insertions(+), 38 deletions(-) diff --git a/packages/sie_server/src/sie_server/adapters/sglang/generation.py b/packages/sie_server/src/sie_server/adapters/sglang/generation.py index 16bbe4a63..8e64415ed 100644 --- a/packages/sie_server/src/sie_server/adapters/sglang/generation.py +++ b/packages/sie_server/src/sie_server/adapters/sglang/generation.py @@ -1085,13 +1085,6 @@ async def generate( # verdict from the multi-candidate path. Inert for non-guard models. if self._guard and ((n is not None and n > 1) or (best_of is not None and best_of > 1)): raise ValueError("guard models support single-candidate generation only (n=1, best_of<=1)") - # Whether the CLIENT asked for logprobs, captured before the guard - # forcing below. Guard models force logprobs on internally to compute - # the verdict threshold; those forced logprobs are an implementation - # detail and MUST NOT leak to a client that did not request them - # (GenerationChunk.logprobs contract). The streaming guard intercept - # uses this to decide whether to strip the forced logprobs. - client_requested_logprobs = logprobs # Thresholding needs the verdict-token distribution — force logprobs on # even if the caller didn't ask. Only affects the n=1 path below. if self._guard: @@ -1604,14 +1597,7 @@ async def generate( verdict = _thresholded_verdict(guard_lp_buffer, self._guard) if verdict is not None: guard_resolved = True - if client_requested_logprobs: - v_idx = _verdict_position(tuple(guard_lp_buffer)) - remaining = tuple(e for i, e in enumerate(guard_lp_buffer) if i != v_idx) or None - else: - remaining = None - chunk = dataclasses.replace( - chunk, text_delta=verdict, is_first=True, logprobs=remaining - ) + chunk = dataclasses.replace(chunk, text_delta=verdict, is_first=True, logprobs=None) elif chunk.done: chunk = dataclasses.replace( chunk, @@ -1720,19 +1706,6 @@ def _p_unsafe_from_entry(entry: Any) -> float | None: return ey / (ey + en) if (ey + en) > 0 else None -def _verdict_position(chunk_logprobs: Any, scan_positions: int = _GUARD_VERDICT_SCAN_POSITIONS) -> int | None: - """Index of the first position (within ``scan_positions``) carrying a verdict - distribution, or ``None``. The consumed/rewritten verdict entry the streaming - intercept drops from client-requested logprobs. - """ - if not chunk_logprobs: - return None - for idx, entry in enumerate(chunk_logprobs[:scan_positions]): - if _p_unsafe_from_entry(entry) is not None: - return idx - return None - - def _p_unsafe_from_verdict_logprobs( chunk_logprobs: Any, scan_positions: int = _GUARD_VERDICT_SCAN_POSITIONS ) -> float | None: diff --git a/packages/sie_server/tests/adapters/test_sglang_generation.py b/packages/sie_server/tests/adapters/test_sglang_generation.py index 050c1a4ce..2d2c7aaa5 100644 --- a/packages/sie_server/tests/adapters/test_sglang_generation.py +++ b/packages/sie_server/tests/adapters/test_sglang_generation.py @@ -2233,10 +2233,8 @@ def test_guard_no_client_logprobs_strips_forced_logprobs_on_error(mock_async_cli @patch("sie_server.adapters.sglang.generation.httpx.AsyncClient") -def test_guard_client_logprobs_preserved_minus_verdict_entry(mock_async_client: MagicMock) -> None: - """M4: client DID request logprobs → logprobs preserved, minus the consumed - verdict entry (the position that produced the verdict). - """ +def test_guard_client_logprobs_omitted_for_rewritten_verdict(mock_async_client: MagicMock) -> None: + """Rewritten verdicts cannot expose logprobs for discarded tokens.""" # Position 0: leading whitespace (no verdict). Position 1: the "Yes" verdict. # Both arrive in one event so the buffer holds two entries at resolution; the # consumed verdict entry (position 1) is dropped, the whitespace entry kept. @@ -2253,10 +2251,7 @@ def test_guard_client_logprobs_preserved_minus_verdict_entry(mock_async_client: chunks = _drive_guard(_guard_adapter(), sse_lines, logprobs=True, top_logprobs=20) verdict_chunk = next(c for c in chunks if c.text_delta) assert verdict_chunk.text_delta == "Yes" - # The consumed verdict entry (position 1) is dropped; the leading whitespace - # entry remains so callers still get the surrounding token metadata. - assert verdict_chunk.logprobs is not None - assert [e["token"] for e in verdict_chunk.logprobs] == [" "] + assert verdict_chunk.logprobs is None @patch("sie_server.adapters.sglang.generation.httpx.AsyncClient") @@ -2538,8 +2533,9 @@ def test_guard_eos_with_verdict_alternatives_is_not_a_verdict(mock_async_client: assert chunks[-1].text_delta == "" +@pytest.mark.parametrize("combined", [False, True]) @patch("sie_server.adapters.sglang.generation.httpx.AsyncClient") -def test_guard_tail_cannot_change_thresholded_verdict(mock_async_client: MagicMock) -> None: +def test_guard_tail_cannot_change_thresholded_verdict(mock_async_client: MagicMock, combined: bool) -> None: lines = [ _guard_event("Yes", [_sglang_token("Yes", -0.1)], [_guard_top(yes=-0.1, no=-3)]), _guard_event( @@ -2549,8 +2545,11 @@ def test_guard_tail_cannot_change_thresholded_verdict(mock_async_client: MagicMo terminal=True, ), ] + if combined: + lines = lines[-1:] mock_async_client.return_value = _make_client_with_stream(_FakeStreamingResponse(lines)) - chunks = _drive_guard(_guard_adapter(), lines) + chunks = _drive_guard(_guard_adapter(), lines, logprobs=True, top_logprobs=20) + assert all(chunk.logprobs is None for chunk in chunks) assert "".join(c.text_delta for c in chunks) == "Yes" assert chunks[-1].done assert chunks[-1].completion_tokens == 2 From 514b07fffc4b5f84985d5861cb12fc1de0564fd5 Mon Sep 17 00:00:00 2001 From: Attila Laszlo Nagy Date: Wed, 16 Sep 2026 18:29:38 +0200 Subject: [PATCH 3/3] fix: validate complete first guard verdict evidence --- .../sie_server/adapters/sglang/generation.py | 30 +++++---- .../tests/adapters/test_sglang_generation.py | 67 ++++++++++++++++++- 2 files changed, 81 insertions(+), 16 deletions(-) diff --git a/packages/sie_server/src/sie_server/adapters/sglang/generation.py b/packages/sie_server/src/sie_server/adapters/sglang/generation.py index 8e64415ed..588a7f4e1 100644 --- a/packages/sie_server/src/sie_server/adapters/sglang/generation.py +++ b/packages/sie_server/src/sie_server/adapters/sglang/generation.py @@ -1657,10 +1657,13 @@ def _guard_verdict_logprobs(event: dict[str, Any]) -> tuple[dict[str, Any], ...] if not isinstance(meta, dict): return () tokens, top = meta.get("output_token_logprobs"), meta.get("output_top_logprobs") - if not isinstance(tokens, list) or not isinstance(top, list) or len(top) < len(tokens): + if not isinstance(tokens, list) or not isinstance(top, list): return () entries: list[dict[str, Any]] = [] - for token, alternatives in zip(tokens[:_GUARD_VERDICT_SCAN_POSITIONS], top, strict=False): + for index, token in enumerate(tokens[:_GUARD_VERDICT_SCAN_POSITIONS]): + if index >= len(top): + return () + alternatives = top[index] if not isinstance(alternatives, list): return () parsed: list[dict[str, Any]] = [] @@ -1672,6 +1675,8 @@ def _guard_verdict_logprobs(event: dict[str, Any]) -> tuple[dict[str, Any], ...] return () parsed.append({"token": raw[2], "logprob": value, "bytes": list(raw[2].encode("utf-8"))}) entries.append({**parsed[0], "top_logprobs": parsed[1:]}) + if parsed[0]["token"].strip().lower() in ("yes", "no"): + break return tuple(entries) @@ -1679,13 +1684,13 @@ def _p_unsafe_from_entry(entry: Any) -> float | None: """``P(unsafe)`` from one OpenAI-shape content token's ``top_logprobs``. Renormalises ``exp(lp_yes)/(exp(lp_yes)+exp(lp_no))`` over the ``yes``/``no`` - verdict tokens in this single position. ``None`` when neither appears. + verdict tokens in this single position. Both probabilities are required. """ if not isinstance(entry, dict) or str(entry.get("token") or "").strip().lower() not in ("yes", "no"): return None lp_yes: float | None = None lp_no: float | None = None - for top in entry.get("top_logprobs") or []: + for top in [entry, *(entry.get("top_logprobs") or [])]: if not isinstance(top, dict): continue tok = str(top.get("token") or "").strip().lower() @@ -1698,11 +1703,11 @@ def _p_unsafe_from_entry(entry: Any) -> float | None: lp_yes = val if lp_yes is None else max(lp_yes, val) elif tok == "no": lp_no = val if lp_no is None else max(lp_no, val) - if lp_yes is None and lp_no is None: + if lp_yes is None or lp_no is None: return None - offset = max(value for value in (lp_yes, lp_no) if value is not None) - ey = math.exp(lp_yes - offset) if lp_yes is not None else 0.0 - en = math.exp(lp_no - offset) if lp_no is not None else 0.0 + offset = max(lp_yes, lp_no) + ey = math.exp(lp_yes - offset) + en = math.exp(lp_no - offset) return ey / (ey + en) if (ey + en) > 0 else None @@ -1713,8 +1718,8 @@ def _p_unsafe_from_verdict_logprobs( ``chunk_logprobs`` is the OpenAI ``content`` shape this adapter builds — ``({"token", "logprob", "top_logprobs": [{"token", "logprob"}, ...]}, ...)``. - Scans the first up-to ``scan_positions`` content tokens for the first whose - ``top_logprobs`` carries a ``yes``/``no`` verdict distribution, then + Scans the first up-to ``scan_positions`` content tokens for the first sampled + ``yes``/``no`` verdict and validates its distribution, then renormalises ``exp(lp_yes)/(exp(lp_yes)+exp(lp_no))`` over those two tokens. Scanning past position 0 keeps a leading whitespace/punctuation/preamble token from hiding the verdict, matching the eval runner's ``content[:3]`` @@ -1723,9 +1728,8 @@ def _p_unsafe_from_verdict_logprobs( if not chunk_logprobs: return None for entry in chunk_logprobs[:scan_positions]: - p_unsafe = _p_unsafe_from_entry(entry) - if p_unsafe is not None: - return p_unsafe + if isinstance(entry, dict) and str(entry.get("token") or "").strip().lower() in ("yes", "no"): + return _p_unsafe_from_entry(entry) return None diff --git a/packages/sie_server/tests/adapters/test_sglang_generation.py b/packages/sie_server/tests/adapters/test_sglang_generation.py index 2d2c7aaa5..698e780cd 100644 --- a/packages/sie_server/tests/adapters/test_sglang_generation.py +++ b/packages/sie_server/tests/adapters/test_sglang_generation.py @@ -2509,18 +2509,18 @@ def test_guard_missing_verdict_distribution_is_typed_error(mock_async_client: Ma @pytest.mark.parametrize("invalid", [float("nan"), float("inf"), float("-inf"), 0.1, True, "-0.1"]) def test_guard_invalid_verdict_probability_cannot_be_safe(invalid: Any) -> None: - logprobs = ({"token": "No", "top_logprobs": [_lp("Yes", invalid), _lp("No", -0.1)]},) + logprobs = ({"token": "No", "logprob": -0.1, "top_logprobs": [_lp("Yes", invalid), _lp("No", -0.1)]},) assert _thresholded_verdict(logprobs, {"threshold": 0.5}) is None @pytest.mark.parametrize("threshold", [float("nan"), float("inf"), -0.1, 1.1, True, "0.5"]) def test_guard_invalid_threshold_cannot_be_safe(threshold: Any) -> None: - logprobs = ({"token": "Yes", "top_logprobs": [_lp("Yes", -0.5), _lp("No", -0.5)]},) + logprobs = ({"token": "Yes", "logprob": -0.5, "top_logprobs": [_lp("Yes", -0.5), _lp("No", -0.5)]},) assert _thresholded_verdict(logprobs, {"threshold": threshold}) is None def test_guard_very_small_probabilities_are_normalized_without_underflow() -> None: - logprobs = ({"token": "Yes", "top_logprobs": [_lp("Yes", -1000), _lp("No", -1001)]},) + logprobs = ({"token": "Yes", "logprob": -1000, "top_logprobs": [_lp("Yes", -1000), _lp("No", -1001)]},) assert _thresholded_verdict(logprobs, {"threshold": 0.5}) == "Yes" @@ -2576,3 +2576,64 @@ def test_guard_malformed_wire_probability_fails_closed( assert chunks[-1].error_code == "invalid_guard_verdict" assert chunks[-1].finish_reason == "error" assert not any(c.text_delta for c in chunks) + + +@pytest.mark.parametrize("malformed_tail", ["sampled", "alternative", "missing_top", "malformed_top"]) +@patch("sie_server.adapters.sglang.generation.httpx.AsyncClient") +def test_guard_valid_verdict_ignores_malformed_trailing_metadata( + mock_async_client: MagicMock, malformed_tail: str +) -> None: + tokens = [_sglang_token("Yes", -0.1), _sglang_token(" because", -0.2)] + top: list[Any] = [_guard_top(yes=-0.1, no=-3), _guard_top(filler=" because")] + if malformed_tail == "sampled": + tokens[1][0] = True + elif malformed_tail == "alternative": + top[1][0][0] = "-0.1" + elif malformed_tail == "missing_top": + top.pop() + else: + top[1] = None + lines = [_guard_event("Yes because", tokens, top, terminal=True)] + mock_async_client.return_value = _make_client_with_stream(_FakeStreamingResponse(lines)) + chunks = _drive_guard(_guard_adapter(), lines, logprobs=True) + assert "".join(chunk.text_delta for chunk in chunks) == "Yes" + assert chunks[-1].error_code is None + assert all(chunk.logprobs is None for chunk in chunks) + + +@pytest.mark.parametrize(("sampled", "opposing"), [("Yes", "No"), ("No", "Yes")]) +@patch("sie_server.adapters.sglang.generation.httpx.AsyncClient") +def test_guard_uses_sampled_probability_when_absent_from_top_alternatives( + mock_async_client: MagicMock, sampled: str, opposing: str +) -> None: + lines = [_guard_event(sampled, [_sglang_token(sampled, -0.1)], [[_sglang_token(opposing, -3)]], terminal=True)] + mock_async_client.return_value = _make_client_with_stream(_FakeStreamingResponse(lines)) + chunks = _drive_guard(_guard_adapter(), lines) + assert "".join(chunk.text_delta for chunk in chunks) == sampled + assert chunks[-1].error_code is None + + +@pytest.mark.parametrize("sampled", ["Yes", "No"]) +@pytest.mark.parametrize("later_verdict", [False, True]) +@patch("sie_server.adapters.sglang.generation.httpx.AsyncClient") +def test_guard_missing_opposing_probability_fails_closed( + mock_async_client: MagicMock, sampled: str, later_verdict: bool +) -> None: + tokens = [_sglang_token(sampled, -0.1)] + top = [[_sglang_token(sampled, -0.1)]] + if later_verdict: + tokens.append(_sglang_token("No", -0.1)) + top.append(_guard_top(yes=-3, no=-0.1)) + lines = [_guard_event(sampled, tokens, top, terminal=True)] + mock_async_client.return_value = _make_client_with_stream(_FakeStreamingResponse(lines)) + chunks = _drive_guard(_guard_adapter(), lines) + assert chunks[-1].error_code == "invalid_guard_verdict" + assert not any(chunk.text_delta for chunk in chunks) + + +def test_guard_first_sampled_verdict_requires_complete_evidence() -> None: + later_safe = {"token": "No", "logprob": -0.1, "top_logprobs": [_lp("Yes", -3), _lp("No", -0.1)]} + incomplete_yes = {"token": "Yes", "logprob": -0.1, "top_logprobs": [_lp("Yes", -0.1)]} + assert _p_unsafe_from_verdict_logprobs((incomplete_yes, later_safe)) is None + missing_sampled = {"token": "Yes", "top_logprobs": [_lp("Yes", -0.1), _lp("No", -3)]} + assert _p_unsafe_from_verdict_logprobs((missing_sampled, later_safe)) is None