diff --git a/.pr/design.md b/.pr/design.md new file mode 100644 index 0000000000..02cbade3cc --- /dev/null +++ b/.pr/design.md @@ -0,0 +1,126 @@ +# Slice 1 design — per-operation timing instrumentation (issue #4589) + +Scope for this PR: **conversation lifecycle create / delete / close**, plus +**event-service load**, **ConversationInfo compose**, and **idle eviction** on +the agent-server, plus the shared telemetry contract that later slices reuse. + +## Design decisions (confirmed with requester) + +### 1. Raw numeric durations are allowlisted for lifecycle ops only + +The existing contract in `telemetry/models.py` deliberately buckets magnitudes +(`ConversationOutcomeProperties`) because "raw counts joined with a timestamp are a +re-identification vector". This issue is a deliberate, documented departure for +*operation latency* only: + +- New field type `DurationMs = Annotated[int, Field(ge=0, le=MAX_DURATION_MS)]` + (integer milliseconds, 24h cap, validation-enforced like every other property). +- A single new event type `agent_server.operation_timing` carrying **no + high-cardinality dimensions**: no `conversation_ref`, no event counts, no + user id (the factory's anonymous per-process `distinct_id` is used). +- A lone duration is not re-identifying (unlike counts joined with a timestamp), + so lifecycle-op latency passes the allowlist. Any future dimension must be + approved through `EXPECTED_PROPERTY_NAMES` and the schema tests. + +The Pydantic schema *is* the allowlist (`models.py` docstring): a leak becomes a +construction-time `ValidationError`, not a review-time observation. + +### 2. Stuck vs slow: per-operation watchdog budget, default 20s + +- Each operation supplies its **expected elapsed-time budget** (`budget_ms`). If + omitted, `DEFAULT_STUCK_BUDGET_MS = 20_000` is used. +- A background watchdog task arms `asyncio.sleep(budget_ms / 1000)`. If the + operation is still in-flight when the budget elapses, it emits a + `stuck=True` event with the elapsed duration so far and marks the timer stuck. + The measurement site is never blocked (watchdog is a separate task; emit is + non-blocking). +- On completion, a **completion event is always emitted** with the final + duration and `stuck` set to whether the watchdog fired. + +Emitting both events makes the two failure shapes distinguishable: + +| Case | Events | +|---|---| +| Fast completion | completion only (`stuck=False`, real ms) | +| Slow but completes | stuck event + completion event (`stuck=True`, real ms) | +| Deadlock (no completion) | stuck event only — no completion event ever arrives | + +The completion event list is the percentile source (p50/p95/p99 over real +durations); the stuck events are the alert stream. + +## Wire contract + +New event (one type, `operation` property names the metric): + +``` +agent_server.operation_timing + kind: "operation_timing" + operation: SafeToken # "conversation_create" | "conversation_delete" | "conversation_close" + # | "event_service_load" | "conversation_info_compose" | "conversation_evict" + duration_ms: DurationMs # raw ms (wall clock) + stuck: bool + stuck_budget_ms: DurationMs # budget that was armed (informational) + evicted_count: Bucket|null # only on conversation_evict; bucketed, never raw +``` + +## Slice-1 mapping to historical failures (#4514 -> fixed by #4570) + +The #4570 fix replaced the global `_lifecycle_lock` with per-conversation +`_conversation_lifecycle(cid)` plus an exclusive `_exclusive_lifecycle()`. The +operations that serialize through those locks are the ones the issue wants timed: + +- `conversation_create` — `ConversationService._start_conversation` + (the POST /conversations path; both new-create and resume go through the + lifecycle lock, so a wedged lock makes start slow/stuck). +- `conversation_delete` — `ConversationService.delete_conversation`. +- `conversation_close` — the per-conversation event-service teardown + (`EventService.__aexit__`) performed under `_exclusive_lifecycle()` at + `ConversationService.__aexit__` (server shutdown). A wedged close here is the + original "stuck close() blocked everything" shape; timing each conversation's + close individually makes the blocked one visible without blocking measurement. +- `event_service_load` — `ConversationService._get_or_load_event_service`, + timed from before `_conversation_lifecycle` acquisition through + `_get_or_load_event_service_locked` (disk hydration + runtime prep). A load + wedged on the lifecycle lock (#4514) surfaces as stuck-without-completion. +- `conversation_info_compose` — both `_compose_conversation_info*` call sites in + `_conversation_info` (live in-memory path incl. `get_state()`, and the + persisted `to_thread` path). The #4417 shape was a GC wedge composing + `ConversationInfo` off the event loop. +- `conversation_evict` — one pass of `_evict_idle_conversations` (only when + at least one conversation is evicted; empty passes emit nothing so they do + not deflate the latency percentile). Emits `evicted_count` as a **bucketed** + magnitude (`COUNT_BOUNDS`), never a raw count — raw counts joined with a + timestamp are the original re-identification vector. + +## Bucketed counts (design decision) + +Per-operation **duration** is allowlisted raw (see decision 1). Magnitudes that +are *counts* — e.g. how many conversations an eviction pass closed — stay on +the bucketed vocabulary (`Bucket` / `COUNT_BOUNDS`), because a raw count joined +with a timestamp is re-identifying while a lone elapsed time is not. The +`operation_timing` event carries at most one such magnitude per operation. + +All sites resolve the sink/factory lazily at emit time (`get_telemetry_sink()`, +`get_event_factory()`), matching the existing `_maybe_subscribe_telemetry` +convention so the live consent decision is honored. Emit is best-effort and +cannot raise out (mirrors `TelemetrySink.emit` "never raise"). + +## Files + +- `telemetry/models.py` — `DurationMs`, `OperationTimingProperties`, + `EventName.OPERATION_TIMING`, union + `EXPECTED_PROPERTY_NAMES`. +- `telemetry/timing.py` (new) — `timed_operation` asynccontextmanager + + watchdog + default emitter (`DEFAULT_EMITTER`, monkeypatchable in tests). +- `telemetry/__init__.py` — re-export `timed_operation`. +- `conversation_service.py` — wrap create/delete/close. +- Tests: `tests/agent_server/telemetry/test_telemetry_timing.py` (helper + + model + schema allowlist), service-level tests in + `tests/agent_server/test_conversation_service.py` (blocked-conversation + scenario asserts stuck terminal value). + +## Follow-up slices (not in this PR) + +`event_service_load`, `event_search`/`bash_event_search`, `conversation_info_compose`, +`llm_call`, `switch_llm`, `stats_streaming`, `subscribe_init_push`, +`acp_restart_secret_lookup`, `model_info_discovery`, `summary_render`, +`conversation_evict`, lease/autosave/pubsub/condensation, and the #4588 gate wiring. \ No newline at end of file diff --git a/openhands-agent-server/openhands/agent_server/conversation_service.py b/openhands-agent-server/openhands/agent_server/conversation_service.py index 4f9abac659..0946439af8 100644 --- a/openhands-agent-server/openhands/agent_server/conversation_service.py +++ b/openhands-agent-server/openhands/agent_server/conversation_service.py @@ -45,7 +45,13 @@ get_event_factory, get_telemetry_sink, ) -from openhands.agent_server.telemetry.sanitizer import model_family, safe_token +from openhands.agent_server.telemetry.sanitizer import ( + COUNT_BOUNDS, + bucket, + model_family, + safe_token, +) +from openhands.agent_server.telemetry.timing import timed_operation from openhands.agent_server.utils import safe_rmtree, utc_now from openhands.sdk import LLM, AgentContext, Event, Message from openhands.sdk.agent import ACPAgent @@ -886,10 +892,14 @@ async def _conversation_info( # Direct embedders/tests can inject a live EventService without a # persisted state file. There is no disk snapshot to list in that # case, so retain the live-state fallback. - state = await event_service.get_state() - conversation_info = await asyncio.to_thread( - _compose_conversation_info_sync, event_service.stored, state, children - ) + async with timed_operation("conversation_info_compose"): + state = await event_service.get_state() + conversation_info = await asyncio.to_thread( + _compose_conversation_info_sync, + event_service.stored, + state, + children, + ) record.execution_status = conversation_info.execution_status return conversation_info @@ -918,9 +928,10 @@ async def _conversation_info( ) if state is None: return None - conversation_info = await asyncio.to_thread( - _compose_conversation_info, record.stored, state, children - ) + async with timed_operation("conversation_info_compose"): + conversation_info = await asyncio.to_thread( + _compose_conversation_info, record.stored, state, children + ) record.state_signature = signature record.stored_signature = stored_signature record.cached_info = conversation_info @@ -1105,8 +1116,10 @@ async def _get_or_load_event_service( and conversation_id not in self._conversation_records ): return None - async with self._conversation_lifecycle(conversation_id): - return await self._get_or_load_event_service_locked(conversation_id) + # Timed before lock acquisition so a lock-wedged load surfaces as stuck. + async with timed_operation("event_service_load"): + async with self._conversation_lifecycle(conversation_id): + return await self._get_or_load_event_service_locked(conversation_id) async def _get_or_load_event_service_locked( self, @@ -1353,12 +1366,14 @@ async def _notify_and_log_errors(): async def start_conversation( self, request: StartConversationRequest ) -> tuple[ConversationInfo, bool]: - return await self._start_conversation(request) + async with timed_operation("conversation_create"): + return await self._start_conversation(request) async def start_acp_conversation( self, request: StartConversationRequest ) -> tuple[ConversationInfo, bool]: - return await self._start_conversation(request) + async with timed_operation("conversation_create"): + return await self._start_conversation(request) async def _start_conversation( self, @@ -1738,6 +1753,10 @@ async def resume_conversation(self, conversation_id: UUID) -> bool: return bool(await self._get_or_load_event_service(conversation_id)) async def delete_conversation(self, conversation_id: UUID) -> bool: + async with timed_operation("conversation_delete"): + return await self._delete_conversation(conversation_id) + + async def _delete_conversation(self, conversation_id: UUID) -> bool: event_services = self._event_services if event_services is None: raise ValueError("inactive_service") @@ -2107,37 +2126,41 @@ async def _evict_idle_conversations(self, ttl_seconds: float) -> None: and event_service.idle_seconds() >= ttl_seconds and event_service.is_idle_evictable() ] - for conversation_id in to_evict: - event_service = event_services.pop(conversation_id, None) - if event_service is None: - continue - # Preserve runtime-only state so rehydration is faithful: - # sync the catalog to the current stored (switch_acp_model / - # secret updates replace it) and hand back credential bindings - # (close() clears them). - record = self._conversation_records.get(conversation_id) - if record is not None: - record.stored = event_service.stored - record.cached_info = None - bindings = dict(event_service.credential_bindings) - try: - await event_service.__aexit__(None, None, None) - except Exception: - logger.warning( - "Failed to evict idle conversation %s", - conversation_id, - exc_info=True, - ) - else: - logger.info( - "Evicted idle conversation %s (idle >= %.0fs)", - conversation_id, - ttl_seconds, - ) - if bindings: - pending = self._credential_bindings.setdefault(conversation_id, {}) - for secret_name, binding in bindings.items(): - pending.setdefault(secret_name, binding) + if not to_evict: + return + async with timed_operation("conversation_evict") as timer: + for conversation_id in to_evict: + await self._evict_one_idle_conversation(conversation_id) + # Magnitude bucketed; raw counts are re-identifying. + timer.evicted_count = bucket(len(to_evict), COUNT_BOUNDS) + + async def _evict_one_idle_conversation(self, conversation_id: UUID) -> None: + event_services = self._event_services + assert event_services is not None + event_service = event_services.pop(conversation_id, None) + if event_service is None: + return + # Preserve runtime-only state (catalog + credential bindings) for + # faithful rehydration; __aexit__ clears the bindings. + record = self._conversation_records.get(conversation_id) + if record is not None: + record.stored = event_service.stored + record.cached_info = None + bindings = dict(event_service.credential_bindings) + try: + await event_service.__aexit__(None, None, None) + except Exception: + logger.warning( + "Failed to evict idle conversation %s", + conversation_id, + exc_info=True, + ) + else: + logger.info("Evicted idle conversation %s", conversation_id) + if bindings: + pending = self._credential_bindings.setdefault(conversation_id, {}) + for secret_name, binding in bindings.items(): + pending.setdefault(secret_name, binding) async def __aexit__(self, exc_type, exc_value, traceback): if self._eviction_task is not None: @@ -2157,11 +2180,15 @@ async def __aexit__(self, exc_type, exc_value, traceback): if event_services is None: return services = tuple(event_services.items()) + + async def _close_event_service(event_service: EventService) -> None: + # Time each close individually so a wedged one surfaces as + # stuck despite the gather waiting on all of them. + async with timed_operation("conversation_close"): + await event_service.__aexit__(exc_type, exc_value, traceback) + results = await asyncio.gather( - *[ - event_service.__aexit__(exc_type, exc_value, traceback) - for _, event_service in services - ], + *[_close_event_service(event_service) for _, event_service in services], return_exceptions=True, ) failed_ids = { diff --git a/openhands-agent-server/openhands/agent_server/telemetry/__init__.py b/openhands-agent-server/openhands/agent_server/telemetry/__init__.py index 3e771c0abb..509435a093 100644 --- a/openhands-agent-server/openhands/agent_server/telemetry/__init__.py +++ b/openhands-agent-server/openhands/agent_server/telemetry/__init__.py @@ -57,6 +57,12 @@ ConversationTelemetryContext, TelemetrySubscriber, ) +from openhands.agent_server.telemetry.timing import ( + DEFAULT_EMITTER, + DEFAULT_STUCK_BUDGET_MS, + OperationTimingResult, + timed_operation, +) __all__ = [ @@ -87,4 +93,8 @@ "reset_telemetry_sink", "resolve", "shutdown_telemetry_sink", + "timed_operation", + "DEFAULT_STUCK_BUDGET_MS", + "DEFAULT_EMITTER", + "OperationTimingResult", ] diff --git a/openhands-agent-server/openhands/agent_server/telemetry/models.py b/openhands-agent-server/openhands/agent_server/telemetry/models.py index 14a163b1b5..cb45eb5ae6 100644 --- a/openhands-agent-server/openhands/agent_server/telemetry/models.py +++ b/openhands-agent-server/openhands/agent_server/telemetry/models.py @@ -62,6 +62,18 @@ Bucket = SafeToken """A bucketed magnitude such as ``11-50``. Never a raw count.""" +MAX_DURATION_MS: Final[int] = 86_400_000 +"""Upper bound for raw durations: 24h in ms. A bound keeps the field from ever +holding an absurd magnitude while leaving real-life operations uncapped.""" + +DurationMs = Annotated[int, Field(ge=0, le=MAX_DURATION_MS)] +"""A raw wall-clock duration in integer milliseconds. + +The one magnitude deliberately exempted from bucketing, reserved for +per-operation latency (see :class:`OperationTimingProperties`). Unlike raw +counts, a duration cannot be joined back to an identity. +""" + class EventName(StrEnum): """Stable event names. The wire value is the member value.""" @@ -73,6 +85,7 @@ class EventName(StrEnum): CONVERSATION_FINISHED = "agent_server.conversation_finished" CONVERSATION_FAILED = "agent_server.conversation_failed" CONVERSATION_ERROR = "agent_server.conversation_error" + OPERATION_TIMING = "agent_server.operation_timing" REQUEST_FAILED = "agent_server.request_failed" @@ -232,12 +245,34 @@ class RequestFailedProperties(_BaseProperties): error_id: SafeToken | None = None +class OperationTimingProperties(_BaseProperties): + """Per-operation latency in raw milliseconds. + + Durations are allowlisted raw: a lone elapsed-time value carries no + re-identification surface, unlike raw counts joined with a timestamp. This + event therefore carries no ``conversation_ref`` and reports magnitudes only + as coarse :data:`Bucket` values (e.g. ``evicted_count``), never raw counts. + ``stuck`` distinguishes a deadlock (watchdog fired, no completion event) + from a merely slow operation (watchdog fired, operation then completed). + """ + + kind: Literal["operation_timing"] = "operation_timing" + + operation: SafeToken + duration_ms: DurationMs + stuck: bool + stuck_budget_ms: DurationMs + evicted_count: Bucket | None = None + """Idle conversations closed by a ``conversation_evict`` pass, bucketed.""" + + DiagnosticProperties = Annotated[ ServerLifecycleProperties | ConversationStartedProperties | ConversationOutcomeProperties | ErrorProperties - | RequestFailedProperties, + | RequestFailedProperties + | OperationTimingProperties, Field(discriminator="kind"), ] @@ -324,5 +359,10 @@ def to_payload(self) -> dict[str, object]: "route_template", "method", "status_code", + "operation", + "duration_ms", + "stuck", + "stuck_budget_ms", + "evicted_count", } ) diff --git a/openhands-agent-server/openhands/agent_server/telemetry/timing.py b/openhands-agent-server/openhands/agent_server/telemetry/timing.py new file mode 100644 index 0000000000..09563ce106 --- /dev/null +++ b/openhands-agent-server/openhands/agent_server/telemetry/timing.py @@ -0,0 +1,145 @@ +"""Wall-clock per-operation timing with an in-flight "stuck" watchdog. + +Wrap a coroutine body in :func:`timed_operation` to emit elapsed wall-clock +duration on completion, plus a ``stuck`` signal if the operation exceeds its +budget. The watchdog runs as a separate task and never blocks the measured +operation, so a wedged operation stays observable. Emission is best-effort and +resolves the sink/factory lazily so the live consent decision is honored. +""" + +import asyncio +import time +from collections.abc import AsyncIterator, Callable +from contextlib import asynccontextmanager, suppress +from dataclasses import dataclass +from typing import Final + +from openhands.agent_server.telemetry import models as m +from openhands.agent_server.telemetry.service import ( + get_event_factory, + get_telemetry_sink, +) +from openhands.sdk.logger import get_logger + + +logger = get_logger(__name__) + +#: Budget applied when a call site does not supply its own expected elapsed time. +DEFAULT_STUCK_BUDGET_MS: Final[int] = 20_000 + + +@dataclass(frozen=True, slots=True) +class OperationTimingResult: + """What a measurement site produced. Emitted verbatim as event properties.""" + + operation: str + duration_ms: int + stuck: bool + stuck_budget_ms: int + evicted_count: str | None = None + """Bucketed magnitude for ops that report one (e.g. evictions closed).""" + + +def _default_emitter(result: OperationTimingResult) -> None: + """Emit one :class:`OperationTimingProperties` event, best-effort. + + Resolves the process sink/factory at emit time so consent changes are + honored and a pre-init NoOp sink is never cached. Never raises: telemetry + must never break the measured operation. + """ + sink = get_telemetry_sink() + if not sink.enabled: + return + factory = get_event_factory() + if factory is None: + return + try: + properties = m.OperationTimingProperties( + operation=result.operation, + duration_ms=result.duration_ms, + stuck=result.stuck, + stuck_budget_ms=result.stuck_budget_ms, + evicted_count=result.evicted_count, + ) + sink.emit(factory.build(m.EventName.OPERATION_TIMING, properties)) + except Exception: + logger.debug("telemetry_operation_timing_failed", exc_info=True) + + +#: Tests swap this to capture events without touching the process sink. +DEFAULT_EMITTER: Callable[[OperationTimingResult], None] = _default_emitter + + +def _clamp_ms(seconds: float) -> int: + return min(round(seconds * 1000), m.MAX_DURATION_MS) + + +class OperationTimer: + """Timing state shared between the measured body and the watchdog task.""" + + def __init__( + self, + operation: str, + budget_ms: int, + emit: Callable[[OperationTimingResult], None], + ) -> None: + self.operation = operation + self.budget_ms = budget_ms + self._emit = emit + self._started = time.monotonic() + self.stuck = False + # Bucketed magnitude attached by the measured body before exit. + self.evicted_count: str | None = None + + @property + def duration_ms(self) -> int: + return _clamp_ms(time.monotonic() - self._started) + + def _emit_now(self, stuck: bool) -> None: + self._emit( + OperationTimingResult( + operation=self.operation, + duration_ms=self.duration_ms, + stuck=stuck, + stuck_budget_ms=self.budget_ms, + evicted_count=self.evicted_count, + ) + ) + + async def _watchdog(self) -> None: + await asyncio.sleep(self.budget_ms / 1000.0) + if self.stuck: + return + self.stuck = True + self._emit_now(stuck=True) + + +@asynccontextmanager +async def timed_operation( + operation: str, + *, + budget_ms: int | None = None, + emit: Callable[[OperationTimingResult], None] | None = None, +) -> AsyncIterator[OperationTimer]: + """Time ``operation`` in wall-clock ms with a stuck watchdog. + + Emits at most one ``stuck`` event (if the budget elapses while the body is + still running) and always one completion event on exit. A deadlock + therefore produces a stuck event with no completion event, while a merely + slow operation produces both — the two failure shapes stay distinguishable. + The watchdog is a separate task and never blocks the body. + + ``budget_ms`` is the expected elapsed time for this specific operation; + when omitted, :data:`DEFAULT_STUCK_BUDGET_MS` (20s) applies. ``emit`` is + injectable for tests and defaults to the process-wide emitter. + """ + emitter = emit if emit is not None else DEFAULT_EMITTER + timer = OperationTimer(operation, budget_ms or DEFAULT_STUCK_BUDGET_MS, emitter) + watchdog = asyncio.create_task(timer._watchdog()) + try: + yield timer + finally: + watchdog.cancel() + with suppress(asyncio.CancelledError): + await watchdog + timer._emit_now(stuck=timer.stuck) diff --git a/tests/agent_server/telemetry/test_telemetry_schema.py b/tests/agent_server/telemetry/test_telemetry_schema.py index d6619c115b..efa263d1e3 100644 --- a/tests/agent_server/telemetry/test_telemetry_schema.py +++ b/tests/agent_server/telemetry/test_telemetry_schema.py @@ -21,6 +21,7 @@ m.ConversationOutcomeProperties, m.ErrorProperties, m.RequestFailedProperties, + m.OperationTimingProperties, ] @@ -32,6 +33,18 @@ def _is_constrained_str(annotation: object) -> bool: return False +def _is_constrained_int(annotation: object) -> bool: + """True if the annotation is a *bounded* int, not a bare ``int``. + + ``OperationTimingProperties`` deliberately carries raw milliseconds as a + bounded ``Annotated[int, Field(ge=..., le=...)]`` (see ``DurationMs``). + """ + if get_origin(annotation) is Annotated: + args = get_args(annotation) + return args[0] is int and len(args) > 1 + return False + + def _permitted(annotation: object) -> bool: # Unwrap optionals / unions. origin = get_origin(annotation) @@ -41,7 +54,7 @@ def _permitted(annotation: object) -> bool: ) if origin is Literal: return all(isinstance(a, str) for a in get_args(annotation)) - if _is_constrained_str(annotation): + if _is_constrained_str(annotation) or _is_constrained_int(annotation): return True return annotation in (bool, int) diff --git a/tests/agent_server/telemetry/test_telemetry_timing.py b/tests/agent_server/telemetry/test_telemetry_timing.py new file mode 100644 index 0000000000..bb116ba2b3 --- /dev/null +++ b/tests/agent_server/telemetry/test_telemetry_timing.py @@ -0,0 +1,145 @@ +"""Per-operation timing: the watchdog contract and the event model. + +A deadlock must be distinguishable from a merely slow operation: a deadlock +produces a ``stuck`` event with no completion event, a slow operation produces +both, and a fast operation produces only the completion event. +""" + +import asyncio + +import pytest +from pydantic import ValidationError + +from openhands.agent_server.telemetry import models as m +from openhands.agent_server.telemetry.timing import ( + DEFAULT_STUCK_BUDGET_MS, + OperationTimingResult, + timed_operation, +) + + +def test_default_stuck_budget_is_20_seconds(): + assert DEFAULT_STUCK_BUDGET_MS == 20_000 + + +@pytest.mark.asyncio +async def test_fast_operation_emits_single_completion_event(): + events: list[OperationTimingResult] = [] + + async with timed_operation( + "conversation_create", budget_ms=1000, emit=events.append + ): + await asyncio.sleep(0.01) + + assert len(events) == 1 + (event,) = events + assert event.operation == "conversation_create" + assert event.stuck is False + assert event.stuck_budget_ms == 1000 + assert event.duration_ms >= 0 + + +@pytest.mark.asyncio +async def test_slow_operation_emits_stuck_then_completion(): + events: list[OperationTimingResult] = [] + + async with timed_operation("conversation_delete", budget_ms=50, emit=events.append): + await asyncio.sleep(0.2) + + assert len(events) == 2 + stuck, completion = events + assert stuck.stuck is True + assert stuck.duration_ms < completion.duration_ms + assert completion.stuck is True + assert completion.duration_ms >= stuck.duration_ms + + +@pytest.mark.asyncio +async def test_deadlock_emits_stuck_without_completion(): + """While the operation is still in flight, only the stuck event exists.""" + events: list[OperationTimingResult] = [] + cm = timed_operation("conversation_close", budget_ms=50, emit=events.append) + await cm.__aenter__() + + await asyncio.sleep(0.1) + + # The watchdog fired; no completion event has been emitted yet, so the + # operation looks deadlocked (stuck event present, completion absent). + assert len(events) == 1 + assert events[0].stuck is True + + await cm.__aexit__(None, None, None) + assert len(events) == 2 + assert events[1].stuck is True + + +@pytest.mark.asyncio +async def test_exception_still_emits_completion(): + events: list[OperationTimingResult] = [] + + with pytest.raises(RuntimeError, match="boom"): + async with timed_operation( + "conversation_create", budget_ms=1000, emit=events.append + ): + raise RuntimeError("boom") + + assert len(events) == 1 + assert events[0].stuck is False + + +@pytest.mark.asyncio +async def test_timer_exposes_duration_to_the_measured_body(): + cm = timed_operation("conversation_create", budget_ms=1000, emit=lambda _: None) + timer = await cm.__aenter__() + try: + assert timer.operation == "conversation_create" + assert timer.budget_ms == 1000 + assert timer.duration_ms >= 0 + assert timer.stuck is False + finally: + await cm.__aexit__(None, None, None) + + +def test_timer_exposes_bucketed_count_to_the_measured_body(): + captured = [] + + async def body(): + async with timed_operation( + "conversation_evict", budget_ms=1000, emit=captured.append + ) as timer: + timer.evicted_count = "1-5" + + asyncio.run(body()) + assert len(captured) == 1 + assert captured[0].evicted_count == "1-5" + + +def test_timing_properties_reject_out_of_bounds_durations(): + m.OperationTimingProperties( + operation="conversation_create", + duration_ms=0, + stuck=False, + stuck_budget_ms=20_000, + ) + with pytest.raises(ValidationError): + m.OperationTimingProperties( + operation="conversation_create", + duration_ms=-1, + stuck=False, + stuck_budget_ms=20_000, + ) + with pytest.raises(ValidationError): + m.OperationTimingProperties( + operation="conversation_create", + duration_ms=0, + stuck=False, + stuck_budget_ms=-1, + ) + # A leak shape (a path) cannot occupy the operation token. + with pytest.raises(ValidationError): + m.OperationTimingProperties( + operation="/Users/alice/src/secret-project/main.py", + duration_ms=5, + stuck=False, + stuck_budget_ms=20_000, + ) diff --git a/tests/agent_server/test_conversation_service.py b/tests/agent_server/test_conversation_service.py index e867187cf8..72e9c160a0 100644 --- a/tests/agent_server/test_conversation_service.py +++ b/tests/agent_server/test_conversation_service.py @@ -34,6 +34,9 @@ StoredConversation, UpdateConversationRequest, ) +from openhands.agent_server.telemetry.timing import ( + timed_operation as _real_timed_operation, +) from openhands.agent_server.utils import safe_rmtree as _safe_rmtree from openhands.sdk import LLM, Agent, AgentBase, Message from openhands.sdk.agent.acp_agent import ACPAgent @@ -3013,6 +3016,330 @@ async def test_delete_conversation_directory_removal_failure( assert mock_rmtree.call_count == 1 +class TestOperationTimingInstrumentation: + """Per-operation timing on the lifecycle hot paths.""" + + def _recording_timed_operation(self, events, default_budget_ms=None): + def wrapped(operation, *, budget_ms=None, emit=None): # noqa: ARG001 + return _real_timed_operation( + operation, + budget_ms=budget_ms or default_budget_ms, + emit=events.append, + ) + + return wrapped + + @pytest.mark.asyncio + async def test_start_conversation_emits_timing_completion( + self, conversation_service, tmp_path, monkeypatch + ): + """A real create emits one ``conversation_create`` completion event.""" + events = [] + monkeypatch.setattr( + "openhands.agent_server.conversation_service.timed_operation", + self._recording_timed_operation(events), + ) + workspace_dir = tmp_path / "workspace" + workspace_dir.mkdir() + request = StartConversationRequest( + agent=Agent(llm=LLM(model="gpt-4o", usage_id="test-llm"), tools=[]), + workspace=LocalWorkspace(working_dir=str(workspace_dir)), + confirmation_policy=NeverConfirm(), + ) + + info, is_new = await conversation_service.start_conversation(request) + + assert is_new is True + assert info.id is not None + assert len(events) == 1 + (event,) = events + assert event.operation == "conversation_create" + assert event.stuck is False + assert event.duration_ms >= 0 + assert event.stuck_budget_ms == 20_000 + + @pytest.mark.asyncio + async def test_delete_blocked_on_peer_lock_emits_stuck( + self, conversation_service, monkeypatch + ): + """A delete blocked on the lifecycle lock surfaces as stuck + no completion. + + This is the wedge shape: one conversation's long-running + operation holds the per-conversation lock, and the delete of that + conversation must be *visible* (stuck signal) without the measurement + site itself completing. + """ + events = [] + monkeypatch.setattr( + "openhands.agent_server.conversation_service.timed_operation", + self._recording_timed_operation(events, default_budget_ms=100), + ) + conversation_id = uuid4() + mock_service = AsyncMock(spec=EventService) + mock_service.conversation_dir = "/tmp/test_conversation" + mock_service.stored = StoredConversation( + id=conversation_id, + workspace=LocalWorkspace(working_dir="/tmp/test_workspace"), + confirmation_policy=NeverConfirm(), + initial_message=None, + metrics=None, + created_at=datetime(2025, 1, 1, 12, 0, 0, tzinfo=UTC), + updated_at=datetime(2025, 1, 1, 12, 30, 0, tzinfo=UTC), + ) + mock_service.get_state.return_value = ConversationState( + id=conversation_id, + agent=_sample_agent(), + workspace=mock_service.stored.workspace, + execution_status=ConversationExecutionStatus.IDLE, + confirmation_policy=mock_service.stored.confirmation_policy, + ) + conversation_service._event_services[conversation_id] = mock_service + + lock = conversation_service._get_conversation_lock(conversation_id) + await lock.acquire() + + task = asyncio.create_task( + conversation_service.delete_conversation(conversation_id) + ) + + # The watchdog fires while the delete is still blocked on the lock. + loop = asyncio.get_running_loop() + deadline = loop.time() + 5 + while not any(e.stuck for e in events) and loop.time() < deadline: + await asyncio.sleep(0.01) + assert any(e.stuck for e in events), "no stuck signal was emitted" + stuck = next(e for e in events if e.stuck) + assert stuck.operation == "conversation_delete" + assert not task.done(), "delete completed while its lock was held" + + with patch( + "openhands.agent_server.conversation_service.safe_rmtree" + ) as mock_rmtree: + mock_rmtree.return_value = True + lock.release() + result = await task + assert result is True + + # A slow-but-alive operation emits a stuck signal and then a + # completion that also carries stuck=True (it *was* stuck). + assert len(events) == 2 + signal, completion = events + assert signal.stuck is True + assert completion.operation == "conversation_delete" + assert completion.stuck is True + assert completion.duration_ms >= signal.duration_ms >= 100 + + @pytest.mark.asyncio + async def test_event_service_load_emits_timing_completion( + self, conversation_service, monkeypatch + ): + """A hydration read emits one ``event_service_load`` completion event.""" + events = [] + monkeypatch.setattr( + "openhands.agent_server.conversation_service.timed_operation", + self._recording_timed_operation(events), + ) + conversation_id = uuid4() + mock_service = AsyncMock(spec=EventService) + mock_service.conversation_dir = "/tmp/test_conversation" + mock_service.stored = StoredConversation( + id=conversation_id, + workspace=LocalWorkspace(working_dir="/tmp/test_workspace"), + confirmation_policy=NeverConfirm(), + ) + mock_service.get_state.return_value = ConversationState( + id=conversation_id, + agent=Agent(llm=LLM(model="gpt-4o", usage_id="test-llm"), tools=[]), + workspace=mock_service.stored.workspace, + execution_status=ConversationExecutionStatus.IDLE, + confirmation_policy=mock_service.stored.confirmation_policy, + ) + conversation_service._event_services[conversation_id] = mock_service + + loaded = await conversation_service._get_or_load_event_service(conversation_id) + + assert loaded is mock_service + assert len(events) == 1 + (event,) = events + assert event.operation == "event_service_load" + assert event.stuck is False + + @pytest.mark.asyncio + async def test_event_service_load_blocked_on_lock_emits_stuck( + self, conversation_service, monkeypatch + ): + """A hydration blocked on the lifecycle lock surfaces as stuck.""" + events = [] + monkeypatch.setattr( + "openhands.agent_server.conversation_service.timed_operation", + self._recording_timed_operation(events, default_budget_ms=100), + ) + conversation_id = uuid4() + mock_service = AsyncMock(spec=EventService) + mock_service.conversation_dir = "/tmp/test_conversation" + mock_service.stored = StoredConversation( + id=conversation_id, + workspace=LocalWorkspace(working_dir="/tmp/test_workspace"), + confirmation_policy=NeverConfirm(), + ) + mock_service.is_open.return_value = True + conversation_service._event_services[conversation_id] = mock_service + lock = conversation_service._get_conversation_lock(conversation_id) + await lock.acquire() + task = asyncio.create_task( + conversation_service._get_or_load_event_service(conversation_id) + ) + await asyncio.sleep(0.25) + assert any(e.stuck for e in events) + assert not task.done(), "load completed while its lock was held" + lock.release() + result = await task + assert result is mock_service + + @pytest.mark.asyncio + async def test_conversation_info_compose_emits_timing_completion( + self, conversation_service, monkeypatch + ): + """Composing a live row emits one ``conversation_info_compose`` event.""" + events = [] + monkeypatch.setattr( + "openhands.agent_server.conversation_service.timed_operation", + self._recording_timed_operation(events), + ) + conversation_id = uuid4() + mock_service = AsyncMock(spec=EventService) + mock_service.conversation_dir = "/tmp/test_conversation" + mock_service.stored = StoredConversation( + id=conversation_id, + workspace=LocalWorkspace(working_dir="/tmp/test_workspace"), + confirmation_policy=NeverConfirm(), + ) + mock_service.is_open.return_value = True + mock_service.get_state.return_value = ConversationState( + id=conversation_id, + agent=Agent(llm=LLM(model="gpt-4o", usage_id="test-llm"), tools=[]), + workspace=mock_service.stored.workspace, + execution_status=ConversationExecutionStatus.IDLE, + confirmation_policy=mock_service.stored.confirmation_policy, + ) + conversation_service._event_services[conversation_id] = mock_service + record = _ConversationRecord( + stored=mock_service.stored, + execution_status=ConversationExecutionStatus.IDLE, + ) + + info = await conversation_service._conversation_info( + conversation_id, record, {} + ) + + assert info is not None + assert info.id == conversation_id + assert len(events) == 1 + (event,) = events + assert event.operation == "conversation_info_compose" + assert event.stuck is False + + @pytest.mark.asyncio + async def test_conversation_info_compose_slow_state_read_emits_stuck( + self, conversation_service, monkeypatch + ): + """A composition wedged on ``get_state`` surfaces as stuck.""" + events = [] + monkeypatch.setattr( + "openhands.agent_server.conversation_service.timed_operation", + self._recording_timed_operation(events, default_budget_ms=100), + ) + conversation_id = uuid4() + mock_service = AsyncMock(spec=EventService) + mock_service.conversation_dir = "/tmp/test_conversation" + mock_service.stored = StoredConversation( + id=conversation_id, + workspace=LocalWorkspace(working_dir="/tmp/test_workspace"), + confirmation_policy=NeverConfirm(), + ) + + async def slow_get_state(): + await asyncio.sleep(0.25) + return ConversationState( + id=conversation_id, + agent=Agent(llm=LLM(model="gpt-4o", usage_id="test-llm"), tools=[]), + workspace=mock_service.stored.workspace, + execution_status=ConversationExecutionStatus.IDLE, + confirmation_policy=mock_service.stored.confirmation_policy, + ) + + mock_service.is_open.return_value = True + mock_service.get_state.side_effect = slow_get_state + conversation_service._event_services[conversation_id] = mock_service + record = _ConversationRecord( + stored=mock_service.stored, + execution_status=ConversationExecutionStatus.IDLE, + ) + + started = asyncio.get_running_loop().time() + info = await conversation_service._conversation_info( + conversation_id, record, {} + ) + + assert info is not None + assert info.id == conversation_id + assert len(events) == 2, "stuck signal then completion" + signal, completion = events + assert signal.operation == "conversation_info_compose" + assert signal.stuck is True + assert completion.operation == "conversation_info_compose" + assert completion.stuck is True + assert completion.duration_ms >= 100 + assert asyncio.get_running_loop().time() - started >= 0.2 + + @pytest.mark.asyncio + async def test_evict_idle_emits_timing_with_bucketed_count( + self, conversation_service, monkeypatch + ): + """An eviction pass emits ``conversation_evict`` with a bucketed count. + + Raw count is never exported: the magnitude is reported through the + coarse ``COUNT_BOUNDS`` vocabulary. + """ + events = [] + monkeypatch.setattr( + "openhands.agent_server.conversation_service.timed_operation", + self._recording_timed_operation(events), + ) + idle_cid = uuid4() + running_cid = uuid4() + for cid in (idle_cid, running_cid): + mock_service = AsyncMock(spec=EventService) + mock_service.conversation_dir = "/tmp/test_conversation" + mock_service.credential_bindings = {} + mock_service.stored = StoredConversation( + id=cid, + workspace=LocalWorkspace(working_dir="/tmp/test_workspace"), + confirmation_policy=NeverConfirm(), + ) + conversation_service._event_services[cid] = mock_service + idle = conversation_service._event_services[idle_cid] + running = conversation_service._event_services[running_cid] + idle.is_open.return_value = True + idle.idle_seconds.return_value = 999.0 + idle.is_idle_evictable.return_value = True + running.is_open.return_value = True + running.idle_seconds.return_value = 999.0 + running.is_idle_evictable.return_value = False + + await conversation_service._evict_idle_conversations(ttl_seconds=60.0) + + # Only the idle one was closed. + idle.__aexit__.assert_awaited_once() + running.__aexit__.assert_not_awaited() + assert len(events) == 1 + (event,) = events + assert event.operation == "conversation_evict" + assert event.stuck is False + # One eviction buckets as 1-5 (COUNT_BOUNDS first edge at 1). + assert event.evicted_count == "1-5" + + class TestSafeRmtree: """Test cases for the _safe_rmtree helper function."""