Feat/datainfra refactor tracker#98
Conversation
|
Caution Review failedThe pull request is closed. 📝 WalkthroughWalkthroughThis pull request introduces comprehensive workflow metrics instrumentation across the codebase. It adds a new Changes
Sequence Diagram(s)sequenceDiagram
participant Client as Client/Caller
participant Route as API Route<br/>(paperscool/research)
participant Store as WorkflowMetricStore
participant DB as Database<br/>(workflow_eval_metrics)
Client->>Route: Request workflow execution
activate Route
Route->>Route: Start timing<br/>(time.perf_counter)
rect rgba(100, 150, 200, 0.5)
Note over Route: Execute workflow phase<br/>(search, build, enrich, etc.)
Route->>Route: Phase completion
end
Route->>Store: record_metric(workflow, stage,<br/>claim_count, evidence_count, elapsed_ms)
activate Store
Store->>Store: Compute coverage_rate<br/>Serialize detail JSON
Store->>DB: INSERT INTO workflow_eval_metrics
deactivate Store
DB-->>Store: Metric persisted
Store-->>Route: Return dict representation
rect rgba(150, 200, 100, 0.5)
Note over Route: Workflow completes<br/>(success/failure)
end
Route-->>Client: Return result
deactivate Route
Client->>Route: GET /metrics/workflows?days=7
activate Route
Route->>Store: summarize(days=7, workflow=..., track_id=...)
activate Store
Store->>DB: Query metrics within window<br/>with filters
Store->>Store: Aggregate by:<br/>- total (runs, success, failed)<br/>- per-day buckets<br/>- per-workflow buckets<br/>- per-track buckets
Store->>Store: Compute averages<br/>and coverage rates
deactivate Store
Store-->>Route: Return summary dict
Route->>Route: Wrap in WorkflowMetricsResponse
Route-->>Client: Return metrics summary (HTTP 200)
deactivate Route
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Possibly related PRs
Poem
✨ 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 |
Summary of ChangesHello @jerry609, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request significantly enhances the application's observability by implementing a comprehensive system for tracking and analyzing workflow performance. It provides a structured way to collect and store detailed metrics about the execution of critical processes, enabling better monitoring of their efficiency and effectiveness. The addition of a dedicated API endpoint for these metrics facilitates easy access to aggregated data, supporting data-driven insights into workflow health and areas for improvement. Highlights
🧠 New Feature in Public Preview: You can now enable Memory to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console. Changelog
Activity
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
There was a problem hiding this comment.
Pull request overview
Adds a workflow-metrics persistence + aggregation layer (SQLAlchemy-backed) and wires it into the Research context and PapersCool workflows, plus exposes a research API endpoint to query summarized workflow metrics.
Changes:
- Introduces
WorkflowEvalMetricModel+WorkflowMetricStoreto record and summarize workflow runs (status/latency/coverage). - Adds a new Research route
GET /api/research/metrics/workflowsto fetch aggregated workflow metrics. - Instruments
research.build_contextand multiplepaperscoolexecution paths to record workflow metrics; adds unit tests for store + route.
Reviewed changes
Copilot reviewed 6 out of 6 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
src/paperbot/infrastructure/stores/models.py |
Adds WorkflowEvalMetricModel table for workflow metrics persistence. |
src/paperbot/infrastructure/stores/workflow_metric_store.py |
New store for recording metrics and summarizing totals/buckets. |
src/paperbot/api/routes/research.py |
Adds metrics summary endpoint and records metrics for build_context. |
src/paperbot/api/routes/paperscool.py |
Records workflow metrics across daily/analyze/approval flows and adds evidence/claim counting helper. |
tests/unit/test_workflow_metric_store.py |
Unit tests for record + summarize behavior and filtering. |
tests/unit/test_research_workflow_metrics_route.py |
Route test for workflow metrics summary endpoint. |
Comments suppressed due to low confidence (1)
src/paperbot/api/routes/research.py:1151
- When
activate_track_idis provided but the track is not found, the handler raises before the try/except that records workflow metrics. This creates a blind spot where an important failure mode is never recorded. Consider recording a failed metric (with a distinct stage/status) just before raising the 404 so the request is still observable.
if req.activate_track_id is not None:
Logger.info("Activating research track", file=LogFiles.HARVEST)
activated = _research_store.activate_track(
user_id=req.user_id, track_id=req.activate_track_id
)
if not activated:
Logger.error("Research track not found", file=LogFiles.HARVEST)
raise HTTPException(status_code=404, detail="Track not found")
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| assert summary["totals"]["runs"] >= 1 | ||
| assert summary["totals"]["claim_count"] >= 12 | ||
| assert summary["totals"]["evidence_count"] >= 9 |
There was a problem hiding this comment.
This test uses >= assertions for totals even though it writes to a fresh per-test SQLite file. Using exact expectations (e.g. runs==1, claim_count==12, evidence_count==9) would make the test more precise and better at catching regressions in filtering/aggregation.
| assert summary["totals"]["runs"] >= 1 | |
| assert summary["totals"]["claim_count"] >= 12 | |
| assert summary["totals"]["evidence_count"] >= 9 | |
| assert summary["totals"]["runs"] == 1 | |
| assert summary["totals"]["claim_count"] == 12 | |
| assert summary["totals"]["evidence_count"] == 9 |
|
|
||
| for row in rows: | ||
| totals["runs"] += 1 | ||
| if row.status == "completed": | ||
| totals["success_runs"] += 1 | ||
| else: |
There was a problem hiding this comment.
summarize() currently treats any status other than "completed" as a failure. Call sites record statuses like "pending_approval" and "rejected" (e.g. paperscool), which will be miscounted as failed runs and inflate failure_stages. Consider explicitly classifying statuses (e.g. success={"completed"}, failure={"failed"}, and either ignore/track separately the rest) so pending/rejected don’t show up as runtime failures.
| for row in rows: | |
| totals["runs"] += 1 | |
| if row.status == "completed": | |
| totals["success_runs"] += 1 | |
| else: | |
| success_statuses = {"completed"} | |
| failure_statuses = {"failed"} | |
| for row in rows: | |
| totals["runs"] += 1 | |
| status = (row.status or "").lower() | |
| if status in success_statuses: | |
| totals["success_runs"] += 1 | |
| elif status in failure_statuses: |
| bucket["runs"] = int(bucket.get("runs") or 0) + 1 | ||
| if row.status == "completed": | ||
| bucket["success_runs"] = int(bucket.get("success_runs") or 0) + 1 | ||
| else: |
There was a problem hiding this comment.
Same issue as above in per-bucket aggregation: _add_to_bucket() increments failed_runs for any non-"completed" status, so buckets will also misreport pending/rejected runs as failures. Align bucket counting with the status classification used for totals.
| else: | |
| elif row.status == "failed": |
| stmt = select(WorkflowEvalMetricModel).where(WorkflowEvalMetricModel.ts >= since) | ||
| if workflow: | ||
| stmt = stmt.where(WorkflowEvalMetricModel.workflow == str(workflow)[:64]) | ||
| if track_id is not None: | ||
| stmt = stmt.where(WorkflowEvalMetricModel.track_id == int(track_id)) | ||
|
|
||
| with self._provider.session() as session: | ||
| rows = session.execute(stmt).scalars().all() | ||
|
|
There was a problem hiding this comment.
summarize() loads all matching metric rows into memory and aggregates in Python. Over a 90-day window this can become expensive (memory + latency) as the table grows. Consider pushing aggregation into SQL (COUNT/SUM/GROUP BY for totals, by_day, by_workflow, by_track) and only fetching aggregated rows.
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a WorkflowMetricStore to centralize and persist metrics for various workflows, which is a great step towards better observability. The implementation is solid, with a new database model, a store class to handle persistence and aggregation, and integration into the paperscool and research API routes. New tests are also included to cover the new functionality.
My review includes a couple of suggestions for improvement: one regarding code duplication in paperscool.py for better maintainability, and another about optimizing the metric aggregation logic in workflow_metric_store.py for better performance as the number of metrics grows.
| session_id=session_id, result=result_payload, status="completed" | ||
| ) | ||
|
|
||
| claims, evidences = _count_report_claims_and_evidence(report) |
There was a problem hiding this comment.
The calculation of claims and evidences is also performed earlier in the if req.require_approval: block (line 649). To avoid code duplication and improve maintainability, you could calculate these values once after the report is finalized (e.g., after the enrichment phase around line 645) and before the approval check, then reuse the variables in both places.
| def summarize( | ||
| self, | ||
| *, | ||
| days: int = 7, | ||
| workflow: Optional[str] = None, | ||
| track_id: Optional[int] = None, | ||
| ) -> Dict[str, Any]: | ||
| window_days = max(1, min(int(days), 90)) | ||
| since = _utcnow() - timedelta(days=window_days) | ||
|
|
||
| stmt = select(WorkflowEvalMetricModel).where(WorkflowEvalMetricModel.ts >= since) | ||
| if workflow: | ||
| stmt = stmt.where(WorkflowEvalMetricModel.workflow == str(workflow)[:64]) | ||
| if track_id is not None: | ||
| stmt = stmt.where(WorkflowEvalMetricModel.track_id == int(track_id)) | ||
|
|
||
| with self._provider.session() as session: | ||
| rows = session.execute(stmt).scalars().all() | ||
|
|
||
| totals = { | ||
| "runs": 0, | ||
| "success_runs": 0, | ||
| "failed_runs": 0, | ||
| "avg_elapsed_ms": 0.0, | ||
| "claim_count": 0, | ||
| "evidence_count": 0, | ||
| "coverage_rate": 0.0, | ||
| } | ||
|
|
||
| by_day: Dict[str, Dict[str, Any]] = {} | ||
| by_workflow: Dict[str, Dict[str, Any]] = {} | ||
| by_track: Dict[str, Dict[str, Any]] = {} | ||
| failure_stages: Dict[str, int] = defaultdict(int) | ||
|
|
||
| for row in rows: | ||
| totals["runs"] += 1 | ||
| if row.status == "completed": | ||
| totals["success_runs"] += 1 | ||
| else: | ||
| totals["failed_runs"] += 1 | ||
| failure_stages[row.stage or "unknown"] += 1 | ||
|
|
||
| totals["avg_elapsed_ms"] += float(row.elapsed_ms or 0.0) | ||
| totals["claim_count"] += int(row.claim_count or 0) | ||
| totals["evidence_count"] += int(row.evidence_count or 0) | ||
|
|
||
| date_key = (row.ts or _utcnow()).date().isoformat() | ||
| day = by_day.setdefault( | ||
| date_key, | ||
| { | ||
| "date": date_key, | ||
| "runs": 0, | ||
| "success_runs": 0, | ||
| "failed_runs": 0, | ||
| "claim_count": 0, | ||
| "evidence_count": 0, | ||
| "avg_elapsed_ms": 0.0, | ||
| "coverage_rate": 0.0, | ||
| }, | ||
| ) | ||
| self._add_to_bucket(day, row) | ||
|
|
||
| wf = by_workflow.setdefault( | ||
| row.workflow or "unknown", | ||
| { | ||
| "workflow": row.workflow or "unknown", | ||
| "runs": 0, | ||
| "success_runs": 0, | ||
| "failed_runs": 0, | ||
| "claim_count": 0, | ||
| "evidence_count": 0, | ||
| "avg_elapsed_ms": 0.0, | ||
| "coverage_rate": 0.0, | ||
| }, | ||
| ) | ||
| self._add_to_bucket(wf, row) | ||
|
|
||
| track_key = str(row.track_id) if row.track_id is not None else "none" | ||
| track_bucket = by_track.setdefault( | ||
| track_key, | ||
| { | ||
| "track_id": row.track_id, | ||
| "runs": 0, | ||
| "success_runs": 0, | ||
| "failed_runs": 0, | ||
| "claim_count": 0, | ||
| "evidence_count": 0, | ||
| "avg_elapsed_ms": 0.0, | ||
| "coverage_rate": 0.0, | ||
| }, | ||
| ) | ||
| self._add_to_bucket(track_bucket, row) | ||
|
|
||
| if totals["runs"] > 0: | ||
| totals["avg_elapsed_ms"] = round(totals["avg_elapsed_ms"] / totals["runs"], 2) | ||
| totals["coverage_rate"] = self._coverage(totals["claim_count"], totals["evidence_count"]) | ||
|
|
||
| return { | ||
| "window_days": window_days, | ||
| "totals": totals, | ||
| "by_day": [self._finalize_bucket(by_day[k]) for k in sorted(by_day.keys())], | ||
| "by_workflow": sorted( | ||
| (self._finalize_bucket(v) for v in by_workflow.values()), | ||
| key=lambda x: int(x.get("runs") or 0), | ||
| reverse=True, | ||
| ), | ||
| "by_track": sorted( | ||
| (self._finalize_bucket(v) for v in by_track.values()), | ||
| key=lambda x: int(x.get("runs") or 0), | ||
| reverse=True, | ||
| ), | ||
| "failure_stages": dict( | ||
| sorted(failure_stages.items(), key=lambda kv: kv[1], reverse=True) | ||
| ), | ||
| } |
There was a problem hiding this comment.
The summarize method fetches all metric rows from the database within the time window and performs aggregation in Python. For large volumes of metrics, this can become inefficient in terms of memory and performance. Consider leveraging the database for aggregation using SQLAlchemy's func module (e.g., func.sum, func.avg, func.count) and group_by. This would offload the aggregation work to the database, which is highly optimized for such tasks, and reduce the amount of data transferred to the application.
Summary by CodeRabbit
New Features
Tests