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..588a7f4e1 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:
@@ -1548,20 +1541,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 +1589,33 @@ 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
+ chunk = dataclasses.replace(chunk, text_delta=verdict, is_first=True, logprobs=None)
+ 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,47 +1652,65 @@ 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):
+ return ()
+ entries: list[dict[str, Any]] = []
+ 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]] = []
+ 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:]})
+ if parsed[0]["token"].strip().lower() in ("yes", "no"):
+ break
+ 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.
+ verdict tokens in this single position. Both probabilities are required.
"""
- 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
- 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()
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:
+ if lp_yes is None or 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(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
-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:
@@ -1755,19 +1718,18 @@ 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]``
- 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
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
@@ -1776,19 +1738,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..698e780cd 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,14 +2228,13 @@ 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")
-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.
@@ -2256,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")
@@ -2497,3 +2489,151 @@ 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", "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", "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", "logprob": -1000, "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 == ""
+
+
+@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, combined: bool) -> 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,
+ ),
+ ]
+ if combined:
+ lines = lines[-1:]
+ mock_async_client.return_value = _make_client_with_stream(_FakeStreamingResponse(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
+
+
+@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)
+
+
+@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
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