feat(eval): pipeline diagnostic & retry hardening#84
Open
yilu331 wants to merge 6 commits into
Open
Conversation
📝 WalkthroughWalkthroughThis PR adds health instrumentation for the agent-success evaluation pipeline: a thread-safe EvalHealth singleton records skip reasons, producer failures (24h rolling), and scheduler ticks; a new ChangesAgent Success Evaluation Health Tracking
Sequence DiagramsequenceDiagram
participant Scheduler as Scheduler Loop
participant Runner as Evaluation Runner
participant Service as Result Service
participant Health as EvalHealth Singleton
participant API as /healthz/eval Endpoint
Scheduler->>Health: record_tick(monotonic_ts)
Note over Health: last_tick_monotonic updated
Runner->>Runner: evaluate session
alt session already evaluated
Runner->>Health: record_skip(ALREADY_EVALUATED)
else no requests
Runner->>Health: record_skip(NO_REQUESTS)
end
Service->>Service: persist results
alt save succeeds
Service->>Health: (no producer failure)
else save fails after retries
Service->>Health: record_producer_failure()
end
API->>Health: get_status()
Health-->>API: status with skip_counts, failures, last_tick
API->>API: compute liveness from tick age
API-->>Client: return {status + liveness}
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
reflexio/server/services/agent_success_evaluation/delayed_group_evaluator.py (1)
94-111:⚠️ Potential issue | 🟠 Major | ⚡ Quick winBound wait duration so liveness ticks don’t go stale while scheduler is idle.
The loop records a tick only per iteration, but Line 100 can block indefinitely and Line 109 can block for very long delays. That causes false amber/red liveness in
/healthz/evaleven when the scheduler thread is healthy.💡 Proposed fix (heartbeat-bounded waits)
GROUP_EVALUATION_DELAY_SECONDS = 600 # 10 minutes +_TICK_HEARTBEAT_SECONDS = 30 @@ if next_fire_time is None: # Nothing scheduled, wait for a wake signal - self._wake_event.wait() + self._wake_event.wait(timeout=_TICK_HEARTBEAT_SECONDS) self._wake_event.clear() continue @@ if wait_seconds > 0: # Wait until the next fire time or a wake signal - self._wake_event.wait(timeout=wait_seconds) + self._wake_event.wait( + timeout=min(wait_seconds, _TICK_HEARTBEAT_SECONDS) + ) self._wake_event.clear() continue🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@reflexio/server/services/agent_success_evaluation/delayed_group_evaluator.py` around lines 94 - 111, The scheduler's waits can block indefinitely and prevent _eval_health.record_tick() from running; update the wait calls that currently call self._wake_event.wait() (the branch when next_fire_time is None and the branch when wait_seconds > 0) to use a bounded timeout (e.g., HEARTBEAT_SEC) so the thread wakes periodically to call _eval_health.record_tick(). Concretely, introduce a heartbeat constant and replace the unconditional wait() with wait(timeout=HEARTBEAT_SEC) in the "no next_fire_time" branch, and replace wait(timeout=wait_seconds) with wait(timeout=min(wait_seconds, HEARTBEAT_SEC)) in the scheduled-wait branch, keeping the existing self._wake_event.clear() logic and using the same mutex/_heap access points.
🧹 Nitpick comments (2)
tests/server/services/agent_success_evaluation/test_save_retry.py (1)
25-25: ⚡ Quick winUse a dedicated reset method instead of re-invoking
__init__.Calling
_HEALTH.__init__()directly to reset singleton state is fragile—ifEvalHealth.__init__later requires arguments or changes signature, this breaks silently. Consider adding areset()method toEvalHealth(or a test fixture) that explicitly clears state.Suggested approach in `_eval_health.py`
# In EvalHealth class def reset(self) -> None: """Reset all counters (test-only).""" with self._lock: self._skip_counts = {r: 0 for r in SkipReason} self._producer_failures: list[datetime] = [] self._last_tick: datetime | None = NoneThen in tests:
- _eval_health._HEALTH.__init__() + _eval_health._HEALTH.reset()🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/server/services/agent_success_evaluation/test_save_retry.py` at line 25, The test resets singleton state by calling _HEALTH.__init__(), which is fragile; add a dedicated reset method on the EvalHealth class (e.g., EvalHealth.reset(self)) that acquires the existing _lock and clears internal state (_skip_counts reset over SkipReason, _producer_failures emptied, _last_tick set to None), then change the test to call _HEALTH.reset() instead of _HEALTH.__init__(); reference EvalHealth, _HEALTH, reset, _lock, _skip_counts, SkipReason, _producer_failures, and _last_tick when making the change.reflexio/server/services/agent_success_evaluation/agent_success_evaluation_service.py (1)
159-159: 💤 Low valueConsider narrowing the exception type.
The bare
Exceptioncatch is intentional for retry resilience, but if the storage layer throws specific exceptions (e.g.,StorageError,IOError), catching those explicitly would prevent accidentally swallowing unrelated errors likeKeyboardInterruptsubclasses or programming bugs.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@reflexio/server/services/agent_success_evaluation/agent_success_evaluation_service.py` at line 159, The bare "except Exception as e" in the agent success evaluation storage/retry block should be narrowed: catch the specific storage-related exceptions (e.g., StorageError, IOError, OSError) raised by the storage layer instead of Exception, or re-raise system/interrupt exceptions (KeyboardInterrupt, SystemExit) if caught; update the except clause in the try/except around the storage call in AgentSuccessEvaluationService (the block that currently reads "except Exception as e") to list the storage exceptions explicitly and/or add an "except Exception as e: if isinstance(e, (KeyboardInterrupt, SystemExit)): raise; else: handle/retry" guard so unrelated errors are not swallowed.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In
`@reflexio/server/services/agent_success_evaluation/delayed_group_evaluator.py`:
- Around line 94-111: The scheduler's waits can block indefinitely and prevent
_eval_health.record_tick() from running; update the wait calls that currently
call self._wake_event.wait() (the branch when next_fire_time is None and the
branch when wait_seconds > 0) to use a bounded timeout (e.g., HEARTBEAT_SEC) so
the thread wakes periodically to call _eval_health.record_tick(). Concretely,
introduce a heartbeat constant and replace the unconditional wait() with
wait(timeout=HEARTBEAT_SEC) in the "no next_fire_time" branch, and replace
wait(timeout=wait_seconds) with wait(timeout=min(wait_seconds, HEARTBEAT_SEC))
in the scheduled-wait branch, keeping the existing self._wake_event.clear()
logic and using the same mutex/_heap access points.
---
Nitpick comments:
In
`@reflexio/server/services/agent_success_evaluation/agent_success_evaluation_service.py`:
- Line 159: The bare "except Exception as e" in the agent success evaluation
storage/retry block should be narrowed: catch the specific storage-related
exceptions (e.g., StorageError, IOError, OSError) raised by the storage layer
instead of Exception, or re-raise system/interrupt exceptions
(KeyboardInterrupt, SystemExit) if caught; update the except clause in the
try/except around the storage call in AgentSuccessEvaluationService (the block
that currently reads "except Exception as e") to list the storage exceptions
explicitly and/or add an "except Exception as e: if isinstance(e,
(KeyboardInterrupt, SystemExit)): raise; else: handle/retry" guard so unrelated
errors are not swallowed.
In `@tests/server/services/agent_success_evaluation/test_save_retry.py`:
- Line 25: The test resets singleton state by calling _HEALTH.__init__(), which
is fragile; add a dedicated reset method on the EvalHealth class (e.g.,
EvalHealth.reset(self)) that acquires the existing _lock and clears internal
state (_skip_counts reset over SkipReason, _producer_failures emptied,
_last_tick set to None), then change the test to call _HEALTH.reset() instead of
_HEALTH.__init__(); reference EvalHealth, _HEALTH, reset, _lock, _skip_counts,
SkipReason, _producer_failures, and _last_tick when making the change.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 319ef730-e4f2-41da-92c5-467129dac742
📒 Files selected for processing (11)
reflexio/server/api_endpoints/health_api.pyreflexio/server/services/agent_success_evaluation/_eval_health.pyreflexio/server/services/agent_success_evaluation/agent_success_evaluation_service.pyreflexio/server/services/agent_success_evaluation/delayed_group_evaluator.pyreflexio/server/services/agent_success_evaluation/group_evaluation_runner.pytests/server/api_endpoints/test_health_api_eval.pytests/server/services/agent_success_evaluation/__init__.pytests/server/services/agent_success_evaluation/test_eligibility_logging.pytests/server/services/agent_success_evaluation/test_eval_health.pytests/server/services/agent_success_evaluation/test_save_retry.pytests/server/services/agent_success_evaluation/test_scheduler_tick.py
The agentic extraction backend hit `return result` before reaching the scheduler invocation in the classic path, so no AgentSuccessEvaluationResult records were ever produced for any org running with extraction_backend=agentic. This was the silent root cause of empty /evaluations tiles observed by operators. Extracts the scheduler invocation into a `_schedule_group_evaluation_if_needed` helper and calls it from BOTH the agentic and classic code paths. Regression test verifies the helper schedules with key=(org_id, user_id, session_id) when the session_id is set, and short-circuits cleanly when it is not. Verified live: published 10 sessions against a backend with the fix, observed "Scheduled group evaluation for key=..." followed by "Firing group evaluation" and "Running group evaluation for session=..." in the logs. /healthz/eval transitioned from liveness=red to green with last_tick_monotonic populated.
There was a problem hiding this comment.
🧹 Nitpick comments (1)
tests/server/services/test_generation_service_scheduling.py (1)
70-75: ⚡ Quick winAssert callback behavior, not just callability.
The current assertion allows regressions where the callback is callable but wired with wrong captured args. Execute the callback and assert
run_group_evaluationreceives the expected values.Proposed test hardening
def test_schedules_with_correct_key_when_session_id_present( service: GenerationService, monkeypatch: pytest.MonkeyPatch ) -> None: @@ scheduler.schedule.assert_called_once() call_args = scheduler.schedule.call_args key = call_args[0][0] if call_args[0] else call_args.kwargs.get("key") callback = call_args[0][1] if len(call_args[0]) > 1 else call_args.kwargs.get("callback") assert key == ("org_test", "user_test", "sess_42") assert callable(callback) + + captured: dict[str, object] = {} + + def fake_run_group_evaluation( + *, + org_id: str, + user_id: str, + session_id: str, + agent_version: str, + source: str | None, + request_context: object, + llm_client: object, + ) -> None: + captured.update( + { + "org_id": org_id, + "user_id": user_id, + "session_id": session_id, + "agent_version": agent_version, + "source": source, + "request_context": request_context, + "llm_client": llm_client, + } + ) + + monkeypatch.setattr( + "reflexio.server.services.generation_service.run_group_evaluation", + fake_run_group_evaluation, + ) + + callback() + assert captured == { + "org_id": "org_test", + "user_id": "user_test", + "session_id": "sess_42", + "agent_version": "v_test", + "source": "ide", + "request_context": service.request_context, + "llm_client": service.client, + }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/server/services/test_generation_service_scheduling.py` around lines 70 - 75, The test currently only asserts the scheduled callback is callable; change it to execute the captured callback and assert it triggers run_group_evaluation with the expected arguments: after obtaining call_args, key and callback (from scheduler.schedule.call_args), call callback() (or await it if it’s async) and then assert run_group_evaluation.assert_called_once_with("org_test", "user_test", "sess_42", any_other_expected_params) (or use assert_called_once() plus assert call args match the tuple), ensuring you reference scheduler.schedule, call_args, key, callback and run_group_evaluation when adding the execution and verification.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@tests/server/services/test_generation_service_scheduling.py`:
- Around line 70-75: The test currently only asserts the scheduled callback is
callable; change it to execute the captured callback and assert it triggers
run_group_evaluation with the expected arguments: after obtaining call_args, key
and callback (from scheduler.schedule.call_args), call callback() (or await it
if it’s async) and then assert
run_group_evaluation.assert_called_once_with("org_test", "user_test", "sess_42",
any_other_expected_params) (or use assert_called_once() plus assert call args
match the tuple), ensuring you reference scheduler.schedule, call_args, key,
callback and run_group_evaluation when adding the execution and verification.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 527d0444-914d-4502-a8d4-a7ae408499de
📒 Files selected for processing (2)
reflexio/server/services/generation_service.pytests/server/services/test_generation_service_scheduling.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Plan A of the /evaluations redesign work. Makes the existing evaluation pipeline observable and self-healing — AND fixes a silent product bug discovered during live e2e verification.
Summary
save_agent_success_evaluation_resultswith backoff (1s / 4s, 3 total attempts); records final failures into EvalHealth instead of silently dropping them.run_group_evaluationbump matchingSkipReasoncounters so the operator can see why sessions are dropping./healthz/evalendpoint surfaces all three signals with a green/amber/red liveness derivation (>5min silent = amber, >30min = red).return resultbefore reaching the scheduler invocation in the classic path, so no AgentSuccessEvaluationResult records were ever produced for any org running withextraction_backend=agentic. This was the silent root cause of empty /evaluations tiles. Extracted into a shared_schedule_group_evaluation_if_neededhelper called from both code paths.Live verification
Discovered and fixed during local e2e testing — published 10 substantive sessions to a backend running this branch, observed:
last_tick_monotonic: null,liveness: red, no scheduler logs.Scheduled group evaluation for key=...→Firing group evaluation→Running group evaluation for session=...in logs;/healthz/evalreportsliveness: greenwith populated tick.Commits
EvalHealth singleton for pipeline diagnosticsscheduler records tick into EvalHealthbump SkipReason counters in each skip pathretry save with backoff and record producer failures/healthz/eval exposes pipeline diagnosticsschedule group evaluation from the agentic path too(discovered via e2e)Test plan
run_group_evaluation/healthz/evalroute tests for green/amber/red liveness derivation_schedule_group_evaluation_if_neededCompanion PR
ReflexioAI/reflexio-enterprise#100 bumps the submodule pointer here.
Operational follow-up (out of scope for Plan A)
During live e2e, the runner logged
No agent_success_evaluation extractor configs found— the deployed config hasagent_success_configs: null, so even when the scheduler fires correctly, the evaluator produces no records. This is a config gap, not a code bug, and is separate from Plan A.