mirror of
https://github.com/razzant/ouroboros.git
synced 2026-10-19 02:02:00 +00:00
Lane C convergence fixes: system-proxy detection mirrors httpx, old-httpx/SDK TypeError fallbacks, notifier stops batch on first transient failure, poller size-gate extraction, honest bounded-git lock/auto-push contracts, hermetic and wiring tests
This commit is contained in:
Anton Razzhigaev
parent
9e00972230
commit
474c3f9c52
9 changed files with 482 additions and 47 deletions
|
|
@ -1509,10 +1509,13 @@ is detected by kernel probes instead of hanging until the read timeout: on
|
|||
Linux/macOS the probe timing is tuned to detect within minutes, other
|
||||
platforms get `SO_KEEPALIVE` with OS-default timing. The cached-client
|
||||
transports also carry SDK-equivalent pool limits (an explicit transport
|
||||
ignores Client-level limits). When HTTP(S)_PROXY/ALL_PROXY env proxies are
|
||||
configured, the cached and web-search clients skip the explicit transport
|
||||
(httpx env-proxy mounts require it absent) — disclosed residual: proxy-routed
|
||||
installs run without keepalive tuning. Further residuals without these socket
|
||||
ignores Client-level limits). When any proxy httpx would honor is configured
|
||||
(HTTP(S)_PROXY/ALL_PROXY env vars, macOS SystemConfiguration, the Windows
|
||||
registry — mirrored via `urllib.request.getproxies()`), the cached and
|
||||
web-search clients skip the explicit transport (httpx env-proxy mounts
|
||||
require it absent) — disclosed residual: proxy-routed installs run without
|
||||
keepalive tuning, as do httpx builds predating the transport
|
||||
`socket_options` parameter (< 0.25). Further residuals without these socket
|
||||
options: the native Anthropic `requests` session and the anthropic web-search
|
||||
client, the GigaChat library client, and the short-lived `llm_probe`
|
||||
ephemeral probe clients.
|
||||
|
|
|
|||
|
|
@ -35,9 +35,17 @@ def remote_httpx_transport(
|
|||
}
|
||||
if limits is not None:
|
||||
kwargs["limits"] = limits
|
||||
if async_client:
|
||||
return httpx.AsyncHTTPTransport(**kwargs)
|
||||
return httpx.HTTPTransport(**kwargs)
|
||||
transport_cls = httpx.AsyncHTTPTransport if async_client else httpx.HTTPTransport
|
||||
try:
|
||||
return transport_cls(**kwargs)
|
||||
except TypeError:
|
||||
# httpx < 0.25 has no ``socket_options`` parameter; the dependency pin
|
||||
# is a bare ``httpx``, so such a venv is legal. Rebuild without the
|
||||
# keepalive tuning (silently absent) rather than killing every remote
|
||||
# call at client construction — same guard shape as the openai
|
||||
# Default-client getattr fallback below.
|
||||
kwargs.pop("socket_options", None)
|
||||
return transport_cls(**kwargs)
|
||||
|
||||
|
||||
def _sdk_pool_limits():
|
||||
|
|
@ -55,17 +63,21 @@ def _sdk_pool_limits():
|
|||
|
||||
|
||||
def env_proxies_configured() -> bool:
|
||||
"""True when HTTP(S)_PROXY/ALL_PROXY-style env proxies are configured.
|
||||
"""True when any proxy httpx would honor is configured.
|
||||
|
||||
httpx honors env proxies only when no explicit transport is passed
|
||||
(``allow_env_proxies = trust_env and transport is None``), so attaching
|
||||
the keepalive transport on a proxy-routed install would silently break
|
||||
proxy routing. ``no_proxy`` alone does not count.
|
||||
Mirrors httpx (``get_environment_proxies`` builds its mounts from
|
||||
``urllib.request.getproxies()``): that includes not only the
|
||||
HTTP(S)_PROXY/ALL_PROXY env vars but also macOS SystemConfiguration and
|
||||
the Windows registry, so a system-proxy install (no env vars; a typical
|
||||
GUI-launched macOS app) is detected too. httpx honors those proxies only
|
||||
when no explicit transport is passed, so attaching the keepalive
|
||||
transport there would silently break the install's only working egress.
|
||||
A lone ``no``/NO_PROXY entry does not count.
|
||||
"""
|
||||
import urllib.request
|
||||
|
||||
proxies = urllib.request.getproxies_environment()
|
||||
return any(scheme != "no" for scheme in proxies)
|
||||
proxies = urllib.request.getproxies()
|
||||
return any(scheme in ("http", "https", "all") for scheme in proxies)
|
||||
|
||||
|
||||
def keepalive_http_client(async_client: bool = False):
|
||||
|
|
@ -89,13 +101,22 @@ def keepalive_http_client(async_client: bool = False):
|
|||
# default construction (no keepalive tuning) rather than failing the
|
||||
# LLM client construction over a tuning concern.
|
||||
return None
|
||||
return cls(
|
||||
transport=remote_httpx_transport(async_client, limits=_sdk_pool_limits())
|
||||
)
|
||||
try:
|
||||
return cls(
|
||||
transport=remote_httpx_transport(async_client, limits=_sdk_pool_limits())
|
||||
)
|
||||
except TypeError:
|
||||
# A future SDK generation whose Default client rejects the transport
|
||||
# object falls back to SDK default construction the same way.
|
||||
return None
|
||||
|
||||
|
||||
def make_no_proxy_client(target: Dict[str, Any], timeout: Any) -> Tuple[Any, Any]:
|
||||
"""Per-call OpenAI client fully isolated from proxy/SSL environment."""
|
||||
"""Per-call OpenAI client fully isolated from proxy/SSL environment.
|
||||
|
||||
``timeout`` is the caller's explicit per-call bound (an ``httpx.Timeout``
|
||||
or a float); no default is applied here.
|
||||
"""
|
||||
import httpx
|
||||
from openai import OpenAI
|
||||
|
||||
|
|
@ -116,7 +137,10 @@ def make_no_proxy_client(target: Dict[str, Any], timeout: Any) -> Tuple[Any, Any
|
|||
|
||||
|
||||
def make_no_proxy_async_client(target: Dict[str, Any], timeout: Any) -> Tuple[Any, Any]:
|
||||
"""Async variant of :func:`make_no_proxy_client`."""
|
||||
"""Async variant of :func:`make_no_proxy_client`.
|
||||
|
||||
``timeout`` is the caller's explicit per-call bound, as above.
|
||||
"""
|
||||
import httpx
|
||||
from openai import AsyncOpenAI
|
||||
|
||||
|
|
|
|||
|
|
@ -167,8 +167,13 @@ async def _check_tasks_notify(
|
|||
msg = (f"{icon} Задача {tid[:8]} готова{tail}" if lang == "ru" else f"{icon} Task {tid[:8]} done{tail}")
|
||||
send_outcome, exc = await _push_notification(api, chat_id, msg)
|
||||
if send_outcome == "transient":
|
||||
# Stop the batch on the first transient failure: every further
|
||||
# send would burn its full transport timeout against the same dead
|
||||
# network (a long backlog could stall one cycle for minutes before
|
||||
# the backoff even starts). State is already preserved — the
|
||||
# unsent summaries stay unseen and retry next cycle.
|
||||
transient = exc
|
||||
continue
|
||||
break
|
||||
# "sent" or permanent "skipped": either way this notification is done.
|
||||
delivered = delivered or send_outcome == "sent"
|
||||
seen.append(tid)
|
||||
|
|
@ -196,11 +201,15 @@ def _make_notifier(api):
|
|||
lang = str(settings.get("TELEGRAM_LANGUAGE") or "en").strip().lower()
|
||||
state = _load_notif_state(api)
|
||||
transient, delivered = await _check_budget_notify(api, settings, chat_id, state, lang)
|
||||
tasks_transient, tasks_delivered = await _check_tasks_notify(
|
||||
api, settings, chat_id, state, lang,
|
||||
)
|
||||
transient = tasks_transient or transient
|
||||
delivered = delivered or tasks_delivered
|
||||
if transient is None:
|
||||
# A transient budget-send failure means the transport is
|
||||
# down right now: skip the tasks batch entirely instead of
|
||||
# burning more transport timeouts, and let the backoff
|
||||
# below pace the retry.
|
||||
transient, tasks_delivered = await _check_tasks_notify(
|
||||
api, settings, chat_id, state, lang,
|
||||
)
|
||||
delivered = delivered or tasks_delivered
|
||||
_save_notif_state(api, state)
|
||||
if transient is not None:
|
||||
# Transition logging: one warning entering degraded, one info
|
||||
|
|
|
|||
|
|
@ -609,6 +609,19 @@ async def _start_poller(api):
|
|||
raise
|
||||
|
||||
|
||||
def _fail_permanent_poll_rejection(api, exc, revalidating: bool) -> None:
|
||||
"""Persist bridge status + log for a permanent polling rejection."""
|
||||
if revalidating:
|
||||
# A permanent getMe rejection during in-loop revalidation
|
||||
# is the startup token-validation failure, merely deferred
|
||||
# by a dead network at startup — label it the same way.
|
||||
_save_bridge_status(api, "error", "telegram_startup_failed")
|
||||
api.log("error", f"Telegram token validation failed: {exc}")
|
||||
else:
|
||||
_save_bridge_status(api, "error", "telegram_rejected")
|
||||
api.log("error", "Telegram polling was permanently rejected.")
|
||||
|
||||
|
||||
async def _poller(api) -> None:
|
||||
client, offset, pinned_chat, max_updates, command_mode, lang, validated = await _start_poller(api)
|
||||
|
||||
|
|
@ -888,15 +901,7 @@ async def _poller(api) -> None:
|
|||
await asyncio.sleep(0.1)
|
||||
except (TelegramRequestRejected, TelegramTransportError) as exc:
|
||||
if isinstance(exc, TelegramRequestRejected) and not exc.transient:
|
||||
if revalidating:
|
||||
# A permanent getMe rejection during in-loop revalidation
|
||||
# is the startup token-validation failure, merely deferred
|
||||
# by a dead network at startup — label it the same way.
|
||||
_save_bridge_status(api, "error", "telegram_startup_failed")
|
||||
api.log("error", f"Telegram token validation failed: {exc}")
|
||||
else:
|
||||
_save_bridge_status(api, "error", "telegram_rejected")
|
||||
api.log("error", "Telegram polling was permanently rejected.")
|
||||
_fail_permanent_poll_rejection(api, exc, revalidating)
|
||||
raise
|
||||
# Transition logging only: one line entering degraded, one on
|
||||
# recovery; consecutive transient failures stay quiet while the
|
||||
|
|
|
|||
|
|
@ -35,12 +35,17 @@ def _target():
|
|||
|
||||
|
||||
def _clear_proxy_env(monkeypatch):
|
||||
"""Keep transport-construction tests hermetic against ambient env proxies."""
|
||||
"""Keep transport-construction tests hermetic against ambient proxies.
|
||||
|
||||
Clears the env vars AND stubs ``urllib.request.getproxies`` (the source
|
||||
the detection mirrors) so system proxies on a macOS/Windows runner cannot
|
||||
flip these tests either."""
|
||||
for name in (
|
||||
"HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY",
|
||||
"http_proxy", "https_proxy", "all_proxy",
|
||||
):
|
||||
monkeypatch.delenv(name, raising=False)
|
||||
monkeypatch.setattr("urllib.request.getproxies", lambda: {})
|
||||
|
||||
|
||||
def test_keepalive_socket_options_enable_keepalive_everywhere():
|
||||
|
|
@ -279,8 +284,9 @@ def test_cached_clients_keep_env_proxy_mounts_on_proxied_installs(monkeypatch):
|
|||
"""With HTTP(S)_PROXY configured the cached clients are built WITHOUT an
|
||||
explicit transport: httpx enables env-proxy mounts only when no transport
|
||||
is passed, so the keepalive transport would break proxy routing."""
|
||||
_clear_proxy_env(monkeypatch)
|
||||
monkeypatch.setenv("HTTPS_PROXY", "http://127.0.0.1:3128")
|
||||
monkeypatch.setattr(
|
||||
"urllib.request.getproxies", lambda: {"https": "http://127.0.0.1:3128"}
|
||||
)
|
||||
sync_captured = {}
|
||||
async_captured = {}
|
||||
|
||||
|
|
@ -315,13 +321,102 @@ def test_cached_clients_keep_env_proxy_mounts_on_proxied_installs(monkeypatch):
|
|||
|
||||
|
||||
def test_env_proxies_detection_ignores_no_proxy_alone(monkeypatch):
|
||||
_clear_proxy_env(monkeypatch)
|
||||
monkeypatch.setenv("no_proxy", "localhost")
|
||||
monkeypatch.setattr("urllib.request.getproxies", lambda: {"no": "localhost"})
|
||||
assert net_transport.env_proxies_configured() is False
|
||||
monkeypatch.setenv("HTTPS_PROXY", "http://127.0.0.1:3128")
|
||||
monkeypatch.setattr(
|
||||
"urllib.request.getproxies",
|
||||
lambda: {"no": "localhost", "https": "http://127.0.0.1:3128"},
|
||||
)
|
||||
assert net_transport.env_proxies_configured() is True
|
||||
|
||||
|
||||
def test_system_only_proxy_with_empty_env_disables_keepalive_client(monkeypatch):
|
||||
"""A system-proxy install (macOS SystemConfiguration / Windows registry —
|
||||
surfaced by urllib.request.getproxies() with NO proxy env vars) must keep
|
||||
SDK default construction: the detection mirrors httpx, so the explicit
|
||||
transport is skipped and the install keeps its only working egress."""
|
||||
for name in (
|
||||
"HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY",
|
||||
"http_proxy", "https_proxy", "all_proxy",
|
||||
):
|
||||
monkeypatch.delenv(name, raising=False)
|
||||
monkeypatch.setattr(
|
||||
"urllib.request.getproxies", lambda: {"http": "http://sysproxy.local:8080"}
|
||||
)
|
||||
assert net_transport.env_proxies_configured() is True
|
||||
assert net_transport.keepalive_http_client() is None
|
||||
assert net_transport.keepalive_http_client(async_client=True) is None
|
||||
|
||||
|
||||
def test_transport_factory_survives_httpx_without_socket_options(monkeypatch):
|
||||
"""httpx < 0.25 has no socket_options parameter (the dependency pin is a
|
||||
bare httpx): the factory retries without it — keepalive tuning silently
|
||||
absent — instead of raising TypeError at every remote client build."""
|
||||
built = []
|
||||
|
||||
class OldTransport:
|
||||
def __init__(self, **kwargs):
|
||||
if "socket_options" in kwargs:
|
||||
raise TypeError(
|
||||
"__init__() got an unexpected keyword argument 'socket_options'"
|
||||
)
|
||||
built.append(kwargs)
|
||||
|
||||
monkeypatch.setattr(httpx, "HTTPTransport", OldTransport)
|
||||
monkeypatch.setattr(httpx, "AsyncHTTPTransport", OldTransport)
|
||||
|
||||
transport = net_transport.remote_httpx_transport(trust_env=False)
|
||||
async_transport = net_transport.remote_httpx_transport(True)
|
||||
|
||||
assert isinstance(transport, OldTransport)
|
||||
assert isinstance(async_transport, OldTransport)
|
||||
assert len(built) == 2
|
||||
assert all("socket_options" not in kwargs for kwargs in built)
|
||||
assert built[0]["trust_env"] is False
|
||||
|
||||
|
||||
def test_keepalive_client_falls_back_to_none_when_default_client_rejects_transport(monkeypatch):
|
||||
"""A Default client class that rejects the transport kwarg (a future SDK
|
||||
generation) degrades to SDK default construction (None) instead of
|
||||
failing the LLM client build."""
|
||||
_clear_proxy_env(monkeypatch)
|
||||
|
||||
def rejecting_default_client(**_kwargs):
|
||||
raise TypeError("unexpected keyword argument 'transport'")
|
||||
|
||||
monkeypatch.setitem(
|
||||
sys.modules,
|
||||
"openai",
|
||||
types.SimpleNamespace(
|
||||
DefaultHttpxClient=rejecting_default_client,
|
||||
DefaultAsyncHttpxClient=rejecting_default_client,
|
||||
),
|
||||
)
|
||||
assert net_transport.keepalive_http_client() is None
|
||||
assert net_transport.keepalive_http_client(async_client=True) is None
|
||||
|
||||
|
||||
def test_cached_client_without_default_classes_still_builds(monkeypatch):
|
||||
"""An openai SDK build without the Default client classes: the cached
|
||||
client is still constructed (max_retries=0), just without an explicit
|
||||
http_client — keepalive tuning absent, construction never fails."""
|
||||
_clear_proxy_env(monkeypatch)
|
||||
captured = {}
|
||||
|
||||
class FakeOpenAI:
|
||||
def __init__(self, **kwargs):
|
||||
captured.update(kwargs)
|
||||
|
||||
monkeypatch.setitem(
|
||||
sys.modules, "openai", types.SimpleNamespace(OpenAI=FakeOpenAI),
|
||||
)
|
||||
client = LLMClient._new_remote_client(_target())
|
||||
|
||||
assert isinstance(client, FakeOpenAI)
|
||||
assert captured.get("max_retries") == 0
|
||||
assert "http_client" not in captured
|
||||
|
||||
|
||||
def test_web_search_openai_client_carries_keepalive_transport(monkeypatch):
|
||||
"""Q16 coverage: the web-search OpenAI client construction rides the
|
||||
shared keepalive transport (the anthropic web-search client is a
|
||||
|
|
@ -353,3 +448,65 @@ def test_web_search_openai_client_carries_keepalive_transport(monkeypatch):
|
|||
assert isinstance(captured["http_client"], FakeDefaultHttpxClient)
|
||||
assert len(built_clients) == 1
|
||||
assert isinstance(built_clients[0].get("transport"), httpx.HTTPTransport)
|
||||
|
||||
|
||||
def test_llm_openrouter_web_search_routes_through_net_transport_seam(monkeypatch):
|
||||
"""Call-site wiring: llm.py's openrouter web-search server tool builds its
|
||||
client through net_transport.web_search_openai_client (seam assertion —
|
||||
no private httpx internals)."""
|
||||
from ouroboros import llm as llm_mod
|
||||
from ouroboros import net_transport as nt_mod
|
||||
|
||||
recorded = {}
|
||||
|
||||
def seam(**kwargs):
|
||||
recorded.update(kwargs)
|
||||
raise RuntimeError("seam sentinel")
|
||||
|
||||
monkeypatch.setattr(nt_mod, "web_search_openai_client", seam)
|
||||
with patch.object(llm_mod, "_physical_candidate", side_effect=AssertionError):
|
||||
try:
|
||||
llm_mod.openrouter_web_search_server_tool(
|
||||
api_key="k", model="openai/gpt-5.5", query="q",
|
||||
search_context_size="low", timeout=12.0,
|
||||
)
|
||||
except RuntimeError as exc:
|
||||
assert "seam sentinel" in str(exc)
|
||||
else:
|
||||
raise AssertionError("client construction seam was not reached")
|
||||
|
||||
assert recorded["api_key"] == "k"
|
||||
assert recorded["base_url"] == "https://openrouter.ai/api/v1"
|
||||
assert recorded["timeout"] == 12.0
|
||||
|
||||
|
||||
def test_tools_web_search_routes_through_net_transport_seam(monkeypatch):
|
||||
"""Call-site wiring: tools/search.py's OpenAI web-search leg builds its
|
||||
client through net_transport.web_search_openai_client. The backend pin
|
||||
'openai' keeps the failure path hermetic (no cascade to ddgs)."""
|
||||
from types import SimpleNamespace
|
||||
|
||||
from ouroboros import net_transport as nt_mod
|
||||
from ouroboros.tools import search as search_mod
|
||||
|
||||
recorded = {}
|
||||
|
||||
def seam(**kwargs):
|
||||
recorded.update(kwargs)
|
||||
raise RuntimeError("seam sentinel")
|
||||
|
||||
monkeypatch.setenv("OUROBOROS_WEBSEARCH_BACKEND", "openai")
|
||||
monkeypatch.setattr(nt_mod, "web_search_openai_client", seam)
|
||||
monkeypatch.setattr(
|
||||
search_mod,
|
||||
"_resolve_openai_client_settings",
|
||||
lambda: ("key", "https://api.test/v1", "openai", "official"),
|
||||
)
|
||||
ctx = SimpleNamespace(task_metadata={}, pending_events=[], task_id="")
|
||||
|
||||
result = search_mod._web_search(ctx, "seam query")
|
||||
|
||||
assert recorded["api_key"] == "key"
|
||||
assert recorded["base_url"] == "https://api.test/v1"
|
||||
assert isinstance(recorded["timeout"], float)
|
||||
assert "seam sentinel" in result
|
||||
|
|
|
|||
|
|
@ -8,9 +8,20 @@ import pathlib
|
|||
import stat
|
||||
import subprocess
|
||||
import time
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
from supervisor import git_ops, update_source
|
||||
|
||||
# The hung-git tests fake `git` with a #!/bin/sh PATH shim; that shim is not
|
||||
# executable on Windows (the real git would run and the assertions would
|
||||
# lie), so they are POSIX-only. The pure-monkeypatch shape tests below stay
|
||||
# cross-platform.
|
||||
_posix_shim = pytest.mark.skipif(
|
||||
os.name == "nt", reason="uses a #!/bin/sh PATH shim; POSIX-only"
|
||||
)
|
||||
|
||||
|
||||
def _git(repo: pathlib.Path, *args: str) -> str:
|
||||
res = subprocess.run(
|
||||
|
|
@ -80,11 +91,15 @@ def test_git_network_bounded_rejects_missing_cwd(tmp_path):
|
|||
assert "cwd" in err
|
||||
|
||||
|
||||
def test_git_network_bounded_timeout_kills_process_tree(tmp_path, monkeypatch):
|
||||
"""A hung network git is killed together with its children and returns the
|
||||
typed timeout shape; nothing keeps holding the repository afterwards — a
|
||||
real follow-up network git command in the same clone must succeed (this
|
||||
passing follow-up IS the lock-release contract; no extra cleanup exists)."""
|
||||
@_posix_shim
|
||||
def test_git_network_bounded_timeout_kills_tree_and_repo_stays_operable(tmp_path, monkeypatch):
|
||||
"""A hung network git is killed together with its children (kill + reap)
|
||||
and returns the typed timeout shape. The shim drops a lockfile under the
|
||||
clone's ``.git`` before sleeping: SIGKILL leaves such stale lockfiles
|
||||
behind — the runner does no cleanup, and clearing them is left to git's
|
||||
own per-file tolerance. The contract asserted here is that the repository
|
||||
stays operable: a real follow-up ``git fetch origin`` in the same clone
|
||||
succeeds."""
|
||||
upstream = _seed_repo(tmp_path / "upstream")
|
||||
clone = tmp_path / "clone"
|
||||
subprocess.run(
|
||||
|
|
@ -100,10 +115,12 @@ def test_git_network_bounded_timeout_kills_process_tree(tmp_path, monkeypatch):
|
|||
shim_dir.mkdir()
|
||||
pid_dir = tmp_path / "pids"
|
||||
pid_dir.mkdir()
|
||||
stale_lock = clone / ".git" / "config.lock"
|
||||
fake_git = shim_dir / "git"
|
||||
fake_git.write_text(
|
||||
"#!/bin/sh\n"
|
||||
'echo $$ > "$NETRES_PID_DIR/parent.pid"\n'
|
||||
'touch "$NETRES_STALE_LOCK"\n'
|
||||
"sleep 300 &\n"
|
||||
'echo $! > "$NETRES_PID_DIR/child.pid"\n'
|
||||
"wait\n",
|
||||
|
|
@ -112,6 +129,7 @@ def test_git_network_bounded_timeout_kills_process_tree(tmp_path, monkeypatch):
|
|||
fake_git.chmod(fake_git.stat().st_mode | stat.S_IXUSR)
|
||||
monkeypatch.setenv("PATH", f"{shim_dir}{os.pathsep}{original_path}")
|
||||
monkeypatch.setenv("NETRES_PID_DIR", str(pid_dir))
|
||||
monkeypatch.setenv("NETRES_STALE_LOCK", str(stale_lock))
|
||||
|
||||
rc, out, err = update_source._git_network_bounded(
|
||||
["fetch", "origin"], cwd=clone, timeout=1.0,
|
||||
|
|
@ -137,8 +155,12 @@ def test_git_network_bounded_timeout_kills_process_tree(tmp_path, monkeypatch):
|
|||
else:
|
||||
raise AssertionError(f"process {pid} survived the bounded timeout kill")
|
||||
|
||||
# Contract: the killed fetch left nothing behind that breaks a real
|
||||
# follow-up network git command in the same clone.
|
||||
# SIGKILL leaves the stale lockfile behind — honesty pin: the runner does
|
||||
# NOT clean it up.
|
||||
assert stale_lock.exists()
|
||||
|
||||
# Contract: the repository stays operable — a real follow-up network git
|
||||
# command in the same clone succeeds despite the stale lockfile.
|
||||
monkeypatch.setenv("PATH", original_path)
|
||||
rc2, _out2, err2 = update_source._git_network_bounded(
|
||||
["fetch", "origin"], cwd=clone, timeout=60,
|
||||
|
|
@ -232,3 +254,88 @@ def test_ci_push_branch_is_bounded_with_repo_cwd_and_keeps_shape(tmp_path, monke
|
|||
ok, message = ci._push_branch(str(repo), "feature")
|
||||
assert ok is False
|
||||
assert "exceeded" in message
|
||||
|
||||
|
||||
def test_run_git_network_cmd_failure_reports_stdout_when_stderr_is_empty(tmp_path, monkeypatch):
|
||||
"""Some git failures report only on stdout; the run_cmd-shaped error must
|
||||
carry that text instead of an empty STDERR-only message."""
|
||||
from ouroboros.tools import git as git_tools
|
||||
|
||||
monkeypatch.setattr(
|
||||
update_source,
|
||||
"_git_network_bounded",
|
||||
lambda args, **_kw: (1, "remote: rejected by hook", ""),
|
||||
)
|
||||
with pytest.raises(RuntimeError) as excinfo:
|
||||
git_tools._run_git_network_cmd(["git", "fetch", "origin"], cwd=tmp_path)
|
||||
message = str(excinfo.value)
|
||||
assert "Command failed: git fetch origin" in message
|
||||
assert "remote: rejected by hook" in message
|
||||
|
||||
|
||||
def test_commit_path_auto_push_timeout_is_best_effort_warning(tmp_path, monkeypatch):
|
||||
"""Through the commit path: a push timeout in the real ``_auto_push`` →
|
||||
real ``push_to_remote`` → bounded-runner chain leaves the commit itself
|
||||
successful, with the push-failed warning carried in the result."""
|
||||
from ouroboros.tools import git as git_module
|
||||
|
||||
ctx = SimpleNamespace(
|
||||
repo_dir=tmp_path,
|
||||
branch_dev="ouroboros",
|
||||
current_task_type="task",
|
||||
task_id="t1",
|
||||
task_metadata={},
|
||||
last_push_succeeded=True,
|
||||
pending_events=[],
|
||||
)
|
||||
|
||||
def fake_run(cmd, cwd=None, **_kw):
|
||||
if cmd[:2] == ["git", "commit"]:
|
||||
return ""
|
||||
if cmd[:2] == ["git", "rev-parse"]:
|
||||
return "a" * 40 + "\n"
|
||||
return ""
|
||||
|
||||
def fake_stage_cycle(_ctx, _msg, _start, **_kw):
|
||||
return {
|
||||
"status": "passed", "message": "",
|
||||
"pre_fingerprint": {"fingerprint": "x"},
|
||||
"post_fingerprint": {"fingerprint": "x", "binding": {}},
|
||||
}
|
||||
|
||||
monkeypatch.setattr(git_module, "run_cmd", fake_run)
|
||||
monkeypatch.setattr(git_module, "_run_reviewed_stage_cycle", fake_stage_cycle)
|
||||
monkeypatch.setattr(git_module, "_task_attributed_commit_paths",
|
||||
lambda _ctx, paths: (paths, None, "", None))
|
||||
monkeypatch.setattr(git_module, "_check_overlapping_review_attempt", lambda _ctx: "")
|
||||
monkeypatch.setattr(git_module, "_prepare_review_commit_worktree",
|
||||
lambda _ctx, _tx: (False, ""))
|
||||
monkeypatch.setattr(git_module, "_verify_reviewed_commit_binding",
|
||||
lambda *_a, **_kw: (True, ""))
|
||||
monkeypatch.setattr(git_module, "_managed_post_commit_tests_gate",
|
||||
lambda *_a, **_kw: "")
|
||||
monkeypatch.setattr(git_module, "_auto_tag_on_version_bump", lambda *_a, **_kw: "")
|
||||
monkeypatch.setattr(git_module, "_post_commit_result", lambda *_a, **_kw: None)
|
||||
monkeypatch.setattr(git_module, "_record_commit_attempt", lambda *_a, **_kw: None)
|
||||
monkeypatch.setattr(git_module, "_acquire_git_lock", lambda _ctx: tmp_path / "git.lock")
|
||||
monkeypatch.setattr(git_module, "_release_git_lock", lambda _path: None)
|
||||
|
||||
from supervisor import update_merge
|
||||
|
||||
monkeypatch.setattr(update_merge, "managed_assisted_tx_for",
|
||||
lambda _task_id, _meta: (None, ""))
|
||||
|
||||
# The push leg stays REAL: _auto_push → push_to_remote → bounded runner.
|
||||
monkeypatch.setattr(git_ops, "_has_remote", lambda _name: True)
|
||||
monkeypatch.setattr(
|
||||
git_ops,
|
||||
"_git_network_bounded",
|
||||
lambda _cmd, **_kw: (git_ops.FETCH_TIMEOUT_RC, "", "git push exceeded 300s and was terminated"),
|
||||
)
|
||||
|
||||
result = git_module._repo_commit_push(ctx, "netres: best-effort push test")
|
||||
|
||||
assert result.startswith("OK: committed to ouroboros:")
|
||||
assert "[push skipped: git push failed:" in result
|
||||
assert "exceeded" in result
|
||||
assert ctx.last_push_succeeded is False
|
||||
|
|
|
|||
|
|
@ -572,6 +572,15 @@ def test_typed_payment_required_maps_to_no_credits():
|
|||
|
||||
|
||||
def test_ephemeral_openai_client_disables_sdk_retries(monkeypatch):
|
||||
# Hermetic against proxied runners: with any proxy httpx would honor the
|
||||
# keepalive helper deliberately returns None (SDK default construction),
|
||||
# which would drop the http_client kwarg this test asserts on.
|
||||
for name in (
|
||||
"HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY",
|
||||
"http_proxy", "https_proxy", "all_proxy",
|
||||
):
|
||||
monkeypatch.delenv(name, raising=False)
|
||||
monkeypatch.setattr("urllib.request.getproxies", lambda: {})
|
||||
captured = {}
|
||||
http_clients = []
|
||||
|
||||
|
|
|
|||
|
|
@ -430,6 +430,44 @@ def test_startup_transient_getme_does_not_kill_poller(tmp_path, monkeypatch):
|
|||
assert any("Telegram poller started" in m for _lvl, m, _f in api.logs)
|
||||
|
||||
|
||||
def test_startup_deferred_validation_with_network_still_down_paces_with_backoff(tmp_path, monkeypatch):
|
||||
"""After a deferred (transient) startup getMe, a network that STAYS down
|
||||
keeps the poller alive: every in-loop revalidation failure paces with the
|
||||
monotone 5→60 backoff instead of hot-looping or raising."""
|
||||
plugin = _load_plugin(tmp_path)
|
||||
failures = {"count": 0}
|
||||
|
||||
class StillDownClient(FakeTelegramClient):
|
||||
async def call(self, method, **kwargs):
|
||||
if method == "getMe":
|
||||
failures["count"] += 1
|
||||
raise plugin.TelegramTransportError("Telegram API timed out during getMe.")
|
||||
return {"ok": True, "result": {}}
|
||||
|
||||
async def get_updates(self, _offset):
|
||||
raise AssertionError("polling must not start before validation succeeds")
|
||||
|
||||
monkeypatch.setattr(plugin, "TelegramClient", StillDownClient)
|
||||
sleeps = []
|
||||
|
||||
async def record_sleep(delay):
|
||||
sleeps.append(delay)
|
||||
if len(sleeps) >= 8:
|
||||
raise asyncio.CancelledError
|
||||
|
||||
monkeypatch.setattr(plugin.asyncio, "sleep", record_sleep)
|
||||
try:
|
||||
asyncio.run(plugin._make_poller(FakeApi(tmp_path))())
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
|
||||
assert failures["count"] >= 8 # startup + one revalidation per backoff sleep
|
||||
assert sleeps[0] == 5
|
||||
assert all(later >= earlier for earlier, later in zip(sleeps, sleeps[1:]))
|
||||
assert max(sleeps) == 60
|
||||
assert sleeps[-2:] == [60, 60]
|
||||
|
||||
|
||||
def test_startup_permanent_rejection_keeps_raise_semantics(tmp_path, monkeypatch):
|
||||
plugin = _load_plugin(tmp_path)
|
||||
|
||||
|
|
|
|||
|
|
@ -184,6 +184,89 @@ def test_task_notification_retries_before_advancing_ledger(tmp_path, monkeypatch
|
|||
assert Flaky.attempts == 2 and state["notified_task_ids"] == ["retry1"]
|
||||
|
||||
|
||||
def test_tasks_notify_stops_batch_on_first_transient_failure(tmp_path, monkeypatch):
|
||||
"""A transient send failure ends the batch: with three pending summaries
|
||||
exactly ONE send is attempted in the cycle (each further send would burn
|
||||
its full transport timeout against the same dead network) and no summary
|
||||
is marked seen, so all three retry next cycle."""
|
||||
nt = _load(); api, data = _api(tmp_path)
|
||||
|
||||
class AlwaysOffline:
|
||||
attempts = 0
|
||||
def __init__(self, _token): pass
|
||||
async def send_message(self, *_args, **_kwargs):
|
||||
AlwaysOffline.attempts += 1
|
||||
raise nt.TelegramTransportError("offline")
|
||||
|
||||
monkeypatch.setattr(nt, "TelegramClient", AlwaysOffline)
|
||||
with open(data / "logs" / "chat.jsonl", "w", encoding="utf-8") as f:
|
||||
for tid in ("t1", "t2", "t3"):
|
||||
f.write(json.dumps({"type": "task_summary", "task_id": tid,
|
||||
"outcome_axes": {"lifecycle": "completed"}}) + "\n")
|
||||
state = {"notified_task_ids": []}
|
||||
transient, delivered = asyncio.run(
|
||||
nt._check_tasks_notify(api, {"TELEGRAM_NOTIFY_TASKS": "on"}, 42, state, "en")
|
||||
)
|
||||
|
||||
assert AlwaysOffline.attempts == 1
|
||||
assert transient is not None and delivered is False
|
||||
assert state["notified_task_ids"] == []
|
||||
|
||||
|
||||
def test_notifier_cycle_makes_one_send_attempt_when_transport_is_down(tmp_path, monkeypatch):
|
||||
"""One dead-network cycle = exactly one send attempt, then backoff: a
|
||||
transient budget-send failure skips the tasks batch entirely."""
|
||||
nt = _load(); api, data = _api(tmp_path)
|
||||
|
||||
class AlwaysOffline:
|
||||
attempts = 0
|
||||
def __init__(self, _token): pass
|
||||
async def send_message(self, *_args, **_kwargs):
|
||||
AlwaysOffline.attempts += 1
|
||||
raise nt.TelegramTransportError("offline")
|
||||
|
||||
monkeypatch.setattr(nt, "TelegramClient", AlwaysOffline)
|
||||
|
||||
async def runtime_state(_api):
|
||||
return {"spent_usd": 850, "budget_limit": 1000, "budget_pct": 85}
|
||||
|
||||
monkeypatch.setattr(nt, "_load_runtime_state", runtime_state)
|
||||
with open(data / "logs" / "chat.jsonl", "w", encoding="utf-8") as f:
|
||||
for tid in ("t1", "t2", "t3"):
|
||||
f.write(json.dumps({"type": "task_summary", "task_id": tid,
|
||||
"outcome_axes": {"lifecycle": "completed"}}) + "\n")
|
||||
tasks_checks = []
|
||||
real_tasks_check = nt._check_tasks_notify
|
||||
|
||||
async def counting_tasks_check(*args, **kwargs):
|
||||
tasks_checks.append(True)
|
||||
return await real_tasks_check(*args, **kwargs)
|
||||
|
||||
monkeypatch.setattr(nt, "_check_tasks_notify", counting_tasks_check)
|
||||
monkeypatch.setattr(
|
||||
nt,
|
||||
"_load_settings",
|
||||
lambda _api: {
|
||||
"TELEGRAM_NOTIFY_BUDGET": "on",
|
||||
"TELEGRAM_NOTIFY_TASKS": "on",
|
||||
"TELEGRAM_CHAT_ID": "42",
|
||||
},
|
||||
)
|
||||
sleeps = []
|
||||
|
||||
async def record_sleep(delay):
|
||||
sleeps.append(delay)
|
||||
raise asyncio.CancelledError
|
||||
|
||||
monkeypatch.setattr(nt.asyncio, "sleep", record_sleep)
|
||||
with pytest.raises(asyncio.CancelledError):
|
||||
asyncio.run(nt._make_notifier(api)())
|
||||
|
||||
assert AlwaysOffline.attempts == 1 # budget send only; tasks batch skipped
|
||||
assert tasks_checks == []
|
||||
assert sleeps == [5] # backoff, not the healthy 30s pacing sleep
|
||||
|
||||
|
||||
def test_notifier_ignores_ambient_chat_id(monkeypatch):
|
||||
nt = _load()
|
||||
monkeypatch.setenv("TELEGRAM_CHAT_ID", "42")
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue