BZ-5359: feat: Migrate observers to evaluation table - #997
Conversation
|
Important Review skippedAuto incremental reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
WalkthroughThe PR adds observer evaluation configuration resolution, detection persistence, alert-first execution, generalized evaluation-result storage, and observer-based analytics with current and legacy detection support. ChangesObserver evaluation lifecycle
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant TemplateAPI
participant EvaluationConfig
participant RealtimeObserver
participant EvaluationResult
participant AnalyticsHandler
TemplateAPI->>EvaluationConfig: synchronize observer evaluators
RealtimeObserver->>EvaluationConfig: resolve observer configuration
RealtimeObserver->>EvaluationResult: persist detection metadata
AnalyticsHandler->>EvaluationResult: fetch current and legacy detections
EvaluationResult-->>AnalyticsHandler: return rows and eligible conversation count
Possibly related PRs
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 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.
Actionable comments posted: 6
🧹 Nitpick comments (3)
app/database/accessor/breeze_buddy/analytics/observer_result.py (1)
13-36: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd exception logging around the query calls.
The accessor contract in this repository requires logging the database exception and re-raising it. These three functions propagate exceptions without any log record, so a failure reaches the router with no accessor-level context. Wrap each call in
try/except Exception, log withlogger.opt(exception=e).error(...), and re-raise.Based on learnings: "If a database lookup raises an exception, you must (1) log the database exception and (2) re-raise it"; and use Loguru's
logger.opt(exception=e).error(...)rather thanexc_info=True.🤖 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 `@app/database/accessor/breeze_buddy/analytics/observer_result.py` around lines 13 - 36, Wrap the database query execution in each of get_observer_detection_rows_from_db, get_legacy_observer_detection_rows_from_db, and get_observer_eligible_conversation_count_from_db with try/except Exception. In each handler, log the exception using logger.opt(exception=e).error(...) with relevant accessor context, then re-raise the original exception unchanged.Source: Learnings
app/api/routers/breeze_buddy/analytics/handlers.py (1)
287-296: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winRun the three queries concurrently.
The three accessor calls are independent. Sequential awaits add their latencies. Use
asyncio.gatherto reduce endpoint response time.⚡ Proposed change
- rows = await get_observer_detection_rows_from_db(filters, limit=limit) - legacy_rows = await get_legacy_observer_detection_rows_from_db(filters, limit=limit) + rows, legacy_rows, eligible_conversations = await asyncio.gather( + get_observer_detection_rows_from_db(filters, limit=limit), + get_legacy_observer_detection_rows_from_db(filters, limit=limit), + get_observer_eligible_conversation_count_from_db(filters), + ) rows = sorted( [*rows, *legacy_rows], key=lambda row: str(row.get("started_at") or ""), reverse=True, )[:limit] - eligible_conversations = await get_observer_eligible_conversation_count_from_db( - filters - )🤖 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 `@app/api/routers/breeze_buddy/analytics/handlers.py` around lines 287 - 296, Update the handler flow around get_observer_detection_rows_from_db, get_legacy_observer_detection_rows_from_db, and get_observer_eligible_conversation_count_from_db to start all three independent calls concurrently with asyncio.gather, then unpack their results before sorting and counting. Preserve the existing filters, limit values, and output behavior.app/database/queries/breeze_buddy/analytics/observer_result.py (1)
48-67: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDate filtering duplicates the shared helper.
Lines 48-67 repeat the IST-to-UTC conversion and the exclusive upper bound already implemented in
build_analytics_where_clause. Extract that block into a shared helper so the date semantics stay identical across analytics queries.🤖 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 `@app/database/queries/breeze_buddy/analytics/observer_result.py` around lines 48 - 67, The date-filtering logic in the current query builder duplicates the shared behavior from build_analytics_where_clause. Extract the date_from/date_to conversion and exclusive date_to upper-bound handling into a reusable helper, then call it from both paths so IST-to-UTC conversion and boundary semantics remain identical.
🤖 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.
Inline comments:
In `@app/ai/voice/agents/breeze_buddy/observers/observer.py`:
- Around line 232-242: Stop persisting raw self._last_detection in the detection
payload built by the observer flow. Replace it with a bounded, explicitly
allowlisted detection representation before writing evaluation_result, excluding
transcript-derived PII, secrets, and arbitrary LLM tool arguments. Add tests
verifying sensitive and unrecognized values are omitted while permitted fields
remain.
In `@app/api/routers/breeze_buddy/analytics/handlers.py`:
- Around line 289-293: Update the merged-row sorting around rows and legacy_rows
to use each row’s started_at datetime directly rather than converting it to a
string. Provide an explicit minimum datetime fallback for missing values,
preserving descending order and the existing limit slicing so last_triggered
selects the most recent row.
- Around line 286-298: Update the analytics handler around
get_observer_detection_rows_from_db and the summary calculations so aggregate
metrics are computed over the full filtered dataset via SQL GROUP BY queries,
while the capped rows collection is used only for recent_fires. Ensure
total_triggers, trigger_rate, action/outcome breakdowns, and trend use uncapped
aggregates; if SQL aggregation cannot be added here, include a truncated flag in
summary whenever the row cap is reached.
In `@app/api/routers/breeze_buddy/templates/handlers.py`:
- Around line 70-85: Make the template persistence flow containing
upsert_observer_evaluation_config atomic with the observer configuration update:
ensure a failed upsert rolls back the template write and returns failure, rather
than being swallowed by the broad exception handler. Remove or replace the
best-effort logger.exception path so resolve_observer_configs cannot serve stale
evaluation_config data after a successful response.
In `@app/database/queries/breeze_buddy/analytics/observer_result.py`:
- Around line 164-181: Update the legacy query’s result selection around the
lateral observer/action expansions to deduplicate detections using DISTINCT ON
(lct.id, observer.value ->> 'name'). Preserve the existing ordering so the
retained row is deterministic, while keeping one result per call and observer
regardless of how many matching actions exist.
- Around line 110-112: Update both get_legacy_observer_detection_rows_query
(app/database/queries/breeze_buddy/analytics/observer_result.py:110-112) and
get_observer_eligible_conversation_count_query
(app/database/queries/breeze_buddy/analytics/observer_result.py:190-195) to
handle the provider filter consistently: either remove provider before calling
build_analytics_where_clause or define the required outbound-number table alias
ou in each query. Apply the same valid approach at both sites so
provider-filtered analytics queries no longer reference an undefined alias.
---
Nitpick comments:
In `@app/api/routers/breeze_buddy/analytics/handlers.py`:
- Around line 287-296: Update the handler flow around
get_observer_detection_rows_from_db, get_legacy_observer_detection_rows_from_db,
and get_observer_eligible_conversation_count_from_db to start all three
independent calls concurrently with asyncio.gather, then unpack their results
before sorting and counting. Preserve the existing filters, limit values, and
output behavior.
In `@app/database/accessor/breeze_buddy/analytics/observer_result.py`:
- Around line 13-36: Wrap the database query execution in each of
get_observer_detection_rows_from_db, get_legacy_observer_detection_rows_from_db,
and get_observer_eligible_conversation_count_from_db with try/except Exception.
In each handler, log the exception using logger.opt(exception=e).error(...) with
relevant accessor context, then re-raise the original exception unchanged.
In `@app/database/queries/breeze_buddy/analytics/observer_result.py`:
- Around line 48-67: The date-filtering logic in the current query builder
duplicates the shared behavior from build_analytics_where_clause. Extract the
date_from/date_to conversion and exclusive date_to upper-bound handling into a
reusable helper, then call it from both paths so IST-to-UTC conversion and
boundary semantics remain identical.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 0eb86c6b-ef17-4699-9186-fb46080fad9a
📒 Files selected for processing (21)
app/ai/voice/agents/breeze_buddy/agent/__init__.pyapp/ai/voice/agents/breeze_buddy/observers/__init__.pyapp/ai/voice/agents/breeze_buddy/observers/factory.pyapp/ai/voice/agents/breeze_buddy/observers/manager.pyapp/ai/voice/agents/breeze_buddy/observers/observer.pyapp/api/routers/breeze_buddy/analytics/__init__.pyapp/api/routers/breeze_buddy/analytics/handlers.pyapp/api/routers/breeze_buddy/templates/handlers.pyapp/database/accessor/breeze_buddy/analytics/observer_result.pyapp/database/accessor/breeze_buddy/analytics/topic_result.pyapp/database/accessor/breeze_buddy/evaluation_config.pyapp/database/accessor/breeze_buddy/evaluation_result.pyapp/database/migrations/045_add_observer_evaluation_type.sqlapp/database/migrations/046_generalize_evaluation_results.sqlapp/database/queries/breeze_buddy/analytics/observer_result.pyapp/database/queries/breeze_buddy/analytics/topic_result.pyapp/database/queries/breeze_buddy/evaluation_config.pyapp/database/queries/breeze_buddy/evaluation_result.pyapp/database/queries/breeze_buddy/topic_result.pyapp/schemas/breeze_buddy/analytics.pytests/test_realtime_observers.py
There was a problem hiding this comment.
Pull request overview
This PR migrates Breeze Buddy “realtime observers” from legacy template.configurations.observers into the evaluation_config table (OBSERVER rows) and records observer detections into the generalized evaluation_result table, then exposes observer-based analytics built on those new records while keeping a legacy fallback path.
Changes:
- Add OBSERVER evaluation type support: new migrations + query/accessor helpers for
evaluation_config(OBSERVER) andevaluation_result(OBSERVER detections). - Update observer runtime to (a) resolve configs from
evaluation_configwith legacy fallback and (b) record observer detections intoevaluation_result. - Add “observer-based” analytics type and SQL/accessors for reporting observer triggers, plus tests covering the new behavior and ordering semantics.
Reviewed changes
Copilot reviewed 21 out of 21 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| tests/test_realtime_observers.py | Adds tests for resolving observer configs from evaluation_config, recording detections, sync/backfill behavior, and alert-vs-terminal ordering. |
| app/schemas/breeze_buddy/analytics.py | Adds OBSERVER_BASED analytics type and observer_name filter option. |
| app/database/queries/breeze_buddy/topic_result.py | Writes topic results into generalized evaluation_result (TOPIC rows). |
| app/database/queries/breeze_buddy/evaluation_result.py | Adds query builder to upsert OBSERVER detections into evaluation_result. |
| app/database/queries/breeze_buddy/evaluation_config.py | Adds get/upsert queries for OBSERVER evaluation_config rows. |
| app/database/queries/breeze_buddy/analytics/topic_result.py | Updates topic analytics queries to read from evaluation_result and filter evaluation_type='TOPIC'. |
| app/database/queries/breeze_buddy/analytics/observer_result.py | Introduces observer analytics queries over evaluation_result with legacy backfill fallback. |
| app/database/migrations/045_add_observer_evaluation_type.sql | Extends evaluation_type enum with OBSERVER. |
| app/database/migrations/046_generalize_evaluation_results.sql | Renames topic_result → evaluation_result, adds evaluation_type, and updates constraints/indexes for generalized storage. |
| app/database/accessor/breeze_buddy/evaluation_result.py | Adds accessor wrapper to save observer detections. |
| app/database/accessor/breeze_buddy/evaluation_config.py | Adds accessor wrappers for get/upsert of observer evaluation_config. |
| app/database/accessor/breeze_buddy/analytics/topic_result.py | Updates log message and framing to reflect evaluation_result backing table. |
| app/database/accessor/breeze_buddy/analytics/observer_result.py | Adds accessors for observer analytics queries (new + legacy fallback). |
| app/api/routers/breeze_buddy/templates/handlers.py | Syncs incoming configurations.observers into OBSERVER evaluation_config row on create/replace template. |
| app/api/routers/breeze_buddy/analytics/handlers.py | Adds get_observer_based_analytics aggregation (summary + trends + breakdowns + recent fires). |
| app/api/routers/breeze_buddy/analytics/init.py | Wires new analytics type → handler mapping. |
| app/ai/voice/agents/breeze_buddy/observers/observer.py | Records observer detections to evaluation_result (best-effort) alongside executing the configured action. |
| app/ai/voice/agents/breeze_buddy/observers/manager.py | Ensures alerts execute before terminal observers in the same batch. |
| app/ai/voice/agents/breeze_buddy/observers/factory.py | Resolves observer configs from OBSERVER evaluation_config row with legacy fallback. |
| app/ai/voice/agents/breeze_buddy/observers/init.py | Exports resolve_observer_configs. |
| app/ai/voice/agents/breeze_buddy/agent/init.py | Uses resolve_observer_configs() to initialize observers from table-backed config. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
2cdb75c to
ebff2ce
Compare
ebff2ce to
2cf21ca
Compare
murdore
left a comment
murdore
left a comment
There was a problem hiding this comment.
Reviewed at 2cf21ca against main, reading the full current version of each changed file rather than the diff alone.
First, the things I checked specifically because they looked like traps, and which are correct:
er.metadata ->> 'handler'looked wrong, sincerecord_detectionpassesresults=[detection]— a JSON array. Butsave_evaluation_results_queryunnests withjsonb_array_elementsand writesresult = btrim(metadata ->> 'type'),metadata = item, so the storedmetadatais the detection object. The accessor is right.ON CONFLICT (template_id, evaluation_type)is backed byevaluation_config_template_type_uniquein migration 044.- The 046/047 split is genuinely necessary — Postgres won't let you use an enum value in the transaction that added it, exactly as the comment in 047 says. Merging them would fail.
- 044's
evaluation_config_json_checkonly requiresjsonb_typeof(configuration) = 'object', so{"observers": [...]}satisfies it. Onlyruntime_checkneeded widening, and 047 does it per-type correctly.
1. MAJOR — the 500-after-commit is right for PUT but wrong for POST
app/api/routers/breeze_buddy/templates/handlers.py:79
sync_observer_evaluation_config raises HTTPException(500) after the template row has already committed, justified in the comment as:
Fail loudly instead; the client retries and the PUT is idempotent.
That holds for replace_template_handler (handlers.py:581), which is @router.put("/templates/{template_id}"). It does not hold for create_template_handler (handlers.py:214), which is @router.post("/templates", status_code=status.HTTP_201_CREATED). A client that retries a failed POST creates a second template, it doesn't re-sync the first — so the failure mode the comment is defending against is traded for silent template duplication.
Worth noting the create path degrades gracefully without the raise: with no evaluation_config row, resolve_observer_configs falls back to the template JSON, so observers still run with the correct config. Only the detection recording is lost, and record_detection already logs that case. The stakes on create are much lower than on replace.
Suggested shape — let the caller decide:
async def sync_observer_evaluation_config(
template_id: str,
configurations: Optional[Dict[str, Any]],
*,
raise_on_failure: bool = True,
) -> None:
...
except Exception as e:
logger.exception(...)
if not raise_on_failure:
return
raise HTTPException(...) from eand pass raise_on_failure=False at the create call site.
2. MINOR (latent) — filter coverage diverges between the two halves of the UNION
app/database/queries/breeze_buddy/analytics/observer_result.py:22
_observer_result_filters honours template/template_id, reseller_id(s), merchant_id(s), observer_name, date_from, date_to. The legacy branch uses build_analytics_where_clause, which additionally honours status, outcome, call_direction, request_id, campaign_id, provider and payload_filters.
get_observer_aggregate_rows_query (:224) UNION ALLs the two, so any of those seven filters would scope the legacy half while leaving the evaluation_result half unscoped. eligible_conversations meanwhile comes from get_observer_eligible_conversation_count_query, which is fully filtered — so total_fires would be inflated against a correctly-filtered denominator and trigger_rate could exceed 100%.
This is not reachable today: the console sends only template_id, date_from, date_to and observer_name, all of which are handled. Flagging it because the divergence is invisible at the call site — the next caller that passes provider or outcome gets silently wrong numbers rather than an error. A comment on _observer_result_filters naming the unsupported keys would be enough.
3. MINOR — the source_id fallback can silently drop rows from analytics
app/ai/voice/agents/breeze_buddy/observers/utils.py:78
source_id = str(getattr(lead, "id", None) or getattr(lead, "call_id", ""))This was raised earlier and the thread is resolved, but the fallback is still at head, so recording what it costs: get_observer_detection_rows_query joins LEFT JOIN lead_call_tracker lct ON lct.id::text = er.source_id and then requires lct.execution_mode IN ('TELEPHONY', 'HOLD_TRANSFER') in the WHERE. A row written with a call_id misses the join, execution_mode is NULL, and the predicate drops it — so the detection is persisted but never appears in analytics, with nothing logged. If lead.id is genuinely always present, dropping the fallback would make that assumption explicit instead of failing quietly.
Nothing here blocks. Item 1 is the one I'd want changed before merge.
7e0f4cb to
4a6f6fe
Compare
murdore
left a comment
murdore
left a comment
There was a problem hiding this comment.
✅ Verified fixed at 4a6f6fe83a02 — re-checked against the new head rather than assuming.
sync_observer_evaluation_config now takes the flag and the create path opts out:
async def sync_observer_evaluation_config(
template_id: str,
configurations: Optional[Dict[str, Any]],
*,
raise_on_failure: bool = True,
) -> None:
...
if not raise_on_failure:
return
raise HTTPException(...)with raise_on_failure=False at the create call site (handlers.py:224-225), and replace_template_handler left raising. That keeps the loud failure exactly where a retry is idempotent, and stops a failed sync from turning a successful POST into a duplicated template. The docstring now spells out why the two paths differ, which is the part that will stop this being "simplified" back later. My item is cleared.
The two low-severity notes from my earlier comment still stand and still don't block: _observer_result_filters silently ignores seven filter keys that the legacy branch honours (unreachable from the console today, but the divergence is invisible at the call site), and the source_id = lead.id or lead.call_id fallback drops detections from analytics rather than merely unjoining them when it fires.
- migrated observer config to evaluation config table. - add support to store result in evaluation result table.
4a6f6fe to
efba922
Compare
| # model-authored and can quote the transcript, so anything not named here is | ||
| # dropped rather than written to evaluation_result. | ||
| _DETECTION_ALLOWED_KEYS = ("reason", "confidence") | ||
| _DETECTION_VALUE_MAX_CHARS = 300 |
There was a problem hiding this comment.
can we add this in dynamic.py
Summary by CodeRabbit