Add harvest_input and knowledge_queue models with pipeline support - #989
Conversation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
Summary by CodeRabbit
WalkthroughChangesModule B noise-filter pipeline
Estimated code review effort: 3 (Moderate) | ~25 minutes Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 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: 1
🧹 Nitpick comments (3)
application/utils/noise_filter/pipeline.py (1)
34-47: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
RunSummary.statusis hardcoded to"ok"and never reflects run outcome.The field is initialized once and never updated (e.g., to signal partial failure when
parse_errors > 0), so callers parsing the JSON summary can't distinguish a clean run from one with parse errors except by separately checkingparse_errors. Either drop the unused field or set it meaningfully (e.g.,"partial"whenparse_errorsor dropped rows are non-zero).🤖 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 `@application/utils/noise_filter/pipeline.py` around lines 34 - 47, Update RunSummary.status so it reflects the run outcome instead of remaining hardcoded to "ok"; set it to the established partial-failure value when parse_errors or dropped_noise is non-zero, while preserving "ok" for clean runs, or remove the field entirely if no meaningful status handling exists.cre.py (1)
304-318: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winInconsistent CLI validation UX vs.
--export/--csv.
--exportwithout--csvis validated inline withparser.error(...)(clean usage message, exit 2).--run_noise_filterwithout--run_idis instead validated insidecre_main.run()by raisingValueError, which surfaces as an unhandled traceback. Consider mirroring the--exportpattern here for a consistent CLI experience.♻️ Suggested fix
args = parser.parse_args() if args.export and not args.csv: parser.error("--export requires --csv <path>") + if args.run_noise_filter and not args.run_id.strip(): + parser.error("--run_noise_filter requires --run_id <pipeline_run_id>")🤖 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 `@cre.py` around lines 304 - 318, Validate the --run_noise_filter and --run_id dependency during CLI argument parsing, alongside the existing --export/--csv validation, using parser.error(...) when --run_noise_filter is set without a nonempty --run_id. Remove or bypass the corresponding ValueError validation in cre_main.run() so invalid CLI input produces the standard usage error instead of an unhandled traceback.application/database/db.py (1)
275-277: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winStatus comment omits the
errorstate actually used by the pipeline.
pipeline.pysetsrow.status = "error"for rows that failChangeRecordvalidation, but this comment only documentspending | processed. Update the comment (and ideally add a comment/constraint enumerating all three values) to keep the state machine documented accurately.📝 Suggested fix
status = sqla.Column( sqla.String, nullable=False, default="pending" - ) # pending | processed + ) # pending | processed | error🤖 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 `@application/database/db.py` around lines 275 - 277, Update the status documentation on the status column in the database model to enumerate all pipeline states: pending, processed, and error. If an existing constraint or state declaration is present nearby, keep it synchronized with these three values without changing the pipeline behavior.
🤖 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 `@application/utils/noise_filter/queue_writer.py`:
- Around line 81-104: Update write_verdicts() to use database-level idempotent
insertion for content_hash, such as PostgreSQL ON CONFLICT DO NOTHING, instead
of relying on the existing pre-query and in-memory deduplication. Ensure
concurrent duplicate hashes are skipped without aborting the transaction or
losing unrelated batch inserts, while preserving the existing WriteStats counts.
---
Nitpick comments:
In `@application/database/db.py`:
- Around line 275-277: Update the status documentation on the status column in
the database model to enumerate all pipeline states: pending, processed, and
error. If an existing constraint or state declaration is present nearby, keep it
synchronized with these three values without changing the pipeline behavior.
In `@application/utils/noise_filter/pipeline.py`:
- Around line 34-47: Update RunSummary.status so it reflects the run outcome
instead of remaining hardcoded to "ok"; set it to the established
partial-failure value when parse_errors or dropped_noise is non-zero, while
preserving "ok" for clean runs, or remove the field entirely if no meaningful
status handling exists.
In `@cre.py`:
- Around line 304-318: Validate the --run_noise_filter and --run_id dependency
during CLI argument parsing, alongside the existing --export/--csv validation,
using parser.error(...) when --run_noise_filter is set without a nonempty
--run_id. Remove or bypass the corresponding ValueError validation in
cre_main.run() so invalid CLI input produces the standard usage error instead of
an unhandled traceback.
🪄 Autofix (Beta)
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: Path: .coderabbit.yml
Review profile: CHILL
Plan: Pro Plus
Run ID: efb18730-0a64-41ab-87d7-358a43ab2162
📒 Files selected for processing (8)
application/cmd/cre_main.pyapplication/database/db.pyapplication/tests/noise_filter/pipeline_test.pyapplication/tests/noise_filter/queue_writer_test.pyapplication/utils/noise_filter/pipeline.pyapplication/utils/noise_filter/queue_writer.pycre.pymigrations/versions/d4e5f6a7b8c9_add_module_b_tables.py
|
hey @northdpole , @Pa04rth is this PR intentionally on hold so module A's PR can get merged or its in the pipeline for a later review. just asking if any changes are required or something like that. |
e9fc308 to
8261930
Compare
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (2)
application/database/db.py (1)
318-320: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winStatus comment is stale — "error" is a valid third state.
The inline comment lists only
pending | processed, butrun_noise_filter(application/utils/noise_filter/pipeline.py) also setsstatus = "error"for rows that failChangeRecordvalidation. Update the comment to document all three states so future readers/queries (e.g. anything filtering onstatus != "processed") don't miss the error state.✏️ Proposed fix
status = sqla.Column( sqla.String, nullable=False, default="pending" - ) # pending | processed + ) # pending | processed | error🤖 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 `@application/database/db.py` around lines 318 - 320, Update the inline comment on the status column to document all valid states: pending, processed, and error. Leave the column definition and behavior unchanged.migrations/versions/d4e5f6a7b8c9_add_module_b_tables.py (1)
76-76: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winConsider a partial index for the unconsumed-lookup index.
ix_knowledge_queue_unconsumedindexesconsumed_atfor all rows, but Module C's access pattern (per theKnowledgeQueueItemdocstring) isWHERE consumed_at IS NULL. A partial index keeps the index small as consumed rows accumulate over time on this long-lived queue table.♻️ Optional refactor
- op.create_index("ix_knowledge_queue_unconsumed", "knowledge_queue", ["consumed_at"]) + op.create_index( + "ix_knowledge_queue_unconsumed", + "knowledge_queue", + ["consumed_at"], + postgresql_where=sa.text("consumed_at IS NULL"), + )🤖 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 `@migrations/versions/d4e5f6a7b8c9_add_module_b_tables.py` at line 76, Update the ix_knowledge_queue_unconsumed definition in the migration to be a partial index covering only rows where consumed_at IS NULL, matching the KnowledgeQueueItem lookup pattern while preserving the existing index name and column.
🤖 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 `@application/utils/noise_filter/pipeline.py`:
- Around line 117-119: Validate that classifier.classify_batch() returns exactly
one verdict for every survivor before constructing triples or updating statuses;
fail the batch through the existing error path if lengths differ. Replace the
truncating zip(survivors, verdicts) in the triples construction with a
strict/equivalent length-checked pairing so incomplete verdicts never reach
write_verdicts() or the processed-status commit.
---
Nitpick comments:
In `@application/database/db.py`:
- Around line 318-320: Update the inline comment on the status column to
document all valid states: pending, processed, and error. Leave the column
definition and behavior unchanged.
In `@migrations/versions/d4e5f6a7b8c9_add_module_b_tables.py`:
- Line 76: Update the ix_knowledge_queue_unconsumed definition in the migration
to be a partial index covering only rows where consumed_at IS NULL, matching the
KnowledgeQueueItem lookup pattern while preserving the existing index name and
column.
🪄 Autofix (Beta)
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: Path: .coderabbit.yml
Review profile: CHILL
Plan: Pro Plus
Run ID: 3025a97a-7541-46bd-b5a8-03d55a33bea5
📒 Files selected for processing (8)
application/cmd/cre_main.pyapplication/database/db.pyapplication/tests/noise_filter/pipeline_test.pyapplication/tests/noise_filter/queue_writer_test.pyapplication/utils/noise_filter/pipeline.pyapplication/utils/noise_filter/queue_writer.pycre.pymigrations/versions/d4e5f6a7b8c9_add_module_b_tables.py
northdpole
left a comment
northdpole
left a comment
There was a problem hiding this comment.
Maintainer review (2026-07-31)
Solid Module B wiring — recall-first, idempotent queue writes, misaligned-classifier guard, dry-run, and run-scoping look correct. Migration chain is a single head (d4e5f6a7b8c9 → a1b2c3d4e5f6). CI green; local pipeline_test + queue_writer_test 11/11.
Please address the inline comments before merge (medium: validation log redaction; low: CLI parser.error; nit: status comment). Once fixed and CI is green, this is merge-ready.
Non-blocking notes (no action required):
write_verdictscommits before rows are markedprocessed— crash between is safe via dedup (may re-spend LLM).- No
SELECT FOR UPDATEif two workers hit the samerun_id(fine for single-runner orchestrator).
d97d1ec to
3298899
Compare
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (2)
application/utils/noise_filter/queue_writer.py (2)
34-36: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd concrete types at the SQLAlchemy boundary.
_row_values()returns a baredict._dialect_insert()andwrite_verdicts()leavesessionuntyped. The current repository target runsmypy --strict application; verify that this branch uses the same target, because these declarations can fail strict type checking. (raw.githubusercontent.com)Use a concrete session and insert type, plus a typed row mapping. Then run the required checks.
As per coding guidelines, run
make mypyfor Python type checking andmake lintto enforce black code formatting and frontend prettier formatting.Also applies to: 77-89
🤖 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 `@application/utils/noise_filter/queue_writer.py` around lines 34 - 36, Update _row_values, _dialect_insert, and write_verdicts with concrete SQLAlchemy session, insert, and row-mapping types rather than bare dict or untyped parameters. Verify the annotations satisfy the repository’s mypy --strict application target, then run make mypy and make lint to validate typing and formatting.Sources: Coding guidelines, Linked repositories
77-83: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winMake the database capability check explicit.
Lines [77-83] map every non-PostgreSQL dialect to the SQLite-specific
Insert. Lines [114-120] also requireRETURNING. SQLAlchemy documents theseInsertvariants as dialect-specific. Its SQLite documentation states that explicitRETURNINGrequires SQLAlchemy 2.0 and SQLite 3.35 or newer. (docs.sqlalchemy.org)If the project supports only PostgreSQL and SQLite, reject other dialects explicitly. Otherwise, add a capability-aware fallback that does not require
RETURNING.Also applies to: 114-120
🤖 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 `@application/utils/noise_filter/queue_writer.py` around lines 77 - 83, Update _dialect_insert to accept only supported PostgreSQL and SQLite dialects, raising an explicit error for any other backend instead of defaulting to SQLite. Ensure the related insert-and-returning flow around KnowledgeQueueItem also handles the documented SQLAlchemy/SQLite RETURNING capability, either by enforcing the supported version or by providing a fallback that avoids RETURNING.
🤖 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 `@application/utils/noise_filter/pipeline.py`:
- Around line 53-59: The pipeline should preserve the original ChangeRecord as
the canonical record for content_hash, queue deduplication, and write_verdicts,
while using the sanitized copy only for classifier input. Update the flow around
_sanitized() and the subsequent classification/persistence steps so sanitized
text cannot alter span provenance or persisted offsets.
---
Nitpick comments:
In `@application/utils/noise_filter/queue_writer.py`:
- Around line 34-36: Update _row_values, _dialect_insert, and write_verdicts
with concrete SQLAlchemy session, insert, and row-mapping types rather than bare
dict or untyped parameters. Verify the annotations satisfy the repository’s mypy
--strict application target, then run make mypy and make lint to validate typing
and formatting.
- Around line 77-83: Update _dialect_insert to accept only supported PostgreSQL
and SQLite dialects, raising an explicit error for any other backend instead of
defaulting to SQLite. Ensure the related insert-and-returning flow around
KnowledgeQueueItem also handles the documented SQLAlchemy/SQLite RETURNING
capability, either by enforcing the supported version or by providing a fallback
that avoids RETURNING.
🪄 Autofix (Beta)
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: Path: .coderabbit.yml
Review profile: CHILL
Plan: Pro Plus
Run ID: cacf58bc-c496-490e-9097-703efb254d24
📒 Files selected for processing (8)
application/cmd/cre_main.pyapplication/database/db.pyapplication/tests/noise_filter/pipeline_test.pyapplication/tests/noise_filter/queue_writer_test.pyapplication/utils/noise_filter/pipeline.pyapplication/utils/noise_filter/queue_writer.pycre.pymigrations/versions/d4e5f6a7b8c9_add_module_b_tables.py
🚧 Files skipped from review as they are similar to previous changes (5)
- application/cmd/cre_main.py
- application/database/db.py
- migrations/versions/d4e5f6a7b8c9_add_module_b_tables.py
- application/tests/noise_filter/pipeline_test.py
- application/tests/noise_filter/queue_writer_test.py
|
@northdpole Finally all green ready for the review of the updated fixes. Let me know if some more fix is required. |
northdpole
left a comment
northdpole
left a comment
There was a problem hiding this comment.
Re-reviewed after your Jul 31 fixes — all three requested items look good:
- validation logs use
e.errors(include_input=False) --run_noise_filterwithout--run_idgoes throughparser.errorincre.py- HarvestInput status comment includes
error
Migration still a single head (d4e5f6a7b8c9 → 6a9d0d62ef41). CI green. LGTM.
2 participants
Summary
Wires Module B (Noise/Relevance Filter) into the orchestrated pipeline: it reads
Module A's harvested chunks from a database table, classifies them (regex →
sanitize → LLM), writes the security-knowledge keepers to a queue for Module C,
and reports a JSON summary. Builds on the Stage 1/1.5/2 code from Weeks 1–4.
What's added
application/database/db.py):harvest_input— Module A writes harvested chunks here (JSONBpayload+pipeline_run_id+status); Module B reads.knowledge_queue— Module B writes classified keepers here (deduped oncontent_hash); Module C reads.queue_writer.py— maps a verdict to aknowledge_queuerow, source-typeaware, deduped on
content_hash; NOISE dropped, KNOWLEDGE/UNCERTAIN kept.pipeline.py(run_noise_filter) — for apipeline_run_id: read pendingharvest_inputrows → regex → sanitize → LLM classify → write keepers → markrows processed → return a
RunSummary.cre.py/cre_main.py) —--run_noise_filter --run_id <id> [--noise_filter_dry_run], prints the summary as JSON and exits (the entrypoint the orchestrator invokes; exit code = completion signal).
d4e5f6a7b8c9— creates the two tables (down_revisionc7d8e9f0a1b2).Behavior
UNCERTAIN always reach the queue.
processed/error; re-running arun_idissafe;
UNIQUE(content_hash)collapses duplicate content.error(not fatal); failed LLM batch→ those chunks become UNCERTAIN (never dropped); infra failure → non-zero exit.
Testing
suite 95/95,
black --checkclean.flask db upgradefrom an empty DBbuilds the schema through our head; a real run (
cre.py --run_noise_filter)classifies a mixed batch, writes KNOWLEDGE rows to
knowledge_queue, and exits0 with a JSON summary.
Notes
queue_writeris the only DB boundary.litellm(slim prod reqs).