retoor/devplacepy
1
0
Fork 1
Code Issues 44 Pull Requests 4 Actions Packages Projects Releases Wiki Activity

Add admin-unlimited workspaces, AI gateway model fallback, and real streaming/thinking control

Browse Source 
Admin-unlimited Dev Workspaces: an admin-owned workspace is now exempt from the
max-workspace-count limit, the max-tunnel-count limit, and the whole
idle-stop/idle-warn/retention-delete lifecycle. Resolved once in
quota.resolve() as Limits.unlimited (owner uid checked against
get_admin_uids()), consumed at the three enforcement points
(provision.ensure, provision.publish_tunnel,
WorkspaceService._advance_lifecycle). Also hardens
get_admin_uids()/get_primary_admin_uid() against a partially-schemaed users
table (uid/role column guard), which a fresh test/init_db() path could hit.

AI gateway per-model automatic fallback: any gateway_models route
(chat/embed/image) can now name a fallback_model, picked on /admin/gateway
from a select box of other configured public model names of the same kind
only (never an internal upstream model id). When a route fails after its own
retries are exhausted, the gateway retries once, automatically, against the
fallback's own provider/pricing/key, before any bytes reach the client
(including for a streaming response). One hop only, no chains or cycles;
self-reference and cross-kind fallbacks are rejected at write time.

AI gateway real upstream streaming and thinking-default control: stream:true
is now forwarded to the upstream and relayed to the client as real SSE
chunks (measured TTFT/inter-token latency) instead of a simulated split
response, and every chat/vision call explicitly disables model "thinking" by
default (admin-overridable via gateway_thinking), with per-dialect handling
for DeepSeek, OpenRouter, and Ollama.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TjdKTWgWpW2SMNW8SFqxz5
This commit is contained in:
retoor
2026-09-03 08:47:57 +02:00
co-authored by Claude Sonnet 5
parent 8ae3f628c7
commit a693a6f4d8
33 changed files with 1310 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/api/admin/gateway/models.py
+89
View File
@@ -135,6 +135,95 @@ def test_model_route_off_peak_requires_both_start_and_end(seeded_db):
assert only_end.json()["ok"] is False
def test_model_route_fallback_round_trip(seeded_db):
admin = admin_session(seeded_db)
backup = _unique_gateway("fb-backup")
primary = _unique_gateway("fb-primary")
created_backup = admin.post(
f"{BASE_URL}/admin/gateway/models",
json={"source_model": backup, "target_model": "vendor/backup", "kind": "chat"},
)
assert created_backup.status_code == 200, created_backup.text[:300]
created_primary = admin.post(
f"{BASE_URL}/admin/gateway/models",
json={
"source_model": primary,
"target_model": "vendor/primary",
"kind": "chat",
"fallback_model": backup,
},
)
assert created_primary.status_code == 200, created_primary.text[:300]
assert created_primary.json()["model"]["fallback_model"] == backup
listed = admin.get(f"{BASE_URL}/admin/gateway/models").json()
row = next(m for m in listed["models"] if m["source_model"] == primary)
assert row["fallback_model"] == backup
admin.delete(f"{BASE_URL}/admin/gateway/models/{primary}")
admin.delete(f"{BASE_URL}/admin/gateway/models/{backup}")
def test_model_route_fallback_must_reference_an_existing_route(seeded_db):
admin = admin_session(seeded_db)
source = _unique_gateway("fb-ghost")
response = admin.post(
f"{BASE_URL}/admin/gateway/models",
json={
"source_model": source,
"target_model": "vendor/z",
"kind": "chat",
"fallback_model": _unique_gateway("does-not-exist"),
},
)
assert response.status_code == 400
assert response.json()["ok"] is False
def test_model_route_fallback_must_be_the_same_kind(seeded_db):
admin = admin_session(seeded_db)
embed_route = _unique_gateway("fb-embed")
chat_route = _unique_gateway("fb-chat")
admin.post(
f"{BASE_URL}/admin/gateway/models",
json={"source_model": embed_route, "target_model": "vendor/e", "kind": "embed"},
)
response = admin.post(
f"{BASE_URL}/admin/gateway/models",
json={
"source_model": chat_route,
"target_model": "vendor/c",
"kind": "chat",
"fallback_model": embed_route,
},
)
assert response.status_code == 400
assert response.json()["ok"] is False
admin.delete(f"{BASE_URL}/admin/gateway/models/{embed_route}")
def test_model_route_fallback_rejects_self_reference(seeded_db):
admin = admin_session(seeded_db)
source = _unique_gateway("fb-self")
response = admin.post(
f"{BASE_URL}/admin/gateway/models",
json={
"source_model": source,
"target_model": "vendor/z",
"kind": "chat",
"fallback_model": source,
},
)
assert response.status_code == 400
assert response.json()["ok"] is False
def test_model_route_validation(seeded_db):
admin = admin_session(seeded_db)
missing_target = admin.post(
tests/api/projects/workspace.py
+38
View File
@@ -109,6 +109,17 @@ def test_workspace_quota_blocks_beyond_limit():
set_setting("workspace_max_per_user", "2")
def test_workspace_quota_does_not_apply_to_an_admin_owner(app_server, seeded_db):
admin = _seeded_user("alice_test")
set_setting("workspace_max_per_user", "1")
try:
run_async(provision.ensure(_project("p-admin-a"), admin))
run_async(provision.ensure(_project("p-admin-b"), admin))
assert provision.count_for_owner(admin["uid"]) == 2
finally:
set_setting("workspace_max_per_user", "2")
def test_tunnel_revives_rather_than_duplicates():
instance = _instance()
first = tunnels.create(instance, "web", 8080, OWNER)
@@ -344,6 +355,21 @@ def test_auto_delete_off_keeps_an_expired_workspace():
assert _reload(instance["uid"]) is not None
def test_advance_lifecycle_never_touches_an_admin_owned_workspace(app_server, seeded_db):
admin = _seeded_user("alice_test")
instance = _instance(
status="running", desired_state="running", workspace_owner_uid=admin["uid"]
)
store.update_instance(
instance["uid"], {"last_active_at": _idle_for(60 * 24 * 400)}
)
_lifecycle([_reload(instance["uid"])])
survivor = _reload(instance["uid"])
assert survivor is not None
assert survivor["desired_state"] == "running"
assert not survivor["idle_warned_at"]
def test_opening_a_workspace_publishes_the_editor_tunnel_automatically():
project = _project()
user = {"uid": OWNER, "username": "owner"}
@@ -1018,6 +1044,18 @@ def test_publish_tunnel_enforces_the_tunnel_quota():
set_setting("workspace_max_tunnels", "5")
def test_publish_tunnel_quota_does_not_apply_to_an_admin_owner(app_server, seeded_db):
admin = _seeded_user("alice_test")
set_setting("workspace_max_tunnels", "1")
try:
instance = _instance(workspace_owner_uid=admin["uid"])
provision.publish_tunnel(instance, "web", 3100, admin["uid"])
provision.publish_tunnel(instance, "api", 3101, admin["uid"])
assert tunnels.count_for_instance(instance["uid"]) == 2
finally:
set_setting("workspace_max_tunnels", "5")
def test_publish_tunnel_refuses_a_port_outside_the_valid_range():
instance = _instance()
with pytest.raises(WorkspaceError):
tests/unit/services/openai_gateway/gateway.py
+353 -6
View File
@@ -1,7 +1,6 @@
# retoor <retoor@molodetz.nl>
import json
import requests
from starlette.requests import Request
from tests.conftest import BASE_URL, run_async
from devplacepy.database import get_table, set_setting
@@ -20,6 +19,38 @@ class FakeResp_openai_gateway:
if self._payload is None:
raise ValueError("no json")
return self._payload
async def aread(self):
return self.content or self.text.encode()
async def aclose(self):
pass
async def aiter_lines(self):
payload = self._payload or {}
chunk_id = payload.get("id", "x")
model = payload.get("model", "")
try:
content = payload["choices"][0]["message"].get("content") or ""
except (KeyError, IndexError, TypeError):
content = ""
def frame(delta=None, finish=None, usage=None):
body = {"id": chunk_id, "object": "chat.completion.chunk", "model": model}
if usage is not None:
body["choices"] = []
body["usage"] = usage
else:
body["choices"] = [{"index": 0, "delta": delta or {}, "finish_reason": finish}]
return f"data: {json.dumps(body)}"
yield frame({"role": "assistant"})
for i in range(0, len(content), 5):
yield frame({"content": content[i : i + 5]})
yield frame({}, finish="stop")
if payload.get("usage"):
yield frame(usage=payload["usage"])
yield "data: [DONE]"
class FakeRequest:
def __init__(self, method, url, json_body):
self.method = method
@@ -33,7 +64,7 @@ class FakeClient_openai_gateway:
def build_request(self, method, url, headers=None, json=None, content=None):
return FakeRequest(method, url, json)
async def send(self, request):
async def send(self, request, stream=False):
self.calls.append((request.url, request.json_body))
body = request.json_body or {}
return FakeResp_openai_gateway(
@@ -256,7 +287,7 @@ class FakeEmbedClient_openai_gateway:
def build_request(self, method, url, headers=None, json=None, content=None):
return FakeRequest(method, url, json)
async def send(self, request):
async def send(self, request, stream=False):
self.calls.append((request.url, request.json_body))
body = request.json_body or {}
return FakeResp_openai_gateway(
@@ -344,7 +375,7 @@ class FakeImageClient_openai_gateway:
def build_request(self, method, url, headers=None, json=None, content=None):
return FakeRequest(method, url, json)
async def send(self, request):
async def send(self, request, stream=False):
self.calls.append((request.url, request.json_body))
body = request.json_body or {}
return FakeResp_openai_gateway(
@@ -615,6 +646,153 @@ def test_chat_unknown_model_falls_back_to_default(local_db, monkeypatch):
assert rt._client.calls[-1][1]["model"] == "deepseek-v4-flash"
class FakeFallbackClient_openai_gateway:
def __init__(self, *a, **k):
self.calls = []
def build_request(self, method, url, headers=None, json=None, content=None):
return FakeRequest(method, url, json)
async def send(self, request, stream=False):
self.calls.append((request.url, request.json_body))
body = request.json_body or {}
model = body.get("model")
if model == "primary-target":
return FakeResp_openai_gateway(status=500, payload={"error": "boom"})
return FakeResp_openai_gateway(
payload={
"id": "x",
"model": model,
"choices": [{"message": {"content": "fallback ok"}}],
}
)
async def aclose(self):
pass
class FakeAlwaysFailClient_openai_gateway:
def __init__(self, *a, **k):
self.calls = []
def build_request(self, method, url, headers=None, json=None, content=None):
return FakeRequest(method, url, json)
async def send(self, request, stream=False):
self.calls.append((request.url, request.json_body))
return FakeResp_openai_gateway(status=500, payload={"error": "boom"})
async def aclose(self):
pass
def test_chat_falls_back_when_the_primary_model_fails(local_db, monkeypatch):
from devplacepy.services.openai_gateway import routing
monkeypatch.setattr(gwmod.httpx, "AsyncClient", FakeFallbackClient_openai_gateway)
routing.model_store.set(
routing.ModelRouteIn(source_model="fb-backup-route", target_model="backup-target")
)
routing.model_store.set(
routing.ModelRouteIn(
source_model="fb-primary-route",
target_model="primary-target",
fallback_model="fb-backup-route",
)
)
try:
svc = GatewayService()
cfg = svc.effective_config()
cfg["gateway_vision_enabled"] = False
cfg["gateway_max_retries"] = 0
rt = svc.runtime()
response = run_async(
rt.handle_chat(
{"model": "fb-primary-route", "messages": [{"role": "user", "content": "hi"}]},
cfg,
("guest", "fallback_chat_success"),
"test",
"default",
)
)
assert response.status_code == 200
assert rt._client.calls[0][1]["model"] == "primary-target"
assert rt._client.calls[-1][1]["model"] == "backup-target"
row = get_table("gateway_usage_ledger").find_one(owner_id="fallback_chat_success")
assert row is not None
assert row["requested_model"] == "fb-primary-route"
assert row["model"] == "backup-target"
assert row["success"] == 1
finally:
routing.model_store.remove("fb-primary-route")
routing.model_store.remove("fb-backup-route")
def test_chat_returns_the_fallback_failure_when_both_models_fail(local_db, monkeypatch):
from devplacepy.services.openai_gateway import routing
monkeypatch.setattr(gwmod.httpx, "AsyncClient", FakeAlwaysFailClient_openai_gateway)
routing.model_store.set(
routing.ModelRouteIn(source_model="fb-backup-route2", target_model="backup-target2")
)
routing.model_store.set(
routing.ModelRouteIn(
source_model="fb-primary-route2",
target_model="primary-target2",
fallback_model="fb-backup-route2",
)
)
try:
svc = GatewayService()
cfg = svc.effective_config()
cfg["gateway_vision_enabled"] = False
cfg["gateway_max_retries"] = 0
rt = svc.runtime()
response = run_async(
rt.handle_chat(
{"model": "fb-primary-route2", "messages": [{"role": "user", "content": "hi"}]},
cfg,
("guest", "fallback_chat_both_fail"),
"test",
"default",
)
)
assert response.status_code == 500
assert len(rt._client.calls) == 2
assert rt._client.calls[0][1]["model"] == "primary-target2"
assert rt._client.calls[1][1]["model"] == "backup-target2"
finally:
routing.model_store.remove("fb-primary-route2")
routing.model_store.remove("fb-backup-route2")
def test_chat_without_a_fallback_never_makes_a_second_call(local_db, monkeypatch):
from devplacepy.services.openai_gateway import routing
monkeypatch.setattr(gwmod.httpx, "AsyncClient", FakeAlwaysFailClient_openai_gateway)
routing.model_store.set(
routing.ModelRouteIn(source_model="fb-no-fallback", target_model="no-fallback-target")
)
try:
svc = GatewayService()
cfg = svc.effective_config()
cfg["gateway_vision_enabled"] = False
cfg["gateway_max_retries"] = 0
rt = svc.runtime()
run_async(
rt.handle_chat(
{"model": "fb-no-fallback", "messages": [{"role": "user", "content": "hi"}]},
cfg,
("guest", "no_fallback_configured"),
"test",
"default",
)
)
assert len(rt._client.calls) == 1
finally:
routing.model_store.remove("fb-no-fallback")
def test_embeddings_unknown_model_falls_back_to_default(local_db, monkeypatch):
monkeypatch.setattr(gwmod.httpx, "AsyncClient", FakeEmbedClient_openai_gateway)
svc = GatewayService()
@@ -656,7 +834,7 @@ def test_images_unknown_model_falls_back_to_default(local_db, monkeypatch):
class FakeClientWithUsage_openai_gateway(FakeClient_openai_gateway):
async def send(self, request):
async def send(self, request, stream=False):
self.calls.append((request.url, request.json_body))
body = request.json_body or {}
return FakeResp_openai_gateway(
@@ -669,7 +847,7 @@ class FakeClientWithUsage_openai_gateway(FakeClient_openai_gateway):
)
def test_stream_options_stripped_from_upstream(local_db, monkeypatch):
def test_stream_forwarded_to_upstream_with_forced_include_usage(local_db, monkeypatch):
monkeypatch.setattr(gwmod.httpx, "AsyncClient", FakeClient_openai_gateway)
svc = GatewayService()
cfg = svc.effective_config()
@@ -679,6 +857,31 @@ def test_stream_options_stripped_from_upstream(local_db, monkeypatch):
{
"messages": [{"role": "user", "content": "hi"}],
"stream": True,
"stream_options": {"include_usage": False},
},
cfg,
("guest", "test"),
"test",
"default",
)
)
upstream_payload = rt._client.calls[-1][1]
assert upstream_payload["stream"] is True
assert upstream_payload["stream_options"] == {"include_usage": True}
assert upstream_payload["thinking"] == {"type": "disabled"}
assert "think" not in upstream_payload
assert "reasoning" not in upstream_payload
def test_non_stream_has_no_stream_options_upstream(local_db, monkeypatch):
monkeypatch.setattr(gwmod.httpx, "AsyncClient", FakeClient_openai_gateway)
svc = GatewayService()
cfg = svc.effective_config()
rt = svc.runtime()
run_async(
rt.handle_chat(
{
"messages": [{"role": "user", "content": "hi"}],
"stream_options": {"include_usage": True},
},
cfg,
@@ -692,6 +895,66 @@ def test_stream_options_stripped_from_upstream(local_db, monkeypatch):
assert upstream_payload["stream"] is False
def test_thinking_disabled_by_default(local_db, monkeypatch):
monkeypatch.setattr(gwmod.httpx, "AsyncClient", FakeClient_openai_gateway)
svc = GatewayService()
cfg = svc.effective_config()
rt = svc.runtime()
run_async(
rt.handle_chat(
{"messages": [{"role": "user", "content": "hi"}]},
cfg,
("guest", "test"),
"test",
"default",
)
)
body = rt._client.calls[-1][1]
assert body["thinking"] == {"type": "disabled"}
def test_client_think_true_enables_upstream_thinking(local_db, monkeypatch):
monkeypatch.setattr(gwmod.httpx, "AsyncClient", FakeClient_openai_gateway)
svc = GatewayService()
cfg = svc.effective_config()
rt = svc.runtime()
run_async(
rt.handle_chat(
{
"messages": [{"role": "user", "content": "hi"}],
"think": True,
},
cfg,
("guest", "test"),
"test",
"default",
)
)
body = rt._client.calls[-1][1]
assert body["thinking"] == {"type": "enabled"}
assert "think" not in body
def test_openrouter_upstream_gets_reasoning_none(local_db, monkeypatch):
monkeypatch.setattr(gwmod.httpx, "AsyncClient", FakeClient_openai_gateway)
svc = GatewayService()
cfg = svc.effective_config()
cfg["gateway_upstream_url"] = "https://openrouter.ai/api/v1/chat/completions"
rt = svc.runtime()
run_async(
rt.handle_chat(
{"messages": [{"role": "user", "content": "hi"}]},
cfg,
("guest", "test"),
"test",
"default",
)
)
body = rt._client.calls[-1][1]
assert body["reasoning"] == {"effort": "none"}
assert "thinking" not in body
def test_include_usage_emits_usage_chunk(local_db, monkeypatch):
monkeypatch.setattr(gwmod.httpx, "AsyncClient", FakeClientWithUsage_openai_gateway)
svc = GatewayService()
@@ -749,6 +1012,90 @@ def test_no_usage_chunk_without_include_usage(local_db, monkeypatch):
assert "[DONE]" in body
def test_stream_headers_carry_no_cost_or_token_data(local_db, monkeypatch):
monkeypatch.setattr(gwmod.httpx, "AsyncClient", FakeClientWithUsage_openai_gateway)
svc = GatewayService()
cfg = svc.effective_config()
rt = svc.runtime()
resp = run_async(
rt.handle_chat(
{"messages": [{"role": "user", "content": "hi"}], "stream": True},
cfg,
("guest", "test"),
"test",
"my-app",
)
)
assert resp.headers["X-Gateway-Model"] == cfg["gateway_model"]
assert resp.headers["X-Gateway-Backend"] == "chat"
assert resp.headers["X-App-Reference"] == "my-app"
assert "x-gateway-cost-usd" not in {k.lower() for k in resp.headers}
assert "x-gateway-total-tokens" not in {k.lower() for k in resp.headers}
def test_stream_records_ttft_and_inter_token_in_ledger(local_db, monkeypatch):
monkeypatch.setattr(gwmod.httpx, "AsyncClient", FakeClientWithUsage_openai_gateway)
svc = GatewayService()
cfg = svc.effective_config()
rt = svc.runtime()
resp = run_async(
rt.handle_chat(
{
"messages": [{"role": "user", "content": "hi"}],
"stream": True,
"stream_options": {"include_usage": True},
},
cfg,
("guest", "ttft_probe"),
"test",
"default",
)
)
async def drain():
async for _ in resp.body_iterator:
pass
run_async(drain())
row = get_table("gateway_usage_ledger").find_one(owner_id="ttft_probe")
assert row is not None
assert row["backend"] == "chat"
assert row["stream_requested"] == 1
assert row["success"] == 1
assert row["total_tokens"] == 10
assert row["ttft_ms"] is not None and row["ttft_ms"] >= 0.0
assert row["inter_token_ms"] is not None and row["inter_token_ms"] >= 0.0
def test_stream_client_disconnect_records_failure_and_closes_upstream(local_db, monkeypatch):
monkeypatch.setattr(gwmod.httpx, "AsyncClient", FakeClient_openai_gateway)
svc = GatewayService()
cfg = svc.effective_config()
rt = svc.runtime()
resp = run_async(
rt.handle_chat(
{"messages": [{"role": "user", "content": "hi"}], "stream": True},
cfg,
("guest", "disconnect_probe"),
"test",
"default",
)
)
async def partial_then_close():
agen = resp.body_iterator
await agen.__anext__()
await agen.aclose()
run_async(partial_then_close())
row = get_table("gateway_usage_ledger").find_one(owner_id="disconnect_probe")
assert row is not None
assert row["success"] == 0
assert row["error_category"] == "client_disconnected"
def test_models_endpoint_publishes_molodetz(local_db):
from devplacepy.services.openai_gateway import routing
tests/unit/services/openai_gateway/routing.py
+85
View File
@@ -55,6 +55,91 @@ def test_chat_route_overlay_and_economy(local_db):
_cleanup(["utor"], ["ut-chat"])
def test_resolve_fallback_returns_none_without_a_configured_fallback(local_db):
r.model_store.set(r.ModelRouteIn(source_model="fb-primary", target_model="v/a"))
try:
assert r.resolve_fallback("fb-primary", "chat") is None
finally:
_cleanup([], ["fb-primary"])
def test_resolve_fallback_returns_none_for_an_unknown_source(local_db):
assert r.resolve_fallback("ghost-fb-source", "chat") is None
def test_resolve_fallback_rejects_self_reference(local_db):
with pytest.raises(ValidationError):
r.ModelRouteIn(
source_model="fb-self", target_model="v/a", fallback_model="fb-self"
)
def test_resolve_fallback_returns_the_configured_route(local_db):
r.model_store.set(
r.ModelRouteIn(
source_model="fb-backup", target_model="v/backup", kind="chat"
)
)
r.model_store.set(
r.ModelRouteIn(
source_model="fb-primary",
target_model="v/primary",
kind="chat",
fallback_model="fb-backup",
)
)
try:
fallback = r.resolve_fallback("fb-primary", "chat")
assert fallback is not None
assert fallback.source_model == "fb-backup"
assert fallback.target_model == "v/backup"
finally:
_cleanup([], ["fb-primary", "fb-backup"])
def test_resolve_fallback_ignores_an_inactive_fallback_route(local_db):
r.model_store.set(
r.ModelRouteIn(
source_model="fb-backup-off",
target_model="v/backup",
kind="chat",
is_active=False,
)
)
r.model_store.set(
r.ModelRouteIn(
source_model="fb-primary-2",
target_model="v/primary",
kind="chat",
fallback_model="fb-backup-off",
)
)
try:
assert r.resolve_fallback("fb-primary-2", "chat") is None
finally:
_cleanup([], ["fb-primary-2", "fb-backup-off"])
def test_resolve_fallback_ignores_a_fallback_of_a_different_kind(local_db):
r.model_store.set(
r.ModelRouteIn(
source_model="fb-backup-embed", target_model="v/e", kind="embed"
)
)
r.model_store.set(
r.ModelRouteIn(
source_model="fb-primary-3",
target_model="v/primary",
kind="chat",
fallback_model="fb-backup-embed",
)
)
try:
assert r.resolve_fallback("fb-primary-3", "chat") is None
finally:
_cleanup([], ["fb-primary-3", "fb-backup-embed"])
def test_vision_merge_overlay(local_db):
r.provider_store.set(
r.ProviderIn(
tests/unit/services/openai_gateway/thinking.py
+103
View File
@@ -0,0 +1,103 @@
# retoor <retoor@molodetz.nl>
from devplacepy.services.openai_gateway.thinking import (
apply_thinking,
client_thinking_enabled,
thinking_dialect,
)
def test_dialect_from_upstream_url():
assert thinking_dialect("https://api.deepseek.com/chat/completions") == "deepseek"
assert thinking_dialect("https://openrouter.ai/api/v1/chat/completions") == "openrouter"
assert thinking_dialect("http://127.0.0.1:11434/api/chat") == "ollama"
assert thinking_dialect("http://ollama.local/v1/chat/completions") == "ollama"
assert thinking_dialect("https://unknown.example/v1/chat/completions") == "deepseek"
def test_client_intent_unspecified():
assert client_thinking_enabled({}) is None
assert client_thinking_enabled({"messages": []}) is None
assert client_thinking_enabled(None) is None
def test_client_intent_disable():
assert client_thinking_enabled({"think": False}) is False
assert client_thinking_enabled({"think": "false"}) is False
assert client_thinking_enabled({"thinking": {"type": "disabled"}}) is False
assert client_thinking_enabled({"reasoning": {"effort": "none"}}) is False
assert client_thinking_enabled({"reasoning": {"enabled": False}}) is False
assert client_thinking_enabled({"enable_thinking": False}) is False
assert client_thinking_enabled(
{"chat_template_kwargs": {"enable_thinking": False}}
) is False
def test_client_intent_enable():
assert client_thinking_enabled({"think": True}) is True
assert client_thinking_enabled({"think": "high"}) is True
assert client_thinking_enabled({"thinking": {"type": "enabled"}}) is True
assert client_thinking_enabled({"reasoning": {"effort": "low"}}) is True
assert client_thinking_enabled({"reasoning_effort": "max"}) is True
def test_default_disables_deepseek_thinking():
payload = apply_thinking(
{"messages": [{"role": "user", "content": "hi"}]},
"https://api.deepseek.com/chat/completions",
)
assert payload["thinking"] == {"type": "disabled"}
assert "think" not in payload
assert "reasoning" not in payload
assert "reasoning_effort" not in payload
def test_default_disables_openrouter_and_ollama():
openrouter = apply_thinking({}, "https://openrouter.ai/api/v1/chat/completions")
assert openrouter["reasoning"] == {"effort": "none"}
assert "thinking" not in openrouter
ollama = apply_thinking({}, "http://127.0.0.1:11434/api/chat")
assert ollama["think"] is False
assert "thinking" not in ollama
def test_client_can_enable_thinking():
payload = apply_thinking(
{"think": True, "messages": []},
"https://api.deepseek.com/chat/completions",
)
assert payload["thinking"] == {"type": "enabled"}
assert "think" not in payload
def test_client_disable_beats_admin_default_on():
payload = apply_thinking(
{"thinking": {"type": "disabled"}},
"https://api.deepseek.com/chat/completions",
default_enabled=True,
)
assert payload["thinking"] == {"type": "disabled"}
def test_admin_default_on_when_client_silent():
payload = apply_thinking(
{"messages": []},
"https://api.deepseek.com/chat/completions",
default_enabled=True,
)
assert payload["thinking"] == {"type": "enabled"}
def test_strips_conflicting_client_fields():
payload = apply_thinking(
{
"think": False,
"reasoning_effort": "high",
"chat_template_kwargs": {"enable_thinking": True, "other": 1},
},
"https://api.deepseek.com/chat/completions",
)
assert payload["thinking"] == {"type": "disabled"}
assert "think" not in payload
assert "reasoning_effort" not in payload
assert payload["chat_template_kwargs"] == {"other": 1}