mirror of
https://github.com/unslothai/unsloth.git
synced 2026-07-09 15:58:41 +00:00
* fix(studio/llama_cpp): disable trust_env on the loopback health probe _wait_for_health() polls http://127.0.0.1:<port>/health with the default httpx trust_env=True, so an ambient HTTP(S)_PROXY in the environment is applied to the loopback request. A proxy that returns 503 for 127.0.0.1 makes every probe fail, so the loop runs until timeout and Studio load hangs (trust_env=False returns 200 immediately). Pass trust_env=False so the local readiness probe never goes through a proxy. This mirrors the existing trust_env=False handling in the sibling llama_http / external_provider HTTP clients. * test(offline_gguf_cache): accept trust_env kwarg in fake_get mock _wait_for_health now calls httpx.get(..., trust_env=False); update the retry test's fake_get to accept the kwarg so it doesn't raise TypeError. * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * fix(studio/llama_cpp): bypass proxies for loopback clients * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * fix(studio/routes): bypass proxies for llama streams * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --------- Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> Co-authored-by: wasimysaid <wasimysdev@gmail.com>
This commit is contained in:
Tai An
GitHub
parent
b72a8c4263
commit
7337729e57
10 changed files with 112 additions and 45 deletions
|
|
@ -7273,7 +7273,13 @@ class LlamaCppBackend:
|
|||
url = f"{self.base_url}/completion"
|
||||
payload = {"prompt": "Hi", "n_predict": 4, "temperature": 0.0, "stream": False}
|
||||
try:
|
||||
resp = httpx.post(url, json = payload, timeout = timeout, headers = self._auth_headers)
|
||||
resp = httpx.post(
|
||||
url,
|
||||
json = payload,
|
||||
timeout = timeout,
|
||||
headers = self._auth_headers,
|
||||
trust_env = False,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.debug(f"MTP decode probe failed: {e}")
|
||||
return False
|
||||
|
|
@ -7425,7 +7431,10 @@ class LlamaCppBackend:
|
|||
return False
|
||||
|
||||
try:
|
||||
resp = httpx.get(url, timeout = 2.0)
|
||||
# trust_env=False: never route the loopback health probe through
|
||||
# an ambient HTTP(S)_PROXY. A proxy that 503s for 127.0.0.1 makes
|
||||
# the probe loop until timeout and hangs Studio load.
|
||||
resp = httpx.get(url, timeout = 2.0, trust_env = False)
|
||||
if resp.status_code == 200:
|
||||
return True
|
||||
except (
|
||||
|
|
@ -7472,7 +7481,7 @@ class LlamaCppBackend:
|
|||
"""
|
||||
url = f"{self.base_url}/props"
|
||||
try:
|
||||
resp = httpx.get(url, timeout = 5.0)
|
||||
resp = httpx.get(url, timeout = 5.0, trust_env = False)
|
||||
if resp.status_code != 200:
|
||||
return None
|
||||
settings = resp.json().get("default_generation_settings") or {}
|
||||
|
|
@ -7552,7 +7561,9 @@ class LlamaCppBackend:
|
|||
which differ only in how they parse the SSE body."""
|
||||
stream_timeout = httpx.Timeout(connect = 10, read = 0.5, write = 10, pool = 10)
|
||||
with httpx.Client(
|
||||
timeout = stream_timeout, limits = httpx.Limits(max_keepalive_connections = 0)
|
||||
timeout = stream_timeout,
|
||||
limits = httpx.Limits(max_keepalive_connections = 0),
|
||||
trust_env = False,
|
||||
) as client:
|
||||
first_token_deadline = time.monotonic() + _DEFAULT_FIRST_TOKEN_TIMEOUT_S
|
||||
with self._stream_with_retry(
|
||||
|
|
@ -9044,7 +9055,7 @@ class LlamaCppBackend:
|
|||
system_text = _block_text(system)
|
||||
|
||||
try:
|
||||
with httpx.Client(timeout = 10, headers = self._auth_headers) as client:
|
||||
with httpx.Client(timeout = 10, headers = self._auth_headers, trust_env = False) as client:
|
||||
|
||||
def _tokenize(text: str) -> int:
|
||||
r = client.post(
|
||||
|
|
@ -9160,7 +9171,7 @@ class LlamaCppBackend:
|
|||
"""Codec name on match, None on non-audio, raises on transport/JSON errors."""
|
||||
if not self.is_loaded:
|
||||
return None
|
||||
with httpx.Client(timeout = 10, headers = self._auth_headers) as client:
|
||||
with httpx.Client(timeout = 10, headers = self._auth_headers, trust_env = False) as client:
|
||||
|
||||
def _detok(tid: int) -> str:
|
||||
# Non-200 means "marker not in vocab" -- keep probing.
|
||||
|
|
@ -9275,7 +9286,9 @@ class LlamaCppBackend:
|
|||
payload["n_probs"] = 1
|
||||
|
||||
with httpx.Client(
|
||||
timeout = httpx.Timeout(300, connect = 10), headers = self._auth_headers
|
||||
timeout = httpx.Timeout(300, connect = 10),
|
||||
headers = self._auth_headers,
|
||||
trust_env = False,
|
||||
) as client:
|
||||
resp = client.post(f"{self.base_url}/completion", json = payload)
|
||||
if resp.status_code != 200:
|
||||
|
|
|
|||
|
|
@ -22,11 +22,7 @@ _LIMITS = httpx.Limits(max_connections = 64, max_keepalive_connections = 32)
|
|||
|
||||
|
||||
def _new_client() -> httpx.AsyncClient:
|
||||
try:
|
||||
return httpx.AsyncClient(limits = _LIMITS)
|
||||
except Exception:
|
||||
# Mirror external_provider: an unsupported env proxy scheme can raise.
|
||||
return httpx.AsyncClient(limits = _LIMITS, trust_env = False)
|
||||
return httpx.AsyncClient(limits = _LIMITS, trust_env = False)
|
||||
|
||||
|
||||
# One client per running event loop: an httpx client binds its transport to the
|
||||
|
|
|
|||
|
|
@ -6723,7 +6723,10 @@ async def openai_completions(request: Request, current_subject: str = Depends(ge
|
|||
# honor stream_options.include_usage per event, while keeping SSE
|
||||
# framing and token bytes intact.
|
||||
_include_usage = bool((body.get("stream_options") or {}).get("include_usage"))
|
||||
client = httpx.AsyncClient(timeout = _llama_streaming_generation_timeout())
|
||||
client = httpx.AsyncClient(
|
||||
timeout = _llama_streaming_generation_timeout(),
|
||||
trust_env = False,
|
||||
)
|
||||
resp = None
|
||||
bytes_iter = None
|
||||
disconnect_event = threading.Event()
|
||||
|
|
@ -7798,7 +7801,10 @@ async def _responses_stream(
|
|||
# `async with`, explicit aclose of lines_iter BEFORE resp / client so
|
||||
# the innermost httpcore byte stream is finalised in this task (not via
|
||||
# the asyncgen GC in a sibling task).
|
||||
client = httpx.AsyncClient(timeout = _llama_streaming_generation_timeout())
|
||||
client = httpx.AsyncClient(
|
||||
timeout = _llama_streaming_generation_timeout(),
|
||||
trust_env = False,
|
||||
)
|
||||
resp = None
|
||||
lines_iter = None
|
||||
disconnect_watcher = None
|
||||
|
|
@ -9308,6 +9314,7 @@ async def _anthropic_passthrough_stream(
|
|||
client = httpx.AsyncClient(
|
||||
timeout = _llama_streaming_generation_timeout(),
|
||||
limits = httpx.Limits(max_keepalive_connections = 0),
|
||||
trust_env = False,
|
||||
)
|
||||
resp = None
|
||||
lines_iter = None
|
||||
|
|
@ -9834,6 +9841,7 @@ async def _openai_passthrough_stream(
|
|||
client = httpx.AsyncClient(
|
||||
timeout = _llama_streaming_generation_timeout(),
|
||||
limits = httpx.Limits(max_keepalive_connections = 0),
|
||||
trust_env = False,
|
||||
)
|
||||
resp = None
|
||||
_truncate_budget = (
|
||||
|
|
|
|||
|
|
@ -57,6 +57,15 @@ def test_media_type_and_status():
|
|||
assert err.status_code == 503
|
||||
|
||||
|
||||
def test_pooled_client_disables_proxy_env():
|
||||
async def _scenario():
|
||||
client = llama_http.nonstreaming_client()
|
||||
assert client.trust_env is False
|
||||
await llama_http.aclose()
|
||||
|
||||
asyncio.run(_scenario())
|
||||
|
||||
|
||||
def test_pooled_client_reused_within_loop_and_recreated_after_close():
|
||||
async def _scenario():
|
||||
a = llama_http.nonstreaming_client()
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@ must be logged and skipped, not fatal. Fakes only.
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
import queue
|
||||
import sys
|
||||
import threading
|
||||
|
|
@ -89,3 +90,31 @@ def test_dispatcher_survives_mailbox_put_error():
|
|||
o._dispatcher_stop.set()
|
||||
t.join(timeout = 5)
|
||||
assert not t.is_alive()
|
||||
|
||||
|
||||
def test_route_llama_streaming_async_clients_disable_proxy_env():
|
||||
"""Local llama-server streaming proxies must ignore ambient HTTP_PROXY."""
|
||||
source = (Path(__file__).resolve().parent.parent / "routes" / "inference.py").read_text(
|
||||
encoding = "utf-8"
|
||||
)
|
||||
tree = ast.parse(source)
|
||||
calls = []
|
||||
for node in ast.walk(tree):
|
||||
if not isinstance(node, ast.Call):
|
||||
continue
|
||||
func = node.func
|
||||
if not (
|
||||
isinstance(func, ast.Attribute)
|
||||
and func.attr == "AsyncClient"
|
||||
and isinstance(func.value, ast.Name)
|
||||
and func.value.id == "httpx"
|
||||
):
|
||||
continue
|
||||
calls.append(node)
|
||||
|
||||
assert len(calls) == 4
|
||||
for call in calls:
|
||||
assert any(
|
||||
kw.arg == "trust_env" and isinstance(kw.value, ast.Constant) and kw.value.value is False
|
||||
for kw in call.keywords
|
||||
), f"httpx.AsyncClient at line {call.lineno} must set trust_env=False"
|
||||
|
|
|
|||
|
|
@ -113,8 +113,14 @@ def _stub_props(
|
|||
body = None,
|
||||
exc = None,
|
||||
):
|
||||
def fake_get(url, timeout = None):
|
||||
def fake_get(
|
||||
url,
|
||||
timeout = None,
|
||||
trust_env = None,
|
||||
):
|
||||
assert url.endswith("/props")
|
||||
|
||||
assert trust_env is False
|
||||
if exc is not None:
|
||||
raise exc
|
||||
return _FakeResponse(status_code, body)
|
||||
|
|
|
|||
|
|
@ -951,7 +951,11 @@ class TestWaitForHealthRetriesOnReadError:
|
|||
|
||||
calls = {"n": 0}
|
||||
|
||||
def fake_get(url, timeout = None):
|
||||
def fake_get(
|
||||
url,
|
||||
timeout = None,
|
||||
trust_env = None,
|
||||
):
|
||||
calls["n"] += 1
|
||||
if calls["n"] == 1:
|
||||
raise httpx.ReadError("WinError 10054")
|
||||
|
|
|
|||
|
|
@ -554,6 +554,8 @@ def test_probe_mtp_decode_uses_api_key_auth(monkeypatch):
|
|||
backend._api_key = "secret"
|
||||
backend._probe_mtp_decode(timeout = 1.0)
|
||||
assert captured["headers"] == {"Authorization": "Bearer secret"}
|
||||
|
||||
assert captured["trust_env"] is False
|
||||
backend._api_key = None
|
||||
backend._probe_mtp_decode(timeout = 1.0)
|
||||
assert captured["headers"] is None
|
||||
|
|
|
|||
|
|
@ -3555,9 +3555,9 @@ const DiffusionCanvas: FC = () => {
|
|||
|
||||
/**
|
||||
* AssistantMessage handles the display and inline-editing of AI responses.
|
||||
*
|
||||
* It utilizes a "Tagged Text" system (<THINK> and <TOOL> tags) to allow users
|
||||
* to edit structured reasoning and tool outputs within a plain-text textarea
|
||||
*
|
||||
* It utilizes a "Tagged Text" system (<THINK> and <TOOL> tags) to allow users
|
||||
* to edit structured reasoning and tool outputs within a plain-text textarea
|
||||
* while preserving the underlying data schema and tool-call metadata.
|
||||
*/
|
||||
const AssistantMessage: FC = () => {
|
||||
|
|
@ -3565,7 +3565,7 @@ const AssistantMessage: FC = () => {
|
|||
const messageId = useAuiState(({ message }) => message.id);
|
||||
const messageContent = useAuiState(({ message }) => message.content);
|
||||
const incognito = useChatRuntimeStore((s) => s.incognito);
|
||||
|
||||
|
||||
// Use global store for editing state to ensure a single source of truth
|
||||
const editingId = useChatRuntimeStore((s) => s.editingMessageId);
|
||||
const setEditingId = useChatRuntimeStore((s) => s.setEditingMessageId);
|
||||
|
|
@ -3588,9 +3588,9 @@ const AssistantMessage: FC = () => {
|
|||
|
||||
const handleSave = async () => {
|
||||
const finalText = textareaRef.current?.value || "";
|
||||
|
||||
|
||||
// Prioritize the specific thread item ID, then fallback to the global active thread ID
|
||||
const remoteId = aui.threadListItem().getState().remoteId
|
||||
const remoteId = aui.threadListItem().getState().remoteId
|
||||
|| useChatRuntimeStore.getState().activeThreadId;
|
||||
|
||||
if (!remoteId || remoteId === "" || remoteId === "/") {
|
||||
|
|
@ -3601,9 +3601,9 @@ const AssistantMessage: FC = () => {
|
|||
|
||||
try {
|
||||
await updateThreadMessage({
|
||||
thread: {
|
||||
export: () => aui.thread().export(),
|
||||
import: (data) => aui.thread().import(data)
|
||||
thread: {
|
||||
export: () => aui.thread().export(),
|
||||
import: (data) => aui.thread().import(data)
|
||||
},
|
||||
messageId,
|
||||
remoteId,
|
||||
|
|
@ -3626,14 +3626,14 @@ const AssistantMessage: FC = () => {
|
|||
<div className="aui-assistant-message-content wrap-break-word min-w-0 text-[#0d0d0d] dark:text-foreground leading-relaxed">
|
||||
{isEditing ? (
|
||||
<div className="flex flex-col gap-2 w-full">
|
||||
<textarea
|
||||
<textarea
|
||||
ref={textareaRef}
|
||||
defaultValue={extractTaggedText(messageContent)}
|
||||
className="w-full p-3 rounded-xl bg-muted border border-border text-foreground focus:ring-2 focus:ring-primary outline-none overflow-y-auto resize-none font-mono text-sm max-h-[70vh]"
|
||||
className="w-full p-3 rounded-xl bg-muted border border-border text-foreground focus:ring-2 focus:ring-primary outline-none overflow-y-auto resize-none font-mono text-sm max-h-[70vh]"
|
||||
autoFocus
|
||||
onInput={adjustHeight}
|
||||
onInput={adjustHeight}
|
||||
onKeyDown={(e) => {
|
||||
e.stopPropagation();
|
||||
e.stopPropagation();
|
||||
if (e.key === 'Enter' && (e.ctrlKey || e.metaKey)) {
|
||||
handleSave();
|
||||
}
|
||||
|
|
@ -3652,10 +3652,10 @@ const AssistantMessage: FC = () => {
|
|||
<GeneratingIndicator />
|
||||
<CancelledIndicator />
|
||||
<DiffusionCanvas />
|
||||
|
||||
{/*
|
||||
We use the standard MessagePrimitive.Parts. This ensures that
|
||||
edited messages maintain the same professional styling,
|
||||
|
||||
{/*
|
||||
We use the standard MessagePrimitive.Parts. This ensures that
|
||||
edited messages maintain the same professional styling,
|
||||
Markdown rendering, and tool-call components as original responses.
|
||||
*/}
|
||||
<MessagePrimitive.Parts
|
||||
|
|
|
|||
|
|
@ -9,7 +9,7 @@ type ThreadImportExport = {
|
|||
type ContentPart = { type: "text" | "reasoning" | "tool"; text: string };
|
||||
|
||||
/**
|
||||
* Extracts only the editable text and reasoning from a message,
|
||||
* Extracts only the editable text and reasoning from a message,
|
||||
* ignoring structured parts like tool calls that cannot be edited as plain text.
|
||||
*/
|
||||
export function extractTaggedText(content: any): string {
|
||||
|
|
@ -23,16 +23,16 @@ export function extractTaggedText(content: any): string {
|
|||
.map((part: any) => {
|
||||
if (typeof part === 'string') return part;
|
||||
if (!part) return "";
|
||||
|
||||
|
||||
// Only extract text from 'text' or 'reasoning' parts.
|
||||
// Tool calls/responses are ignored here so they aren't accidentally
|
||||
// Tool calls/responses are ignored here so they aren't accidentally
|
||||
// deleted or corrupted by the user in the textarea.
|
||||
const text = part.text || part.content || "";
|
||||
if (!text) return "";
|
||||
|
||||
|
||||
switch (part.type) {
|
||||
case 'reasoning':
|
||||
// Trim the text first so we don't accumulate newlines
|
||||
case 'reasoning':
|
||||
// Trim the text first so we don't accumulate newlines
|
||||
// around the tags on every save.
|
||||
return `${open}THINK${close}\n${text.trim()}\n${open}/THINK${close}`;
|
||||
case 'text':
|
||||
|
|
@ -57,12 +57,12 @@ function parseTaggedTextToContent(text: string): ContentPart[] {
|
|||
const index = match.index;
|
||||
|
||||
if (index > lastIndex) {
|
||||
// Trim the extracted content to remove any leading/trailing
|
||||
// Trim the extracted content to remove any leading/trailing
|
||||
// newlines created by the tag wrapping process.
|
||||
const content = text.substring(lastIndex, index).trim();
|
||||
if (content) parts.push({ type: currentType, text: content });
|
||||
}
|
||||
|
||||
|
||||
currentType = fullTag.startsWith("</") ? "text" : (tagName === "THINK" ? "reasoning" : "tool");
|
||||
lastIndex = index + fullTag.length;
|
||||
}
|
||||
|
|
@ -91,7 +91,7 @@ export async function updateThreadMessage(args: {
|
|||
throw new Error(`Message with ID ${messageId} not found in thread.`);
|
||||
}
|
||||
|
||||
const { parentId: originalParentId } = targetMessageEntry;
|
||||
const { parentId: originalParentId } = targetMessageEntry;
|
||||
const { createdAt: originalCreatedAt } = targetMessageEntry.message;
|
||||
|
||||
const updatedMessages = currentExport.messages.map((m) => {
|
||||
|
|
@ -101,18 +101,18 @@ export async function updateThreadMessage(args: {
|
|||
let finalContent: any[] = [];
|
||||
|
||||
if (Array.isArray(originalContent)) {
|
||||
const firstEditableIndex = originalContent.findIndex((part: any) =>
|
||||
const firstEditableIndex = originalContent.findIndex((part: any) =>
|
||||
part.type === 'text' || part.type === 'reasoning'
|
||||
);
|
||||
|
||||
if (firstEditableIndex === -1) {
|
||||
const nonEditableParts = originalContent.filter((part: any) =>
|
||||
const nonEditableParts = originalContent.filter((part: any) =>
|
||||
part.type !== 'text' && part.type !== 'reasoning'
|
||||
);
|
||||
finalContent = [...parsedEditableContent, ...nonEditableParts];
|
||||
} else {
|
||||
const before = originalContent.slice(0, firstEditableIndex);
|
||||
const after = originalContent.slice(firstEditableIndex + 1).filter((part: any) =>
|
||||
const after = originalContent.slice(firstEditableIndex + 1).filter((part: any) =>
|
||||
part.type !== 'text' && part.type !== 'reasoning'
|
||||
);
|
||||
finalContent = [...before, ...parsedEditableContent, ...after];
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue