refactor(craft): consolidate turn/queue/liveness/client timeouts into a registry (#13631)

This commit is contained in:
Wenxi authored and GitHub committed 2026-07-30 23:02:51 +00:00
1 parent 4009db6005
commit 650fd922cf
21 files changed
+333 -105

No files matched your search

@@ -51,9 +51,13 @@ from onyx.db.scheduled_task import (
mark_run_status,
)
from onyx.server.features.build.scheduled_tasks.executor import (
DEFAULT_EXECUTOR_BUDGET_SECONDS,
run_scheduled_task_logic,
)
from onyx.server.features.build.timeouts import (
QUEUE_RESIDENCY_SECONDS,
TURN_BUDGET_SECONDS,
TURN_RECLAIM_SLACK_SECONDS,
)
from onyx.server.features.build.utils import is_craft_enabled_for_user
from onyx.server.settings.store import load_settings
@@ -65,19 +69,15 @@ from onyx.server.settings.store import load_settings
# concurrent ticks just see "no rows" and exit.
DISPATCH_BATCH_SIZE = 50
# How long the executor message can sit on the `scheduled_tasks` queue
# before Celery drops it. After 15 min we'd rather let the next beat tick
# re-dispatch a new run row than execute a stale one.
RUN_EXPIRES_SECONDS = 15 * 60
# Stuck-run sweeper thresholds. The QUEUED threshold needs to comfortably
# exceed the longest plausible queue wait + executor startup. The RUNNING
# threshold should exceed the executor budget by enough margin that a
# well-behaved run that hits its own budget gets there first and marks
# itself FAILED (rather than the sweeper). Budget is 30 min by default;
# 45 min gives 50% slack.
STUCK_QUEUED_OLDER_THAN = timedelta(minutes=15)
STUCK_RUNNING_OLDER_THAN = timedelta(seconds=DEFAULT_EXECUTOR_BUDGET_SECONDS + 15 * 60)
# Celery expiry and the stuck-QUEUED sweep are one policy: a run that sat
# queued past the residency limit is dropped unexecuted and its row reclaimed
# by the same threshold. A RUNNING row is stuck only past budget + slack, so a
# well-behaved run that hits its own budget always marks itself FAILED first.
RUN_EXPIRES_SECONDS = QUEUE_RESIDENCY_SECONDS
STUCK_QUEUED_OLDER_THAN = timedelta(seconds=QUEUE_RESIDENCY_SECONDS)
STUCK_RUNNING_OLDER_THAN = timedelta(
seconds=TURN_BUDGET_SECONDS + TURN_RECLAIM_SLACK_SECONDS
)
# --- Dispatch ----------------------------------------------------------------
+21 -11
View File
@@ -37,6 +37,10 @@ SANDBOX_MAX_CONCURRENT_PER_ORG = int(
SANDBOX_APPROVAL_WAIT_TIMEOUT_SECONDS = int(
os.environ.get("SANDBOX_APPROVAL_WAIT_TIMEOUT_SECONDS", "180")
)
# Margin a client-side wait must add over the approval window so a decision
# landing at the wire never races the client's own timeout. Shared by the
# AGENTS.md guidance and the inactivity-backstop default below.
SANDBOX_APPROVAL_WAIT_MARGIN_SECONDS = 20
SANDBOX_IDLE_CLEANUP_INTERVAL_SECONDS = int(
os.environ.get("SANDBOX_IDLE_CLEANUP_INTERVAL_SECONDS", "60")
)
@@ -168,18 +172,21 @@ SANDBOX_DOCKER_CPU_LIMIT = float(os.environ.get("SANDBOX_DOCKER_CPU_LIMIT", "1.0
SSE_KEEPALIVE_INTERVAL = float(os.environ.get("SSE_KEEPALIVE_INTERVAL", "15.0"))
# Maximum time opencode-serve may go without emitting a turn event. Coarse
# liveness backstop only: it must stay above opencode's own per-tool timeouts
# (bash defaults to 180s here, webfetch 120s) so it never pre-empts a healthy
# long-running tool — it exists to catch stalls opencode does not bound itself
# (LLM-stream hangs, non-bash/MCP tool hangs). The turn budget is the hard ceiling.
# liveness backstop only: it must stay above every in-tool wait so it never
# pre-empts a healthy long-running tool — opencode's bash tool defaults to
# 180s, and a proxy-parked approval inside a tool call holds the stream
# silent for the full approval window, hence the derivation. It exists to
# catch stalls opencode does not bound itself (LLM-stream hangs, non-bash/MCP
# tool hangs). The turn budget is the hard ceiling.
OPENCODE_PROMPT_INACTIVITY_TIMEOUT_SECONDS = float(
os.environ.get("OPENCODE_PROMPT_INACTIVITY_TIMEOUT_SECONDS", "200.0")
os.environ.get(
"OPENCODE_PROMPT_INACTIVITY_TIMEOUT_SECONDS",
str(
SANDBOX_APPROVAL_WAIT_TIMEOUT_SECONDS + SANDBOX_APPROVAL_WAIT_MARGIN_SECONDS
),
)
)
# Hard ceiling for background prompt-slot renewal, so a leaked holder cannot
# retain mutual exclusion indefinitely.
PROMPT_SLOT_KEEP_ALIVE_MAX_SECONDS = 30 * 60.0
# Prompt-slot lock lease; renewed on every sandbox event/keepalive, so a dead
# holder strands the slot for at most this long.
PROMPT_SLOT_LEASE_SECONDS = float(os.environ.get("PROMPT_SLOT_LEASE_SECONDS", "120.0"))
@@ -203,8 +210,11 @@ OPENCODE_SERVE_CONNECT_TIMEOUT = float(
OPENCODE_SERVE_REQUEST_TIMEOUT = float(
os.environ.get("OPENCODE_SERVE_REQUEST_TIMEOUT", "30.0")
)
# Idle timeout for /event SSE. The reader reconnects (with backoff) if the
# stream is silent for this long.
# Idle timeout for the raw /event SSE connection to opencode-serve. The
# reader reconnects (with backoff) if no bytes arrive for this long. Its
# floor is opencode-serve's own emission cadence on /event — NOT our
# downstream UI keepalive, which is synthesized after the bus and never
# reaches this connection.
OPENCODE_SERVE_EVENT_READ_TIMEOUT = float(
os.environ.get("OPENCODE_SERVE_EVENT_READ_TIMEOUT", "60.0")
)
@@ -21,6 +21,7 @@ from onyx.db.enums import Permission
from onyx.db.models import User
from onyx.error_handling.error_codes import OnyxErrorCode
from onyx.error_handling.exceptions import OnyxError
from onyx.server.features.build.configs import SSE_KEEPALIVE_INTERVAL
from onyx.server.features.build.db.build_session import get_build_session
from onyx.server.features.build.interactive_turns.executor import (
start_interactive_turn_runner,
@@ -33,16 +34,18 @@ from onyx.server.features.build.interactive_turns.state import (
)
from onyx.server.features.build.session.manager import SessionManager
from onyx.server.features.build.session.streaming import SSE_KEEPALIVE
from onyx.server.features.build.timeouts import POLL_INTERVAL_SECONDS
from onyx.utils.logger import setup_logger
router = APIRouter()
logger = setup_logger()
LIVE_STREAM_READY_POLL_SECONDS = 0.25
LIVE_STREAM_KEEPALIVE_SECONDS = 15.0
LIVE_STREAM_RUNNER_RETRY_SECONDS = 5.0
TERMINAL_STREAM_EVENT_TYPES = frozenset(("prompt_response", "error"))
# Throttle on runner-claim retries from the attach stream; must be well under
# RUNNER_STALE_AFTER_SECONDS so a stolen turn restarts promptly.
LIVE_STREAM_RUNNER_RETRY_SECONDS = 5.0
def _format_stream_error(detail: str) -> str:
payload = {
@@ -194,7 +197,7 @@ def get_interactive_turn_events(
yield SSE_KEEPALIVE
maybe_start_runner()
time.sleep(LIVE_STREAM_READY_POLL_SECONDS)
time.sleep(POLL_INTERVAL_SECONDS)
try:
with get_session_with_current_tenant() as stream_db_session:
@@ -203,7 +206,7 @@ def get_interactive_turn_events(
for chunk in session_manager.subscribe_to_existing_session_events(
session_id,
user_id,
keepalive_seconds=LIVE_STREAM_KEEPALIVE_SECONDS,
keepalive_seconds=SSE_KEEPALIVE_INTERVAL,
):
yield chunk
chunk_type = _stream_chunk_type(chunk)
@@ -229,8 +232,9 @@ def get_interactive_turn_events(
continue
terminal_error_detail = None
# Not redundant with the attach-time force start: a dead
# runner's turn only becomes reclaimable 90s after its last
# heartbeat; claim no-ops while the turn is healthy.
# runner's turn only becomes reclaimable
# RUNNER_STALE_AFTER_SECONDS after its last heartbeat;
# claim no-ops while the turn is healthy.
maybe_start_runner()
if terminal_error_detail:
yield _format_stream_error(terminal_error_detail)
@@ -31,10 +31,6 @@ from onyx.server.features.build.sandbox.event_schema import (
)
from onyx.server.features.build.sandbox.event_schema import Error as SandboxError
from onyx.server.features.build.sandbox.models import PromptAttachment
from onyx.server.features.build.sandbox.serve_transport import (
PROMPT_SLOT_FAST_FAIL_ACQUIRE_SECONDS,
PROMPT_SLOT_WAIT_OUT_ORPHAN_SECONDS,
)
from onyx.server.features.build.sandbox.sse import SSEKeepalive
from onyx.server.features.build.session.interrupt_signal import (
clear_interrupt,
@@ -42,6 +38,11 @@ from onyx.server.features.build.session.interrupt_signal import (
)
from onyx.server.features.build.session.manager import SessionManager
from onyx.server.features.build.session.streaming import BuildStreamingState
from onyx.server.features.build.timeouts import (
PROMPT_SLOT_FAST_FAIL_ACQUIRE_SECONDS,
PROMPT_SLOT_WAIT_OUT_ORPHAN_SECONDS,
TURN_BUDGET_SECONDS,
)
from onyx.utils.logger import setup_logger
from shared_configs.contextvars import (
CURRENT_TENANT_ID_CONTEXTVAR,
@@ -50,8 +51,6 @@ from shared_configs.contextvars import (
logger = setup_logger()
DEFAULT_INTERACTIVE_TURN_BUDGET_SECONDS = 30 * 60
MAX_TIMEOUT_CONTINUATIONS = 2
_TOOL_TIMEOUT_CONTINUATION_PROMPT = (
"Your last step was cancelled — it exceeded the "
@@ -131,7 +130,7 @@ def start_interactive_turn_runner(turn_id: UUID) -> None:
def run_claimed_interactive_build_turn(
turn: InteractiveTurn,
*,
budget_seconds: int = DEFAULT_INTERACTIVE_TURN_BUDGET_SECONDS,
budget_seconds: int = TURN_BUDGET_SECONDS,
) -> None:
"""Execute a turn that this runner has already claimed in CacheBackend."""
cache = get_cache_backend()
@@ -225,7 +224,7 @@ def _drive_interactive_turn(
return
try:
# Re-fence after the (possibly long) slot wait: a reclaim acquire
# Re-check ownership after the (possibly long) slot wait: a reclaim acquire
# can block past RUNNER_STALE_AFTER_SECONDS, letting another
# runner steal the turn — it must not reach the prompt POST.
if not touch_turn(cache=cache, turn_id=turn_id, runner_id=runner_id):
@@ -10,8 +10,19 @@ from uuid import UUID, uuid4
from onyx.cache.interface import CacheBackend, CacheLock
from onyx.server.features.build.sandbox.models import PromptAttachment
from onyx.server.features.build.timeouts import (
ACTIVE_TURN_TTL_SECONDS,
REQUEST_ID_TTL_SECONDS,
RUNNER_STALE_AFTER_SECONDS,
)
from onyx.utils.datetime import datetime_to_utc
# Turn-state mutex: guards a handful of Redis round-trips per mutation; the
# lease only needs to exceed the longest critical section. Ownership is
# decided by the runner_id compare, never by this lease.
TURN_LOCK_LEASE_SECONDS = 60.0
TURN_LOCK_WAIT_SECONDS = 10.0
class InteractiveTurnStatus(StrEnum):
QUEUED = "QUEUED"
@@ -28,11 +39,6 @@ TURN_STATUS_FAILED = InteractiveTurnStatus.FAILED
TURN_STATUS_CANCELLED = InteractiveTurnStatus.CANCELLED
ACTIVE_TURN_STATUSES = frozenset((TURN_STATUS_QUEUED, TURN_STATUS_RUNNING))
ACTIVE_TURN_TTL_SECONDS = 45 * 60
REQUEST_ID_TTL_SECONDS = 60 * 60
TURN_LOCK_LEASE_SECONDS = 60.0
TURN_LOCK_WAIT_SECONDS = 10.0
RUNNER_STALE_AFTER_SECONDS = 90.0
class InteractiveTurnLockError(Exception):
@@ -25,6 +25,11 @@ from onyx.server.features.build.sandbox.image.sandbox_daemon.contract import (
FilesystemListResponse,
)
from onyx.server.features.build.sandbox.models import FilesystemEntry
from onyx.server.features.build.timeouts import (
BULK_TRANSFER_TIMEOUT_SECONDS,
CONNECT_TIMEOUT_SECONDS,
RPC_TIMEOUT_SECONDS,
)
_SIDECAR_CHUNK_SIZE = 8 * 1024 * 1024
@@ -130,7 +135,7 @@ class SidecarClient:
sandbox_id: UUID,
session_id: UUID,
path: str,
timeout_seconds: float = 30.0,
timeout_seconds: float = RPC_TIMEOUT_SECONDS,
) -> list[FilesystemEntry]:
payload = FilesystemListRequest(session_id=session_id, path=path)
body = payload.model_dump_json().encode()
@@ -222,7 +227,7 @@ class SidecarClient:
archive_file: IO[bytes],
sha256_hex: str,
operation_label: str,
timeout_seconds: float = 300.0,
timeout_seconds: float = BULK_TRANSFER_TIMEOUT_SECONDS,
) -> None:
def body() -> Iterator[bytes]:
archive_file.seek(0)
@@ -248,7 +253,7 @@ class SidecarClient:
sandbox_id: UUID,
endpoint_path: str,
operation_label: str,
timeout_seconds: float = 300.0,
timeout_seconds: float = BULK_TRANSFER_TIMEOUT_SECONDS,
) -> None:
body = b""
self._post(
@@ -350,7 +355,7 @@ class SidecarClient:
def _timeout_for_post(remaining_seconds: float) -> httpx.Timeout:
return httpx.Timeout(
remaining_seconds,
connect=min(5.0, remaining_seconds),
connect=min(CONNECT_TIMEOUT_SECONDS, remaining_seconds),
read=remaining_seconds,
write=remaining_seconds,
)
@@ -17,6 +17,7 @@ from typing import Any
import httpx
from onyx.server.features.build.timeouts import CONNECT_TIMEOUT_SECONDS
from onyx.utils.logger import setup_logger
logger = setup_logger()
@@ -52,7 +53,7 @@ class PodEventBus:
auth: httpx.Auth | None,
*,
directory: str | None = None,
connect_timeout: float = 10.0,
connect_timeout: float = CONNECT_TIMEOUT_SECONDS,
event_read_timeout: float | None = None,
reload_auth: Callable[[], httpx.Auth | None] | None = None,
) -> None:
@@ -28,8 +28,8 @@ from onyx.server.features.build.configs import (
OPENCODE_PROMPT_INACTIVITY_TIMEOUT_SECONDS,
OPENCODE_SERVE_EVENT_READ_TIMEOUT,
OPENCODE_SERVER_USERNAME,
PROMPT_SLOT_KEEP_ALIVE_MAX_SECONDS,
PROMPT_SLOT_LEASE_SECONDS,
SSE_KEEPALIVE_INTERVAL,
)
from onyx.server.features.build.db.sandbox import get_sandbox_by_id
from onyx.server.features.build.sandbox.event_schema import (
@@ -48,6 +48,11 @@ from onyx.server.features.build.sandbox.opencode.serve_client import (
translate_opencode_event,
)
from onyx.server.features.build.sandbox.sse import SSEKeepalive
from onyx.server.features.build.timeouts import (
POLL_INTERVAL_SECONDS,
PROMPT_SLOT_FAST_FAIL_ACQUIRE_SECONDS,
PROMPT_SLOT_KEEP_ALIVE_MAX_SECONDS,
)
from onyx.server.metrics.craft_sandbox import (
SandboxProvisionPhase,
time_provision_phase,
@@ -63,13 +68,6 @@ _API_SERVER_HOSTNAME = os.environ.get("HOSTNAME", "unknown")
# opencode-serve boot lags backend Ready by ~1–3s warm, up to ~15s cold.
OPENCODE_SERVE_READY_TIMEOUT_SECONDS = 30
OPENCODE_SERVE_READY_POLL_INTERVAL_SECONDS = 0.25
# How long a new turn waits for the previous turn's slot before giving up.
PROMPT_SLOT_FAST_FAIL_ACQUIRE_SECONDS = 10.0
# Must exceed the lease so a reclaimed turn can wait out a dead holder's slot.
PROMPT_SLOT_WAIT_OUT_ORPHAN_SECONDS = PROMPT_SLOT_LEASE_SECONDS + 10.0
# Live attach streams are UI-facing. Coalesce adjacent text deltas just long
# enough to avoid one React update per tiny opencode token burst, while keeping
@@ -412,7 +410,7 @@ class _ServeMixin:
last_err = f"health_check returned status={status} for {url}"
except Exception as e:
last_err = f"{url}: {type(e).__name__}: {e}"
time.sleep(OPENCODE_SERVE_READY_POLL_INTERVAL_SECONDS)
time.sleep(POLL_INTERVAL_SECONDS)
finally:
for _, client in clients:
client.close()
@@ -808,7 +806,7 @@ class _ServeMixin:
opencode_session_id: str,
*,
directory: str,
keepalive_seconds: float = 15.0,
keepalive_seconds: float = SSE_KEEPALIVE_INTERVAL,
) -> Generator[SandboxEvent, None, None]:
"""Stream translated sandbox events for an opencode session. Caller closes
via ``GeneratorExit``. ``directory`` is required: opencode-serve scopes
@@ -4,7 +4,10 @@ from collections.abc import Iterable
from pathlib import Path
from onyx.db.models import ExternalApp
from onyx.server.features.build.configs import SANDBOX_APPROVAL_WAIT_TIMEOUT_SECONDS
from onyx.server.features.build.configs import (
SANDBOX_APPROVAL_WAIT_MARGIN_SECONDS,
SANDBOX_APPROVAL_WAIT_TIMEOUT_SECONDS,
)
from onyx.utils.logger import setup_logger
logger = setup_logger()
@@ -156,7 +159,9 @@ def generate_agent_instructions(
)
content = content.replace(
"{{APPROVAL_CLIENT_TIMEOUT_SECONDS}}",
str(SANDBOX_APPROVAL_WAIT_TIMEOUT_SECONDS + 20),
str(
SANDBOX_APPROVAL_WAIT_TIMEOUT_SECONDS + SANDBOX_APPROVAL_WAIT_MARGIN_SECONDS
),
)
content = content.replace("{{DISABLED_TOOLS_SECTION}}", disabled_tools_section)
content = content.replace("{{CONNECTABLE_APPS_LIST}}", connectable_apps_section)
@@ -13,12 +13,11 @@ import httpx
from onyx.error_handling.error_codes import OnyxErrorCode
from onyx.error_handling.exceptions import OnyxError
from onyx.server.features.build.timeouts import CONNECT_TIMEOUT_SECONDS
from onyx.utils.logger import setup_logger
logger = setup_logger()
_PROBE_TIMEOUT_SECONDS = 5.0
_validated = False
_validated_lock = threading.Lock()
@@ -28,7 +27,7 @@ def _probe_health(base_url: str) -> httpx.Response | None:
try:
return httpx.get(
f"{base_url}/health",
timeout=_PROBE_TIMEOUT_SECONDS,
timeout=CONNECT_TIMEOUT_SECONDS,
follow_redirects=True,
)
except httpx.HTTPError:
@@ -52,6 +52,7 @@ from onyx.server.features.build.scheduled_tasks.schedule import (
human_readable,
next_n_fires,
)
from onyx.server.features.build.timeouts import QUEUE_RESIDENCY_SECONDS
from onyx.utils.datetime import datetime_to_utc
from onyx.utils.logger import setup_logger
from shared_configs.contextvars import get_current_tenant_id
@@ -63,11 +64,6 @@ logger = setup_logger()
# Constants
# ---------------------------------------------------------------------------
# Number of seconds before a queued executor task is dropped. Mirrors
# ``check-for-pruning`` and ``check-for-indexing`` — 15 minutes is plenty for
# the worker to pick up the task; anything still queued past that is dead.
EXECUTOR_TASK_EXPIRES_SECONDS = 900
# Number of future fires returned by the detail endpoint for UI preview.
NEXT_RUNS_PREVIEW_COUNT = 3
@@ -343,7 +339,7 @@ def _enqueue_executor(run_id: UUID) -> None:
kwargs={"run_id": str(run_id), "tenant_id": tenant_id},
queue=OnyxCeleryQueues.SCHEDULED_TASKS,
priority=OnyxCeleryPriority.MEDIUM,
expires=EXECUTOR_TASK_EXPIRES_SECONDS,
expires=QUEUE_RESIDENCY_SECONDS,
)
@@ -60,17 +60,12 @@ from onyx.server.features.build.sandbox.event_schema import (
from onyx.server.features.build.session.locks import session_creation_lock
from onyx.server.features.build.session.manager import SessionManager
from onyx.server.features.build.session.streaming import BuildStreamingState
from onyx.server.features.build.timeouts import TURN_BUDGET_SECONDS
from onyx.utils.logger import setup_logger
logger = setup_logger()
# Per-run wall-clock budget (monotonic). Tasks that blow past this are
# marked ``failed (error_class=timeout)``. The stuck-run sweeper uses a
# slightly larger threshold (45 min) so a hung run that fails to honor the
# budget still gets cleaned up out-of-band.
DEFAULT_EXECUTOR_BUDGET_SECONDS = 30 * 60
# Summary length on the run row (per spec: ~120 chars of final agent
# message).
SUMMARY_MAX_CHARS = 120
@@ -167,7 +162,7 @@ def _notify(
def run_scheduled_task_logic(
run_id: UUID,
*,
budget_seconds: int = DEFAULT_EXECUTOR_BUDGET_SECONDS,
budget_seconds: int = TURN_BUDGET_SECONDS,
) -> None:
"""Execute a single scheduled-task run end-to-end.
@@ -632,10 +627,3 @@ def _drive_agent(
# approval gate, budget exceeded, exception). Matches the
# interactive path's finally in _stream_cli_agent_response.
prompt_slot_cm.__exit__(None, None, None)
# Re-export for the Celery task wrapper.
__all__ = [
"DEFAULT_EXECUTOR_BUDGET_SECONDS",
"run_scheduled_task_logic",
]
@@ -25,6 +25,7 @@ from onyx.db.scheduled_task import get_scheduled_run_context
from onyx.error_handling.error_codes import OnyxErrorCode
from onyx.error_handling.exceptions import OnyxError
from onyx.redis.redis_pool import get_redis_client
from onyx.server.features.build.configs import SSE_KEEPALIVE_INTERVAL
from onyx.server.features.build.db.build_session import (
allocate_nextjs_port,
get_build_session,
@@ -75,6 +76,7 @@ from onyx.server.features.build.session.sandbox_lifecycle import (
rollback_failed_provisioning,
)
from onyx.server.features.build.session.streaming import SSE_KEEPALIVE
from onyx.server.features.build.timeouts import POLL_INTERVAL_SECONDS
from onyx.server.features.build.utils import sanitize_filename, validate_file
from onyx.skills.push import build_user_skills_payload
from onyx.utils.logger import setup_logger
@@ -1026,10 +1028,6 @@ def get_session_scheduled_run_context(
)
LIVE_STREAM_READY_POLL_SECONDS = 1.0
LIVE_STREAM_KEEPALIVE_SECONDS = 15.0
def _scheduled_run_is_running(
*,
db_session: Session,
@@ -1094,7 +1092,7 @@ def get_session_scheduled_run_events(
break
yield SSE_KEEPALIVE
time.sleep(LIVE_STREAM_READY_POLL_SECONDS)
time.sleep(POLL_INTERVAL_SECONDS)
try:
with get_session_with_current_tenant() as stream_db_session:
@@ -1102,7 +1100,7 @@ def get_session_scheduled_run_events(
for chunk in session_manager.subscribe_to_existing_session_events(
session_id,
user_id,
keepalive_seconds=LIVE_STREAM_KEEPALIVE_SECONDS,
keepalive_seconds=SSE_KEEPALIVE_INTERVAL,
):
yield chunk
stream_db_session.expire_all()
@@ -36,7 +36,6 @@ from onyx.server.features.build.configs import (
MAX_TOTAL_UPLOAD_SIZE_BYTES,
MAX_UPLOAD_FILES_PER_SESSION,
OPENCODE_DISABLED_TOOLS,
PROMPT_SLOT_KEEP_ALIVE_MAX_SECONDS,
)
from onyx.server.features.build.db.build_session import (
allocate_nextjs_port,
@@ -62,10 +61,7 @@ from onyx.server.features.build.sandbox.models import (
FilesystemEntry,
PromptAttachment,
)
from onyx.server.features.build.sandbox.serve_transport import (
PROMPT_SLOT_FAST_FAIL_ACQUIRE_SECONDS,
PromptSlot,
)
from onyx.server.features.build.sandbox.serve_transport import PromptSlot
from onyx.server.features.build.sandbox.snapshot_manager import SnapshotManager
from onyx.server.features.build.sandbox.util.agent_instructions import (
build_connectable_apps_list,
@@ -95,6 +91,10 @@ from onyx.server.features.build.session.sandbox_lifecycle import (
hydrate_managed_content,
)
from onyx.server.features.build.session.streaming import BuildStreamingState
from onyx.server.features.build.timeouts import (
PROMPT_SLOT_FAST_FAIL_ACQUIRE_SECONDS,
PROMPT_SLOT_KEEP_ALIVE_MAX_SECONDS,
)
from onyx.skills.push import build_user_skills_payload
from onyx.utils.logger import setup_logger
from onyx.utils.threadpool_concurrency import start_thread_with_context
@@ -103,6 +103,9 @@ from shared_configs.contextvars import get_current_tenant_id
logger = setup_logger()
# Webapp-ready probe on the UI-poll hot path; any response (even 404) counts.
_WEBAPP_PROBE_TIMEOUT_SECONDS = 2.0
_DISPOSE_PENDING_TTL_SECONDS = 24 * 3600
@@ -1431,7 +1434,7 @@ class SessionManager:
f"{internal_url}/api/build/sessions/{session_id}/webapp"
"/_next/static/onyx-ready-probe.js"
)
with httpx.Client(timeout=2.0) as client:
with httpx.Client(timeout=_WEBAPP_PROBE_TIMEOUT_SECONDS) as client:
client.get(probe_url)
return True
except Exception:
@@ -32,7 +32,6 @@ from onyx.db.models import BuildSession
from onyx.sandbox_proxy import approval_cache
from onyx.server.features.build import connect_app
from onyx.server.features.build.configs import (
PROMPT_SLOT_KEEP_ALIVE_MAX_SECONDS,
SANDBOX_HEARTBEAT_REFRESH_INTERVAL_SECONDS,
)
from onyx.server.features.build.db.build_session import (
@@ -69,6 +68,7 @@ from onyx.server.features.build.sandbox.models import PromptAttachment
from onyx.server.features.build.sandbox.opencode.serve_client import _merge_field_meta
from onyx.server.features.build.sandbox.serve_transport import PromptSlot
from onyx.server.features.build.sandbox.sse import SSEKeepalive
from onyx.server.features.build.timeouts import PROMPT_SLOT_KEEP_ALIVE_MAX_SECONDS
from onyx.utils.logger import setup_logger
from onyx.utils.threadpool_concurrency import start_thread_with_context
from shared_configs.contextvars import CURRENT_TENANT_ID_CONTEXTVAR
@@ -0,0 +1,121 @@
"""Craft timeout registry.
Every Craft time constant is one of four things:
1. a **root** — a policy number a human chooses,
2. a **derivation** — arithmetic over a root, encoded here so tuning the root
moves every dependent value with it,
3. the **fixed kit** — genuinely independent short timeouts in a small
taxonomy (connect / RPC / bulk transfer / mutex lease), or
4. a **cadence** — poll and retry intervals.
Residence rule: a constant lives here only if it is used by multiple files,
is a derivation (or feeds one), or belongs to the shared client-I/O taxonomy.
A single-file, free-standing constant is defined at its point of use.
Env-tunable values (anything with its own ``os.environ`` read) live in
``configs.py`` even when their default is derived (e.g.
``OPENCODE_PROMPT_INACTIVITY_TIMEOUT_SECONDS``); pure non-tunable derivations
over those roots live here (e.g. ``RUNNER_STALE_AFTER_SECONDS``).
Ordering invariants (asserted by
``tests/unit/onyx/server/features/craft/test_timeout_registry.py``):
cadences < probes/connect < mutex leases
< QUEUE_RESIDENCY < TURN_BUDGET < ACTIVE_TURN_TTL < REQUEST_ID_TTL
RUNNER_STALE_AFTER == 6 x SSE_KEEPALIVE_INTERVAL
OPENCODE_PROMPT_INACTIVITY > SANDBOX_APPROVAL_WAIT (configs.py)
SANDBOX_HEARTBEAT_REFRESH << SANDBOX_IDLE_TIMEOUT (configs.py)
The client-I/O taxonomy (connect / RPC / bulk) is orthogonal to the ordering
chain — a bulk transfer legitimately outlasts most other bounds.
"""
from onyx.server.features.build.configs import (
PROMPT_SLOT_LEASE_SECONDS,
SSE_KEEPALIVE_INTERVAL,
)
# =============================================================================
# Turn family (root: TURN_BUDGET_SECONDS)
# =============================================================================
# Wall-clock budget of one agent turn, interactive or scheduled.
TURN_BUDGET_SECONDS = 30 * 60
# Margin past the budget before out-of-band cleanup (stuck-run sweeper, turn
# TTL expiry) treats a run/turn as abandoned: a well-behaved turn that hits
# its own budget must always finalize itself first.
TURN_RECLAIM_SLACK_SECONDS = 15 * 60
# Redis TTL of the turn record + active-turn pointer. Refreshed on every
# heartbeat; the absolute floor is the budget plus slack so a live
# budget-compliant turn's admission block never evaporates under it.
ACTIVE_TURN_TTL_SECONDS = TURN_BUDGET_SECONDS + TURN_RECLAIM_SLACK_SECONDS
# Post-terminal retention of the turn record and the client_request_id →
# turn_id dedupe key: an idempotent send-message retry (or the attach stream
# reading a FAILED turn's error) must still resolve after the turn finishes.
TURN_RETENTION_SECONDS = 15 * 60
REQUEST_ID_TTL_SECONDS = ACTIVE_TURN_TTL_SECONDS + TURN_RETENTION_SECONDS
# Ceiling for background prompt-slot renewal by holders whose progress is
# client-paced or opaque (session delete, subagent turns). It IS the turn
# budget: a slot held longer than any turn may run is leaked.
PROMPT_SLOT_KEEP_ALIVE_MAX_SECONDS = TURN_BUDGET_SECONDS
# =============================================================================
# Liveness cadence family (root: SSE_KEEPALIVE_INTERVAL, configs.py)
# =============================================================================
# A turn runner heartbeats on every consumed event, and the transport emits a
# keepalive at least every SSE_KEEPALIVE_INTERVAL even while the agent is
# silent — so the runner is dead after k missed keepalives, not after any
# turn-length-derived time. Derived so env-tuning the keepalive moves this
# with it; a fixed value would silently break turn recovery.
RUNNER_STALE_AFTER_SECONDS = 6 * SSE_KEEPALIVE_INTERVAL
# =============================================================================
# Queue residency family (root: QUEUE_RESIDENCY_SECONDS)
# =============================================================================
# One policy, three enforcement points: how long a queued scheduled run may
# sit before (a) Celery expires the message unexecuted and (b) the stuck-run
# sweeper reclaims its QUEUED row. They must move together or
# expired-but-unswept / swept-but-unexpired windows appear.
QUEUE_RESIDENCY_SECONDS = 15 * 60
# =============================================================================
# Fixed kit — client I/O taxonomy
# =============================================================================
# Establishing a connection to an in-cluster peer; also full-request probes
# on hot paths, where any response counts.
CONNECT_TIMEOUT_SECONDS = 5.0
# Unary request/response calls with small payloads (sidecar directory
# listings, managed-content pushes — the latter retried on failure).
RPC_TIMEOUT_SECONDS = 30.0
# Archive upload/download (session snapshots, opencode history). Sized to
# "snapshotting can take minutes"; the reaper is fail-closed on snapshot
# failure, so shrinking this converts slow-but-working snapshots into
# never-sleeping sandboxes.
BULK_TRANSFER_TIMEOUT_SECONDS = 300.0
# =============================================================================
# Fixed kit — mutexes (critical-section leases, never budget-derived)
# =============================================================================
# Prompt-slot acquire policy: a second turn racing a live one bounces fast
# ("concurrent turn in flight") instead of queueing; a reclaimed turn must
# instead wait out a dead holder's full lease before taking the slot.
PROMPT_SLOT_FAST_FAIL_ACQUIRE_SECONDS = 10.0
PROMPT_SLOT_WAIT_OUT_ORPHAN_SECONDS = PROMPT_SLOT_LEASE_SECONDS + 10.0
# =============================================================================
# Cadences
# =============================================================================
# Shared interval for poll-a-fast-local-resource loops (live-stream readiness;
# provisioning wait loops adopt it as they move onto the registry).
POLL_INTERVAL_SECONDS = 0.5
@@ -26,7 +26,7 @@ from onyx.server.features.build.sandbox.event_schema import (
PromptResponse,
)
from onyx.server.features.build.sandbox.event_schema import Error as SandboxError
from onyx.server.features.build.sandbox.serve_transport import (
from onyx.server.features.build.timeouts import (
PROMPT_SLOT_FAST_FAIL_ACQUIRE_SECONDS,
PROMPT_SLOT_WAIT_OUT_ORPHAN_SECONDS,
)
@@ -4,7 +4,6 @@ from collections.abc import Callable
from uuid import UUID, uuid4
from onyx.server.features.build.interactive_turns.state import (
REQUEST_ID_TTL_SECONDS,
TURN_STATUS_FAILED,
TURN_STATUS_QUEUED,
TURN_STATUS_RUNNING,
@@ -19,6 +18,7 @@ from onyx.server.features.build.interactive_turns.state import (
touch_turn,
)
from onyx.server.features.build.sandbox.models import PromptAttachment
from onyx.server.features.build.timeouts import REQUEST_ID_TTL_SECONDS
from tests.unit.fakes import FakeCache
@@ -122,7 +122,7 @@ def test_get_interactive_turn_events_waits_then_forwards_live_stream(
) -> Iterator[str]:
assert session_id_arg == session_id
assert user_id_arg == user_id
assert keepalive_seconds == turns_api.LIVE_STREAM_KEEPALIVE_SECONDS
assert keepalive_seconds == turns_api.SSE_KEEPALIVE_INTERVAL
finish_turn(
cache=cache,
turn_id=turn_id,
@@ -286,7 +286,7 @@ def test_get_interactive_turn_events_yields_failed_turn_error_after_stream_chunk
) -> Iterator[str]:
assert session_id_arg == session_id
assert user_id_arg == user_id
assert keepalive_seconds == turns_api.LIVE_STREAM_KEEPALIVE_SECONDS
assert keepalive_seconds == turns_api.SSE_KEEPALIVE_INTERVAL
finish_turn(
cache=cache,
turn_id=turn_id,
@@ -354,7 +354,7 @@ def test_get_interactive_turn_events_retries_runner_start_mid_stream(
) -> Iterator[str]:
assert session_id_arg == session_id
assert user_id_arg == user_id
assert keepalive_seconds == turns_api.LIVE_STREAM_KEEPALIVE_SECONDS
assert keepalive_seconds == turns_api.SSE_KEEPALIVE_INTERVAL
for i in range(3):
yield (
"event: message\n"
@@ -429,7 +429,7 @@ def test_get_interactive_turn_events_rate_limits_runner_retry_within_window(
) -> Iterator[str]:
assert session_id_arg == session_id
assert user_id_arg == user_id
assert keepalive_seconds == turns_api.LIVE_STREAM_KEEPALIVE_SECONDS
assert keepalive_seconds == turns_api.SSE_KEEPALIVE_INTERVAL
for i in range(3):
yield (
"event: message\n"
@@ -8,13 +8,13 @@ from uuid import uuid4
from onyx.server.features.build.configs import (
OPENCODE_PROMPT_INACTIVITY_TIMEOUT_SECONDS,
PROMPT_SLOT_KEEP_ALIVE_MAX_SECONDS,
)
from onyx.server.features.build.sandbox.event_schema import PromptResponse
from onyx.server.features.build.sandbox.kubernetes.kubernetes_sandbox_manager import (
KubernetesSandboxManager,
)
from onyx.server.features.build.sandbox.models import PromptAttachment
from onyx.server.features.build.timeouts import PROMPT_SLOT_KEEP_ALIVE_MAX_SECONDS
class _FakeServeClient:
@@ -0,0 +1,95 @@
"""Pins the Craft timeout registry: root values, derivations, and the
ordering invariants documented in ``onyx/server/features/build/timeouts.py``.
Value pins are the spec (hardcoded, not re-derived); relationship asserts are
the invariants that must survive any retuning of a root. Constants defined at
their point of use (single-file, free-standing) are imported from there.
"""
from onyx.server.features.build.configs import (
OPENCODE_PROMPT_INACTIVITY_TIMEOUT_SECONDS,
PROMPT_SLOT_LEASE_SECONDS,
SANDBOX_APPROVAL_WAIT_TIMEOUT_SECONDS,
SANDBOX_HEARTBEAT_REFRESH_INTERVAL_SECONDS,
SANDBOX_IDLE_TIMEOUT_SECONDS,
SSE_KEEPALIVE_INTERVAL,
)
from onyx.server.features.build.interactive_turns.api import (
LIVE_STREAM_RUNNER_RETRY_SECONDS,
)
from onyx.server.features.build.interactive_turns.state import (
TURN_LOCK_LEASE_SECONDS,
TURN_LOCK_WAIT_SECONDS,
)
from onyx.server.features.build.timeouts import (
ACTIVE_TURN_TTL_SECONDS,
BULK_TRANSFER_TIMEOUT_SECONDS,
CONNECT_TIMEOUT_SECONDS,
POLL_INTERVAL_SECONDS,
PROMPT_SLOT_FAST_FAIL_ACQUIRE_SECONDS,
PROMPT_SLOT_KEEP_ALIVE_MAX_SECONDS,
PROMPT_SLOT_WAIT_OUT_ORPHAN_SECONDS,
QUEUE_RESIDENCY_SECONDS,
REQUEST_ID_TTL_SECONDS,
RPC_TIMEOUT_SECONDS,
RUNNER_STALE_AFTER_SECONDS,
TURN_BUDGET_SECONDS,
)
def test_root_values_pin_the_spec() -> None:
assert TURN_BUDGET_SECONDS == 30 * 60
assert QUEUE_RESIDENCY_SECONDS == 15 * 60
def test_turn_derivations() -> None:
assert ACTIVE_TURN_TTL_SECONDS == 45 * 60
assert REQUEST_ID_TTL_SECONDS == 60 * 60
assert REQUEST_ID_TTL_SECONDS > ACTIVE_TURN_TTL_SECONDS
# The identity is the spec: a slot held past any possible turn is leaked.
assert PROMPT_SLOT_KEEP_ALIVE_MAX_SECONDS == TURN_BUDGET_SECONDS
def test_keepalive_derivations() -> None:
# The coupling IS the spec: env-tuning the keepalive must move runner
# staleness with it — a fixed stale threshold under a raised keepalive
# steals turns from healthy silent tool calls. The literal pins the
# multiplier and the default keepalive together.
assert RUNNER_STALE_AFTER_SECONDS == 6 * SSE_KEEPALIVE_INTERVAL
assert RUNNER_STALE_AFTER_SECONDS == 90.0
assert LIVE_STREAM_RUNNER_RETRY_SECONDS < RUNNER_STALE_AFTER_SECONDS
def test_approval_inactivity_coupling() -> None:
# A tool call parked at the approval proxy holds the event stream silent
# for the full approval window; the inactivity backstop must outlast it.
assert (
OPENCODE_PROMPT_INACTIVITY_TIMEOUT_SECONDS
> SANDBOX_APPROVAL_WAIT_TIMEOUT_SECONDS
)
def test_lock_invariants() -> None:
assert PROMPT_SLOT_WAIT_OUT_ORPHAN_SECONDS > PROMPT_SLOT_LEASE_SECONDS
assert PROMPT_SLOT_FAST_FAIL_ACQUIRE_SECONDS < PROMPT_SLOT_LEASE_SECONDS
assert TURN_LOCK_WAIT_SECONDS < TURN_LOCK_LEASE_SECONDS
def test_ordering_chain() -> None:
mutex_leases = (TURN_LOCK_LEASE_SECONDS, PROMPT_SLOT_LEASE_SECONDS)
assert POLL_INTERVAL_SECONDS < CONNECT_TIMEOUT_SECONDS
assert CONNECT_TIMEOUT_SECONDS < min(mutex_leases)
assert max(mutex_leases) < QUEUE_RESIDENCY_SECONDS
assert QUEUE_RESIDENCY_SECONDS < TURN_BUDGET_SECONDS
assert TURN_BUDGET_SECONDS < ACTIVE_TURN_TTL_SECONDS
assert ACTIVE_TURN_TTL_SECONDS < REQUEST_ID_TTL_SECONDS
assert RPC_TIMEOUT_SECONDS < BULK_TRANSFER_TIMEOUT_SECONDS
def test_idle_family_invariants() -> None:
# Heartbeat refresh must be far under the idle timeout or live sandboxes
# read as idle between refreshes.
assert (
SANDBOX_HEARTBEAT_REFRESH_INTERVAL_SECONDS * 10 <= SANDBOX_IDLE_TIMEOUT_SECONDS
)