feat: Phase 5-6 — stream processing, drift detection, and a full review-driven hardening pass#45
Merged
Conversation
…ng, and update classification logic to support multi-class outputs.
…on and persistence to PostgreSQL and MongoDB
…n attributes to events and renaming the classifier endpoint to /v1/moderations
…nd implement batching for moderation requests
…and add queue monitoring
… reliability with database connection handling.
…code review Fixes seven issues flagged as highest-severity in a multi-agent code review (#29, #30, #31, #32, #33, #34, #5) — each one triggers on a normal operation (restart, redeploy, routine drift run) and either loses data or corrupts a signal something downstream automates on. - Kafka: point KAFKA_LOG_DIRS at the mounted PVC — apache/kafka:latest was defaulting to /tmp/kraft-combined-logs, silently discarding all topic data on every pod restart (#29) - Postgres: move drift_stats into Terraform's schema init so a plain `terraform apply` provisions it; drop the now-redundant manual migration from dev-start.sh (#30) - Drift job: read the active model version from model_registry (mirroring the classifier's own active/staging selection) instead of "last row written to classifications wins", which raced during rolling restarts (#33) - Drift job: exit early on a too-small reference window instead of computing PSI/JSD against an unreliable baseline and risking a false drift_flagged=True (#34) - Classifier batcher: stop() now drains the queue and fails in-flight batches with a clear exception instead of leaking futures that hang forever past a CancelledError (#31) - Stream processor: flagged_content writes upsert on (span_id, text_type) via bulk_write, with a matching unique partial index in Mongo, so Kafka redelivery can no longer duplicate retraining data (#32) - Evaluation pipeline: implement benchmark.py + validate.py (previously empty files) — scores a candidate model against a real 3780-row ground-truth set and gates promotion on accuracy vs. a minimum and vs. baseline, closing the "any model gets promoted with zero validation" gap (#5) Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
…nd bucket mismatch Fixes six "fix soon" issues from the code review (#35, #36, #37, #6, #42, #43) — real bugs with bounded blast radius, as opposed to the data-loss/false-signal class fixed in 03d2f9b. - Optimizer: derive model_path from the quantize stage's actual upload_stage() return value instead of hardcoding the "models" bucket prefix, so it can't diverge from a customized MINIO_BUCKET (#35) - Drift job: cap read_current_scores at 100k most-recent rows (via a subquery, not a naive LIMIT that would return the oldest rows) to bound driver memory on a large --hours window (#36) - Classifier: close the asyncpg pool before discarding it on a post-creation startup failure, instead of leaking the connection (#37) - Stream processor: close the replacement connection too when a reconnect retry itself fails, instead of leaking it on every double-failure (#6) - Classifier: unify MAX_BATCH_SIZE — schemas.py now reads config.py's settings.max_batch_size instead of a second hardcoded constant, so the env var actually controls both the batcher and the request-validation cap (previously the ClassifierBatchBackpressure alert's "raise MAX_BATCH_SIZE" remediation was a no-op). Stream processor's chunk size is now the CLASSIFY_CHUNK_SIZE env var instead of a silent hardcoded copy (#42) - Classifier: skip registering a local-filesystem model_path in model_registry — that path only exists on one pod's filesystem, and a second pod reading it back from the registry would crash on startup (#43) Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
…d API cleanup Fixes the last 20 open review issues (#8, #9, #13, #14, #15, #16, #17, #19, #20, #21, #22, #23, #24, #25, #27, #28, #38, #39, #40, #41, #44), closes #26 as resolved-by-different-means rather than fixed. Classifier service: - Switch tokenizer padding to max_length so inference latency doesn't vary with what else happens to land in the same batch (#8) - Extract _deployed_at_from_dir() to remove duplicated onnx-mtime logic between Classifier.__init__ and _resolve_model_dir; parse each report.json once instead of twice when auto-selecting a model (#19, #20) - Extract _classify_and_persist() shared by classify_batch()/moderate() — removes ~25 duplicated lines of timing/metrics/persistence logic (#21) - Remove the persist field from ModerationRequest (OpenAI-compatible public schema) now that nothing needs to skip persistence there — the stream processor calls /classify/batch instead, which keeps its own persist field since that endpoint is explicitly internal/testing-only (#24) - write_classification/write_classifications_batch now include the same span_id/text_type columns + ON CONFLICT DO NOTHING conflict target as the stream processor's writer, instead of a plain INSERT with no idempotency (#25) - Update CLAUDE.md's classifier design rule to describe the actual async+executor/batcher pattern instead of a stale "must be sync def" rule the code no longer followed (#40) - Cache the MinIO S3 client with lru_cache instead of rebuilding it (TLS handshake + credential resolution) on every download call (#17) Stream processor: - Call /classify/batch instead of /v1/moderations — returns {label, score} directly, eliminating the flagged/category_scores.harm remapping that coupled this consumer to the OpenAI-compatible wire format (#23) Optimizer pipeline: - Add pipelines/optimizer/__main__.py; invoke via `python -m pipelines.optimizer` instead of the fragile `...pipeline` module path (updated README.md, docs/local-dev.md, dev-start.sh, explanation.md) (#13) - Cache the MinIO S3 client; parallelize upload_stage()'s file uploads via ThreadPoolExecutor instead of uploading one at a time (#17, #27) - register_model() now takes a connection instead of opening its own per call; pipeline.py dedupes the register+report if/else branches that only differed by model_path and a suffix string (#22, #28) Infra: - Pin apache/kafka and jaegertracing/all-in-one off :latest to specific versions for reproducible applies (#9) - Align docker-compose's Prometheus/Grafana versions with Terraform's (the primary deployment path) instead of drifting onto different major versions (#16) - Split prometheus.compose.yml from prometheus.yml — the classifier scrape target differs (compose network name vs. k8s FQDN), and listing both targets in one file would leave one permanently down in every environment, firing ClassifierDown nonstop (#41) Dev tooling: - Wrap all 9 wait_for_port call sites so a timeout warns and continues instead of aborting the whole script under set -e (#38) - Replace bare SQL string interpolation (PG_PASSWORD, $_minio_path) with psql -v variable substitution (#39) - Delete the dead mock_predict fixture and the placeholder test now that real tests exist (#14, #15) Drift pipeline: - Symmetric clamp in _bin_scores: negative scores now fold into bin 0 instead of silently vanishing from the PSI/JSD left join (#44) #26 (ad-hoc SQL migrations) is closed as resolved-by-different-means: schema ownership already moved fully into Terraform's postgres_init ConfigMap in an earlier commit; adopting Alembic/Flyway now would be new infrastructure ahead of this project's current phase. Verified: classifier test suite (15/15), drift PySpark test suite (14/14), terraform validate, two live smoke tests (real ONNX model load + predict, real /classify/batch HTTP call against a running server), and a direct PySpark check that negative scores clamp to bin 0. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
…sion Commit a1f4421's fix for #23 moved the stream processor from /v1/moderations back to /classify/batch to avoid a response-shape remapping. That undid a deliberate architectural decision from 46ea2aa ("adopting OpenTelemetry GenAI semantic conventions... renaming the classifier endpoint to /v1/moderations"), which had moved the stream processor onto /v1/moderations specifically so Sentinel's own highest-volume internal traffic exercises the same OpenAI-compatible code path external callers use. I made that change without checking git history for why the endpoint was chosen in the first place — caught after the fact. Reverted the endpoint choice back to /v1/moderations. Properly fixed the actual problem #24 was about (the `persist` field leaking Sentinel-internal plumbing into the public OpenAI-compatible schema) using the header-based approach that issue originally suggested and I'd skipped: an X-Sentinel-Skip-Persist request header instead of a body field. ModerationRequest carries no Sentinel-internal fields now, and the stream processor still gets to skip classifier-side persistence. The response-shape coupling #23 raised is real but inherent to calling an OpenAI-compatible endpoint from internal code — extracted it into a named _moderation_results_to_label_score() helper so the translation is a visible, intentional boundary rather than an inline remap, without abandoning the dogfooding architecture. Verified: 15/15 classifier tests pass, and two live smoke tests against a running server — /v1/moderations without the header returns the full public response, with the header still returns correctly (schema/routing confirmed; DB persistence itself untestable here with no live database). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
|
Warning Review limit reached
Next review available in: 56 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Free Run ID: ⛔ Files ignored due to path filters (2)
📒 Files selected for processing (59)
Note 🎁 Summarized by CodeRabbit FreeYour organization is on the Free plan. CodeRabbit will generate a high-level summary and a walkthrough for each pull request. For a comprehensive line-by-line review, please upgrade your subscription to CodeRabbit Pro by visiting https://app.coderabbit.ai/login. Comment |
CI's Lint & format job failed on PR #45 — 4 pre-existing unsorted-import errors (pipelines/drift/tests/test_db.py, test_metrics.py; the import I added to pipelines/optimizer/pipeline.py wasn't alphabetized either) plus formatting drift across files touched in this session's fix commits. No behavior change — ruff check --fix + ruff format only. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
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.
Summary
Merges Phases 5-6 (trace ingestion/stream processing + drift detection) along with a full round of code-review-driven fixes.
Features:
/v1/moderations(OpenAI Moderation API-compatible)pipelines/evaluation/) implemented from scratch: benchmarks a candidate model against a real 3780-row ground-truth set and gates promotion on accuracy vs. a minimum and vs. baselineFixes (from a structured multi-agent code review of the full branch — 34 issues found and resolved across three passes):
drift_statstable missing from Terraform's schema (only created by an ad-hoc script), drift job attributing stats to the wrong model version during rollouts, a false-positive drift signal possible on a too-small reference window, classifier batcher leaking futures (requests hang forever) on shutdown, MongoDB writes not idempotent under Kafka redeliverydev-start.sherror handling that aborted the whole script underset -einstead of warning and continuing, bare SQL string interpolation replaced withpsql -vsubstitution, Kafka/Jaeger images pinned off:latest, Prometheus/Grafana version drift between the k8s and docker-compose dev paths resolvedOne correction worth flagging in review: an earlier fix in this pass had moved the stream processor from
/v1/moderationsback to/classify/batchto avoid a response-shape remapping — this accidentally reverted a deliberate prior decision (dogfooding the same OpenAI-compatible endpoint external callers use). It's been corrected back to/v1/moderations, with the actual underlying concern (an internalpersistfield leaking into the public schema) fixed properly via anX-Sentinel-Skip-Persistheader instead.Test plan
pytest tests/— 15/15 passingpytest pipelines/drift/tests/test_metrics.py— 14/14 passing (PySpark)terraform validate— passing/classify/batchand/v1/moderations(with and withoutX-Sentinel-Skip-Persist) against a running server🤖 Generated with Claude Code