From 28834a2098758808c804ca07e2837302e01c3fb3 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Tue, 1 Sep 2026 09:24:54 -0700 Subject: [PATCH] test: raise tight wall-clock bounds that flaked on loaded CI runners Seven test files asserted sub-2s wall-clock bounds (elapsed < 0.5/1.0s, stop(timeout=1.0), event waits of 0.5-2s). Under CI load these fired on healthy code: main run 33455779041 alone flaked 6 of them in one pass (observed 1.01s vs 0.5, 1.20s vs 1.0, 3.61s vs 3.0, 1.55s vs 1.0, stop(1.0) returning False, lease TTL 0.1s expiring before the authority change was observed). Per the AGENTS.md flake policy (waits >= 2s), bounds are raised to 5s+ while keeping their teeth: every hang path they guard blocks for 10s+ (release.wait holds), so the loosened bounds still distinguish bounded from unbounded behavior. The authority-loss test gets a 30s lease TTL so lease expiry can no longer preempt the authority-change assertion. --- tests/cron/test_cleanup_timeout.py | 8 +-- .../gateway/test_73771_media_resend_dedup.py | 24 ++++---- .../test_hosted_room_gateway_lifecycle.py | 6 +- tests/gateway/test_pending_drain_race.py | 10 ++-- .../test_kanban_init_lock_bounded.py | 4 +- tests/hermes_cli/test_plugins.py | 6 +- .../test_hosted_room_driver_runtime.py | 58 ++++++++++--------- 7 files changed, 61 insertions(+), 55 deletions(-) diff --git a/tests/cron/test_cleanup_timeout.py b/tests/cron/test_cleanup_timeout.py index 6d17502e34..6b967afc2a 100644 --- a/tests/cron/test_cleanup_timeout.py +++ b/tests/cron/test_cleanup_timeout.py @@ -70,8 +70,8 @@ def test_run_job_bounds_sessiondb_finalization(tmp_path): success, _output, final_response, error = run_job(job) elapsed = time.monotonic() - started - assert fake_db.entered.wait(timeout=0.5) - assert elapsed < 0.5 + assert fake_db.entered.wait(timeout=2.0) + assert elapsed < 5.0 assert success is True assert final_response == "ok" assert error is None @@ -88,8 +88,8 @@ def test_agent_teardown_is_bounded(): _teardown_cron_agent(agent, "cleanup-agent-hang", timeout_seconds=0.02) elapsed = time.monotonic() - started - assert agent.entered.wait(timeout=0.5) - assert elapsed < 0.5 + assert agent.entered.wait(timeout=2.0) + assert elapsed < 5.0 finally: release.set() diff --git a/tests/gateway/test_73771_media_resend_dedup.py b/tests/gateway/test_73771_media_resend_dedup.py index d0490e8056..3c54e1c055 100644 --- a/tests/gateway/test_73771_media_resend_dedup.py +++ b/tests/gateway/test_73771_media_resend_dedup.py @@ -353,18 +353,19 @@ async def test_bare_path_history_lookup_timeout_fails_open(tmp_path, monkeypatch started = time.monotonic() await adapter._process_message_background(event, build_session_key(event.source)) - # The lookup times out after 0.02s and fails open; the generous 1.0s - # bound only guards against delivery hanging on the wedged read - # indefinitely, without flaking on loaded CI hosts. Delivery of the - # document below is the real fail-open assertion. - assert time.monotonic() - started < 1.0 + # The lookup times out after 0.02s and fails open; the bound only guards + # against delivery hanging on the wedged read indefinitely. 1.0s still + # flaked on loaded CI runners (observed 1.55s on main run 33455779041), + # so keep it >= 5s per the flake policy. Delivery of the document below + # is the real fail-open assertion. + assert time.monotonic() - started < 5.0 assert adapter.documents == [str(pdf)] @pytest.mark.asyncio async def test_history_lookup_saturation_fails_open_without_new_worker(monkeypatch): """Wedged lookups are bounded and cannot consume unbounded worker threads.""" - monkeypatch.setattr("gateway.platforms.base._HISTORY_MEDIA_LOOKUP_TIMEOUT_SECONDS", 1.0) + monkeypatch.setattr("gateway.platforms.base._HISTORY_MEDIA_LOOKUP_TIMEOUT_SECONDS", 5.0) monkeypatch.setattr( "gateway.platforms.base._HISTORY_MEDIA_LOOKUP_ADMISSION", threading.BoundedSemaphore(2), @@ -381,13 +382,13 @@ async def test_history_lookup_saturation_fails_open_without_new_worker(monkeypat calls += 1 if calls == 2: two_started.set() - release.wait(timeout=1) + release.wait(timeout=10) return None monkeypatch.setattr(adapter, "_history_media_paths_for_session", blocked_lookup) first = asyncio.create_task(adapter._bounded_history_media_paths_for_session("one")) second = asyncio.create_task(adapter._bounded_history_media_paths_for_session("two")) - deadline = time.monotonic() + 1 + deadline = time.monotonic() + 5 while not two_started.is_set() and time.monotonic() < deadline: await asyncio.sleep(0.005) assert two_started.is_set() @@ -397,9 +398,10 @@ async def test_history_lookup_saturation_fails_open_without_new_worker(monkeypat elapsed = time.monotonic() - began assert third is None - # Saturation must fail open immediately (no waiting on the 1.0s lookup - # timeout); 0.5s is a generous bound that stays flake-free on loaded CI. - assert elapsed < 0.5 + # Saturation must fail open immediately (no waiting on the 5.0s lookup + # timeout); 2.0s keeps the distinction while staying flake-free on + # loaded CI runners. + assert elapsed < 2.0 assert calls == 2 release.set() await asyncio.gather(first, second) diff --git a/tests/gateway/test_hosted_room_gateway_lifecycle.py b/tests/gateway/test_hosted_room_gateway_lifecycle.py index fe692d41cc..679d7ac145 100644 --- a/tests/gateway/test_hosted_room_gateway_lifecycle.py +++ b/tests/gateway/test_hosted_room_gateway_lifecycle.py @@ -185,7 +185,7 @@ def test_gateway_restart_resumes_queued_room_for_multiplexed_profile(tmp_path): ) ) finally: - assert resumed.stop(timeout=1.0) + assert resumed.stop(timeout=5.0) assert rpc.submits == ["ops"] assert hosted_room_driver.list_tasks(db, room_id="room-1", status="settled") @@ -226,8 +226,8 @@ def test_dashboard_and_gateway_workers_share_one_fenced_execution_owner(tmp_path ) time.sleep(0.05) finally: - assert gateway.stop(timeout=1.0) - assert dashboard.stop(timeout=1.0) + assert gateway.stop(timeout=5.0) + assert dashboard.stop(timeout=5.0) assert len(gateway_rpc.submits) + len(dashboard_rpc.submits) == 1 events = hosted_rooms.read_events(db, room_id="room-1", since_seq=0)["events"] diff --git a/tests/gateway/test_pending_drain_race.py b/tests/gateway/test_pending_drain_race.py index 10ca90dc75..479e769264 100644 --- a/tests/gateway/test_pending_drain_race.py +++ b/tests/gateway/test_pending_drain_race.py @@ -109,7 +109,7 @@ async def test_pending_drain_keeps_active_session_guard_live(): await adapter.handle_message(_make_event(text="M1")) # Wait until M1 is actively running inside the handler. - await asyncio.wait_for(first_started.wait(), timeout=1.0) + await asyncio.wait_for(first_started.wait(), timeout=5.0) # Assert: session is active. assert sk in adapter._active_sessions @@ -126,7 +126,7 @@ async def test_pending_drain_keeps_active_session_guard_live(): try: # Pause inside the handoff's typing cleanup. Production has already # cleared the guard and has not yet transferred task ownership. - await asyncio.wait_for(handoff_entered.wait(), timeout=2.0) + await asyncio.wait_for(handoff_entered.wait(), timeout=5.0) # Across the drain transition, the Event object must be the SAME # reference (not replaced, not deleted). @@ -141,7 +141,7 @@ async def test_pending_drain_keeps_active_session_guard_live(): # Finish drain without relying on scheduler speed. release_handoff.set() - await asyncio.wait_for(second_processed.wait(), timeout=2.0) + await asyncio.wait_for(second_processed.wait(), timeout=5.0) finally: release_handoff.set() await adapter.cancel_background_tasks() @@ -190,7 +190,7 @@ async def test_finally_cleanup_drains_late_arrival_pending(): await adapter.handle_message(_make_event(text="M1")) # Drain: wait for the late-drain task itself to process LATE. - await asyncio.wait_for(late_processed.wait(), timeout=2.0) + await asyncio.wait_for(late_processed.wait(), timeout=5.0) await adapter.cancel_background_tasks() @@ -218,7 +218,7 @@ async def test_no_pending_cleans_up_normally(): # Await the task that owns this session rather than sampling cleanup after # an arbitrary wall-clock delay. owner_task = adapter._session_tasks[sk] - await asyncio.wait_for(asyncio.shield(owner_task), timeout=2.0) + await asyncio.wait_for(asyncio.shield(owner_task), timeout=5.0) assert sk not in adapter._active_sessions, ( "_active_sessions was not cleaned up after a normal turn with no pending" diff --git a/tests/hermes_cli/test_kanban_init_lock_bounded.py b/tests/hermes_cli/test_kanban_init_lock_bounded.py index d7730712c6..38c5782713 100644 --- a/tests/hermes_cli/test_kanban_init_lock_bounded.py +++ b/tests/hermes_cli/test_kanban_init_lock_bounded.py @@ -66,7 +66,7 @@ def test_initialized_path_connect_skips_init_lock(kanban_home): start = time.monotonic() kb.connect().close() elapsed = time.monotonic() - start - assert elapsed < 1.0, f"fast-path connect blocked on the init lock ({elapsed:.2f}s)" + assert elapsed < 5.0, f"fast-path connect blocked on the init lock ({elapsed:.2f}s)" finally: release.set() t.join(timeout=5) @@ -85,7 +85,7 @@ def test_first_init_connect_is_bounded_when_lock_held(kanban_home, monkeypatch): conn.close() elapsed = time.monotonic() - start # Proceeded within roughly the timeout window (not unbounded). - assert 0.4 <= elapsed < 3.0, f"expected bounded ~0.6s acquire, got {elapsed:.2f}s" + assert 0.4 <= elapsed < 8.0, f"expected bounded ~0.6s acquire, got {elapsed:.2f}s" assert str(db_path.resolve()) in kb._INITIALIZED_PATHS finally: release.set() diff --git a/tests/hermes_cli/test_plugins.py b/tests/hermes_cli/test_plugins.py index 4f4f67be0c..5ae4d5aee1 100644 --- a/tests/hermes_cli/test_plugins.py +++ b/tests/hermes_cli/test_plugins.py @@ -1048,7 +1048,7 @@ class TestForceReloadSymmetry: assert started.wait(timeout=1.0) assert results == [{"ok": True}] - assert elapsed < 1.0, f"caller blocked for {elapsed:.2f}s after timeout" + assert elapsed < 5.0, f"caller blocked for {elapsed:.2f}s after timeout" hold.set() def test_hook_callback_within_timeout_returns_value(self, monkeypatch): @@ -1132,7 +1132,7 @@ class TestForceReloadSymmetry: elapsed = time.monotonic() - t0 assert len(starts) == 1 - assert elapsed < 1.0 + assert elapsed < 5.0 hold.set() def test_pre_tool_call_timeout_fail_closed(self, monkeypatch): @@ -1166,7 +1166,7 @@ class TestForceReloadSymmetry: elapsed = time.monotonic() - t0 assert msg == _PRE_TOOL_CALL_TIMEOUT_BLOCK_MESSAGE - assert elapsed < 1.0 + assert elapsed < 5.0 # Still-running / suppression window must also fail closed. msg2 = resolve_pre_tool_block("web_search", {"query": "y"}) diff --git a/tests/tui_gateway/test_hosted_room_driver_runtime.py b/tests/tui_gateway/test_hosted_room_driver_runtime.py index ef1cbbc9fe..9b22670151 100644 --- a/tests/tui_gateway/test_hosted_room_driver_runtime.py +++ b/tests/tui_gateway/test_hosted_room_driver_runtime.py @@ -492,7 +492,7 @@ def test_waiting_room_does_not_block_an_independent_local_room(tmp_path: Path): assert state.get_task(db, identities[0])["status"] == "running" _wait_for(lambda: len(runtime.status()["current_tasks"]) == 1) assert len(runtime.status()["current_tasks"]) == 1 - assert runtime.stop(timeout=1.0) + assert runtime.stop(timeout=5.0) def test_rotated_bounded_scheduler_eventually_runs_later_room(tmp_path: Path): @@ -537,7 +537,7 @@ def test_rotated_bounded_scheduler_eventually_runs_later_room(tmp_path: Path): runtime.start() _wait_for(lambda: state.get_task(db, identity)["status"] == "settled") - assert runtime.stop(timeout=1.0) + assert runtime.stop(timeout=5.0) def test_queued_task_routes_profile_and_credentials_without_overrides(db: Path): @@ -548,7 +548,7 @@ def test_queued_task_routes_profile_and_credentials_without_overrides(db: Path): runtime.start() _wait_for(lambda: state.get_task(db, identity)["status"] == "settled") - assert runtime.stop(timeout=1.0) + assert runtime.stop(timeout=5.0) create = next(params for method, params in rpc.calls if method == "create") submit = next(params for method, params in rpc.calls if method == "submit") @@ -580,7 +580,7 @@ def test_worker_settles_without_any_client_transport(db: Path): assert runtime.status()["running"] is True assert runtime.status()["cycles"] >= 1 - assert runtime.stop(timeout=1.0) + assert runtime.stop(timeout=5.0) def test_policy_hooks_prepare_and_publish_terminal_idempotently(db: Path): @@ -601,7 +601,7 @@ def test_policy_hooks_prepare_and_publish_terminal_idempotently(db: Path): runtime.start() _wait_for(lambda: state.get_task(db, identity)["status"] == "settled") - assert runtime.stop(timeout=1.0) + assert runtime.stop(timeout=5.0) assert prepared assert published == [(ROOM_ID, identity.task_id, "settled")] @@ -630,7 +630,7 @@ def test_transport_resolver_selects_member_transport_without_forking_state( runtime.start() _wait_for(lambda: state.get_task(db, identity)["status"] == "settled") - assert runtime.stop(timeout=1.0) + assert runtime.stop(timeout=5.0) assert resolutions assert all(binding == BINDING for binding, _, _ in resolutions) @@ -766,7 +766,7 @@ def test_waiting_room_does_not_block_an_independent_room(tmp_path: Path): assert waiting.submitted.wait(1.0) _wait_for(lambda: state.get_task(db, identities[1])["status"] == "settled") assert state.get_task(db, identities[0])["status"] == "running" - assert runtime.stop(timeout=1.0) + assert runtime.stop(timeout=5.0) def test_bounded_scheduler_eventually_runs_later_room(tmp_path: Path): @@ -812,7 +812,7 @@ def test_bounded_scheduler_eventually_runs_later_room(tmp_path: Path): runtime.start() _wait_for(lambda: state.get_task(db, identity)["status"] == "settled") - assert runtime.stop(timeout=1.0) + assert runtime.stop(timeout=5.0) def test_existing_canonical_session_is_resumed_not_duplicated(db: Path): @@ -824,7 +824,7 @@ def test_existing_canonical_session_is_resumed_not_duplicated(db: Path): runtime.start() _wait_for(lambda: state.get_task(db, identity)["status"] == "settled") - assert runtime.stop(timeout=1.0) + assert runtime.stop(timeout=5.0) assert not [call for call in rpc.calls if call[0] == "create"] resume = next(params for method, params in rpc.calls if method == "resume") @@ -962,13 +962,13 @@ def test_oversized_terminal_reply_is_bounded_without_waiting_for_deadline(db: Pa runtime = _runtime(db, rpc, turn_timeout_seconds=30) runtime.start() - assert rpc.submitted.wait(timeout=1.0) + assert rpc.submitted.wait(timeout=5.0) rpc.complete( identity.task_id, content="é" * (MAX_TERMINAL_TEXT_BYTES + 100), ) _wait_for(lambda: state.get_task(db, identity)["status"] == "settled") - assert runtime.stop(timeout=1.0) + assert runtime.stop(timeout=5.0) result = state.get_task(db, identity)["result"] assert result["truncated"] is True @@ -1052,7 +1052,7 @@ def test_turn_deadline_stops_exact_attempt_and_publishes_durable_failure(db: Pat runtime.start() _wait_for(lambda: state.get_task(db, identity)["status"] == "failed") - assert runtime.stop(timeout=1.0) + assert runtime.stop(timeout=5.0) failed = state.get_task(db, identity) assert failed["result"] == { @@ -1119,7 +1119,7 @@ def test_deadline_releases_worker_capacity_for_later_room(tmp_path: Path): runtime.start() _wait_for(lambda: state.get_task(db, identities[0])["status"] == "failed") _wait_for(lambda: state.get_task(db, identities[1])["status"] == "settled") - assert runtime.stop(timeout=1.0) + assert runtime.stop(timeout=5.0) assert state.get_task(db, identities[0])["result"]["reason_code"] == ( "turn_deadline_exceeded" @@ -1190,7 +1190,7 @@ def test_retry_ignores_late_receipt_from_prior_execution_generation(db: Path): runtime.start() assert rpc.submitted.wait(1.0) time.sleep(0.04) - assert runtime.stop(timeout=1.0) + assert runtime.stop(timeout=5.0) task = state.get_task(db, identity) assert task["status"] == "running" @@ -1222,7 +1222,7 @@ def test_active_recovered_turn_is_never_resubmitted(db: Path): runtime.start() time.sleep(0.08) - assert runtime.stop(timeout=1.0) + assert runtime.stop(timeout=5.0) assert state.get_task(db, identity)["status"] == "running" assert not [call for call in rpc.calls if call[0] == "submit"] @@ -1435,7 +1435,7 @@ def test_ambiguous_recovery_remains_indeterminate(db: Path): runtime.start() _wait_for(lambda: state.get_task(db, identity)["status"] == "indeterminate") - assert runtime.stop(timeout=1.0) + assert runtime.stop(timeout=5.0) assert not [call for call in rpc.calls if call[0] == "submit"] @@ -1570,7 +1570,7 @@ def test_post_submit_observation_failure_preserves_recoverable_outcome(db: Path) rpc.complete(identity.task_id, content="Recovered after a transient read.") runtime.wakeup() _wait_for(lambda: state.get_task(db, identity)["status"] == "settled") - assert runtime.stop(timeout=1.0) + assert runtime.stop(timeout=5.0) task = state.get_task(db, identity) assert task["result"]["text"] == "Recovered after a transient read." @@ -1595,7 +1595,7 @@ def test_cancellation_is_persisted_before_interrupt_and_fences_late_result( rpc.complete(identity.task_id, content="Too late.") runtime.wakeup() time.sleep(0.05) - assert runtime.stop(timeout=1.0) + assert runtime.stop(timeout=5.0) assert cancelled["status"] == "cancelled" assert observed_status == ["stopping"] @@ -1628,7 +1628,7 @@ def test_transient_remote_stop_failure_stays_pending_and_retries(db: Path): runtime.wakeup() _wait_for(lambda: state.get_task(db, identity)["status"] == "cancelled") assert attempts >= 2 - assert runtime.stop(timeout=1.0) + assert runtime.stop(timeout=5.0) assert state.get_task(db, identity)["status"] == "cancelled" @@ -1777,7 +1777,7 @@ def test_completion_wins_a_race_with_unacknowledged_stop(db: Path): assert result["status"] == "settled" assert result["result"]["text"] == "Already done." - assert runtime.stop(timeout=1.0) + assert runtime.stop(timeout=5.0) def test_restart_harvests_completion_before_retrying_durable_stop(db: Path): @@ -1836,7 +1836,7 @@ def test_restart_harvests_completion_before_retrying_durable_stop(db: Path): runtime.start() _wait_for(lambda: state.get_task(db, identity)["status"] == "settled") - assert runtime.stop(timeout=1.0) + assert runtime.stop(timeout=5.0) settled = state.get_task(db, identity) assert stopping["status"] == "stopping" @@ -1974,7 +1974,7 @@ def test_pending_local_approval_is_reported_with_safe_choices(db: Path): assert member == PROFILE assert action["request_id"] == "approval-1" assert action["approval"]["choices"] == ["once", "deny"] - assert runtime.stop(timeout=1.0) + assert runtime.stop(timeout=5.0) def test_cancel_never_interrupts_a_newer_task_in_the_same_session(db: Path): @@ -2001,7 +2001,7 @@ def test_cancel_never_interrupts_a_newer_task_in_the_same_session(db: Path): assert all(params["expected_task_id"] == identity.task_id for params in skipped) assert rpc.states[session_id]["active"] is True assert rpc.states[session_id]["task_id"] == "task-2" - assert runtime.stop(timeout=1.0) + assert runtime.stop(timeout=5.0) def test_status_reports_room_blocked_on_unresolved_indeterminate_task(db: Path): @@ -2031,7 +2031,7 @@ def test_status_reports_room_blocked_on_unresolved_indeterminate_task(db: Path): runtime.start() _wait_for(lambda: ROOM_ID in runtime.status()["blocked_rooms"]) - assert runtime.stop(timeout=1.0) + assert runtime.stop(timeout=5.0) assert state.get_task(db, identity)["status"] == "indeterminate" @@ -2040,7 +2040,11 @@ def test_authority_loss_stops_terminal_commit(db: Path): identity = _identity() _admit(db, identity) rpc = FakeSessionRPC(auto_complete=False) - runtime = _runtime(db, rpc, lease_ttl_seconds=0.1) + # Generous lease TTL: this test is about AUTHORITY loss. A short TTL let + # a loaded CI runner expire the lease before the authority change was + # observed, so last_error flipped to "driver lease is stale or expired" + # (flaky main run 33455779041). + runtime = _runtime(db, rpc, lease_ttl_seconds=30.0) runtime.start() assert rpc.submitted.wait(1.0) @@ -2056,7 +2060,7 @@ def test_authority_loss_stops_terminal_commit(db: Path): rpc.complete(identity.task_id) runtime.wakeup() _wait_for(lambda: runtime.status()["last_error"] is not None) - assert runtime.stop(timeout=1.0) + assert runtime.stop(timeout=5.0) assert state.get_task(db, identity)["status"] == "running" assert "authority changed" in runtime.status()["last_error"] @@ -2071,7 +2075,7 @@ def test_profile_turn_lock_covers_resolve_submit_and_terminal_observation(db: Pa runtime.start() _wait_for(lambda: state.get_task(db, identity)["status"] == "settled") - assert runtime.stop(timeout=1.0) + assert runtime.stop(timeout=5.0) assert locks.events == [("lock-enter", PROFILE), ("lock-exit", PROFILE)] methods = [method for method, _params in rpc.calls]