M4 [PROTO]: Re-expose bulk filter triage as a streaming Filters RPC #193

New Issue

2026-07-28T14:04:33Z

KofTwentyTwo commented

2026-07-28 14:04:33 +00:00

(Migrated from github.com)

Context

Story for milestone M4. Read docs/STORY-WORKFLOW.md first — it defines the process, gates, and patterns; this issue defines only the specific work. Per the original GH issue (#176) this reconciles: retiring the legacy JSON-RPC IPC removed a filter.triage_keyset handler that had real bulk-triage logic — a chunked scan of ALL stored messages, evaluating every persisted filter rule against each one, writing execution logs and outbox mutations for matches, and reporting batch progress as it went. That handler no longer exists anywhere in the tree (confirmed: no trace of it in _reference/nuncio-mcp/_reference/nuncio-tui, and git log shows no surviving commit content for it beyond the issue's own description) and the current service Filters (CreateRule/ListRules/DeleteRule/ValidateRule/PreviewRule) has no equivalent — today there is no way to apply the currently-active filter rule set retroactively against messages that were already synced before the rules existed (or before M4-S1 wired live evaluation into sync at all). crates/nuncio-store/src/db.rs's get_message_chunk(last_id, limit) (line 1366) already implements exactly the keyset-chunking primitive ("Fetch a page of messages using Keyset Chunking") this capability needs — it is currently unused in production, only exercised by its own unit tests.

Goal

A Filters.Triage gRPC RPC that scans every stored message in bounded chunks (via DatabaseEngine::get_message_chunk), evaluates the daemon's live FilterEngine against each one, applies the SAME action-routing semantics M4-S1 wires into live sync (immediate MARK READ/MARK UNREAD, outbox-enqueued MOVE/COPY/FLAG/UNFLAG/DELETE/FORWARD/CALL WEBHOOK, an execution-log entry per match), and streams batch progress back to the caller as it works through the store — never a single blocking call with no feedback for what could be a very large mailbox. This must not fabricate progress or completion: counts reported on the stream must be the real, cumulative counts of what has actually been scanned/matched/mutated so far.

Files to touch / patterns to mirror

crates/nuncio-proto/proto/nuncio/v1/nuncio.proto — extend service Filters (line 437) with a server-streaming RPC: rpc Triage(TriageRequest) returns (stream TriageProgress); — the ONLY server-streaming RPC pattern already in the proto is System.Subscribe (rpc Subscribe(SubscribeRequest) returns (stream Event);, line 24) — mirror its stream usage. TriageRequest { optional string rule_id = 1; uint32 chunk_size = 2; } (rule_id unset = evaluate against ALL active rules, matching FilterEngine::evaluate's normal behaviour; chunk_size of 0 should mean "use a sane server-side default", e.g. 100). TriageProgress { uint64 scanned_count = 1; uint64 matched_count = 2; uint64 actions_applied_count = 3; string last_message_id = 4; bool done = 5; } — one message emitted after each chunk is processed, with cumulative (not per-chunk) counts, and a final message with done = true.
crates/nunciod/src/grpc.rs — FiltersGrpcService (struct at line 756): add a triage method matching tonic's server-streaming signature. SystemGrpcService::subscribe (around line 171-190) is the only existing server-streaming RPC in this file, but its mechanism does not fit here: it maps an ALREADY-LIVE stream (self.event_bus.subscribe_events(), a tokio::sync::broadcast receiver wrapped in BroadcastStream) — it has nothing to actively compute. Triage instead needs to actively drive a chunked DB scan and push results as it goes, which is a genuinely new pattern in this codebase: spawn a tokio::task that owns the scan loop, feed a tokio::sync::mpsc::channel::<Result<TriageProgress, Status>>, and return Response::new(Box::pin(ReceiverStream::new(rx))) as Self::TriageStream (associated type Pin<Box<dyn Stream<Item = Result<TriageProgress, Status>> + Send + 'static>>, matching SubscribeStream's shape). Only the trait-level shape (associated type + Response<Self::TriageStream>) is worth copying from subscribe; the internal mechanism is different by necessity. Loop inside the spawned task: db.get_message_chunk(last_id, chunk_size) → for each Email, evaluate + route actions (see next bullet) → after the chunk, send a TriageProgress on the mpsc sender (stop looping if the receiver was dropped, i.e. the caller disconnected) → advance last_id to the chunk's last message id → stop when a chunk returns fewer than chunk_size rows and send a final done: true message.
Depends on M4-S1's evaluate-and-route helper. M4-S1 adds a reusable "evaluate FilterEngine against one Email and apply/enqueue its matched actions + write an execution log" helper (as part of wiring live sync) — call that SAME helper here per message rather than re-implementing action-routing a second time. If M4-S1 has not merged yet when you start this story, coordinate with the PO on its exact function signature/location (likely crates/nunciod/src/sync.rs or a new crates/nunciod/src/filter_apply.rs) so both stories converge on one implementation instead of two independently-evolving copies of the same logic.
crates/nuncio-proto/src/client.rs — no new connect_* function needed (connect_filters/AuthenticatedFiltersClient already cover the whole Filters service); the new triage method returns a tonic::Streaming<TriageProgress> the caller reads via repeated .message().await calls.
crates/nuncio-cli/src/args.rs — add a FilterSubcommand::Triage { rule_id: Option<String>, chunk_size: Option<u32> } variant, mirroring the existing subcommand style (lines 379-451).
crates/nuncio-cli/src/runner.rs — add a handle_filter_triage that connects via self.connect_filters_client(), calls .triage(...), and loops reading the returned stream, printing/accumulating progress as it goes (this will be the CLI's first real streaming-RPC consumer — System.Subscribe streaming is currently only exercised by nunciod's own tests, per the comment at runner.rs line 1878, not by the CLI itself; you are not extending that path, just establishing your own for Triage).
crates/nunciod/tests/grpc_auth_test.rs — add an unauthenticated-rejection case for Triage (mirroring rejects_calls_without_a_bearer_token); Filters is already mounted behind its own BearerAuthInterceptor, so this is a regression assertion, not new mounting code.

Implementation steps

Confirm with the PO whether M4-S1 has merged; if so, import and reuse its evaluate-and-route helper directly. If not, either wait or write the minimal version of that helper here and flag to the PO that M4-S1's implementer should converge on it (do not silently let two independent copies of the same action-routing logic exist after both stories land).
Add Triage/TriageRequest/TriageProgress to nuncio.proto, following §5 of STORY-WORKFLOW.md for the golden regen. Confirm with the PO that M4-S3 (also [PROTO], also touching service Filters) is not simultaneously in flight before editing the proto — only one proto-touching story at a time.
Implement FiltersGrpcService::triage: read System::subscribe first (in this same file) for the trait-level shape (associated *Stream type, Response<Self::TriageStream>), then build the actual mpsc-channel-plus-spawned-task mechanism described above (subscribe's own BroadcastStream-over-an-existing-channel approach does not apply here — there is no existing "triage progress" channel to subscribe to; you are actively producing one).
Inside the spawned worker: loop db.get_message_chunk, evaluate+route per message via the shared helper, accumulate cumulative counts, send a TriageProgress after each chunk, and a final done: true message when a chunk comes back short.
Add the CLI subcommand + streaming consumer.
Tests: an offline E2E (via crates/nunciod/tests/, mirroring spine_e2e_test.rs's daemon-boot style) that seeds several stored messages (some matching an active rule, some not) directly through DatabaseEngine::save_email, calls Triage, drains the stream, and asserts: (a) the final cumulative matched_count/actions_applied_count are correct; (b) the matching messages show the expected side effects (read-flag flip / queued outbox row, same assertions style as M4-S1's tests); (c) chunking actually occurred across multiple TriageProgress messages for a store larger than one chunk_size (prove the streaming/chunking is real, not a single batch dressed up as a stream).
Add the unauthenticated-rejection regression test.
Run cargo verify; regenerate the contract golden if the descriptor test fails.

Acceptance criteria

New RPC Triage(TriageRequest) returns (stream TriageProgress) added to service Filters; new messages TriageRequest/TriageProgress added, additive within nuncio.v1.
Contract golden regenerated; nuncio-proto's contract_stability::descriptor_matches_committed_golden test passes.
Triage scans the WHOLE store via real keyset chunking (DatabaseEngine::get_message_chunk), not a single unbounded query.
Matched messages get the same real side effects M4-S1 defines for live sync (immediate local action / enqueued outbox mutation / execution log entry) — reusing that story's helper, not a second independent implementation.
Streamed TriageProgress messages carry real, cumulative, non-fabricated counts; a done: true final message is sent when the scan completes.
E2E proves multi-chunk streaming actually happens for a store bigger than one chunk (not a single disguised batch response).
An unauthenticated call to Triage is rejected with Code::Unauthenticated.
CLI filter triage command consumes the stream and reports progress.
cargo verify green (fmt + clippy -D warnings all-targets + test-all), 0 ignored.

Definition of done & process

Follow docs/STORY-WORKFLOW.md: no fabricated data; comment convention (intent-only, no story/GH/phase refs in .rs/.proto); mock-only tests. Branch story/m4-filters-triage off dev; PR to dev with the gate output pasted as evidence; the PO reviews before the issue is closed.

Dependencies / notes

[PROTO]: confirm with the PO that no other proto-touching story is in flight before starting — M4-S3 also touches service Filters; recommend it lands first (smaller, self-contained), but sequence per the PO's call. Code-depends on M4-S1's evaluate-and-route helper (see "Files to touch" above) — this is a real dependency, not just proto-file coordination; do not start the core triage logic until that helper exists or you've agreed its shape with the PO. Reconciles GH #176. Size: M.

## Context Story for milestone M4. **Read [`docs/STORY-WORKFLOW.md`](https://github.com/KofTwentyTwo/nuncio/blob/dev/docs/STORY-WORKFLOW.md) first** — it defines the process, gates, and patterns; this issue defines only the specific work. Per the original GH issue (#176) this reconciles: retiring the legacy JSON-RPC IPC removed a `filter.triage_keyset` handler that had real bulk-triage logic — a chunked scan of ALL stored messages, evaluating every persisted filter rule against each one, writing execution logs and outbox mutations for matches, and reporting batch progress as it went. That handler no longer exists anywhere in the tree (confirmed: no trace of it in `_reference/nuncio-mcp`/`_reference/nuncio-tui`, and `git log` shows no surviving commit content for it beyond the issue's own description) and the current `service Filters` (`CreateRule`/`ListRules`/`DeleteRule`/`ValidateRule`/`PreviewRule`) has no equivalent — today there is no way to apply the currently-active filter rule set retroactively against messages that were already synced before the rules existed (or before `M4-S1` wired live evaluation into sync at all). `crates/nuncio-store/src/db.rs`'s `get_message_chunk(last_id, limit)` (line 1366) already implements exactly the keyset-chunking primitive ("Fetch a page of messages using Keyset Chunking") this capability needs — it is currently unused in production, only exercised by its own unit tests. ## Goal A `Filters.Triage` gRPC RPC that scans every stored message in bounded chunks (via `DatabaseEngine::get_message_chunk`), evaluates the daemon's live `FilterEngine` against each one, applies the SAME action-routing semantics `M4-S1` wires into live sync (immediate `MARK READ`/`MARK UNREAD`, outbox-enqueued `MOVE`/`COPY`/`FLAG`/`UNFLAG`/`DELETE`/`FORWARD`/`CALL WEBHOOK`, an execution-log entry per match), and streams batch progress back to the caller as it works through the store — never a single blocking call with no feedback for what could be a very large mailbox. This must not fabricate progress or completion: counts reported on the stream must be the real, cumulative counts of what has actually been scanned/matched/mutated so far. ## Files to touch / patterns to mirror - `crates/nuncio-proto/proto/nuncio/v1/nuncio.proto` — extend `service Filters` (line 437) with a server-streaming RPC: `rpc Triage(TriageRequest) returns (stream TriageProgress);` — the ONLY server-streaming RPC pattern already in the proto is `System.Subscribe` (`rpc Subscribe(SubscribeRequest) returns (stream Event);`, line 24) — mirror its `stream` usage. `TriageRequest { optional string rule_id = 1; uint32 chunk_size = 2; }` (`rule_id` unset = evaluate against ALL active rules, matching `FilterEngine::evaluate`'s normal behaviour; `chunk_size` of `0` should mean "use a sane server-side default", e.g. 100). `TriageProgress { uint64 scanned_count = 1; uint64 matched_count = 2; uint64 actions_applied_count = 3; string last_message_id = 4; bool done = 5; }` — one message emitted after each chunk is processed, with cumulative (not per-chunk) counts, and a final message with `done = true`. - `crates/nunciod/src/grpc.rs` — `FiltersGrpcService` (struct at line 756): add a `triage` method matching tonic's server-streaming signature. `SystemGrpcService::subscribe` (around line 171-190) is the only existing server-streaming RPC in this file, but its mechanism does not fit here: it maps an ALREADY-LIVE stream (`self.event_bus.subscribe_events()`, a `tokio::sync::broadcast` receiver wrapped in `BroadcastStream`) — it has nothing to actively compute. `Triage` instead needs to actively drive a chunked DB scan and push results as it goes, which is a genuinely new pattern in this codebase: spawn a `tokio::task` that owns the scan loop, feed a `tokio::sync::mpsc::channel::<Result<TriageProgress, Status>>`, and return `Response::new(Box::pin(ReceiverStream::new(rx)))` as `Self::TriageStream` (associated type `Pin<Box<dyn Stream<Item = Result<TriageProgress, Status>> + Send + 'static>>`, matching `SubscribeStream`'s shape). Only the trait-level shape (associated type + `Response<Self::TriageStream>`) is worth copying from `subscribe`; the internal mechanism is different by necessity. Loop inside the spawned task: `db.get_message_chunk(last_id, chunk_size)` → for each `Email`, evaluate + route actions (see next bullet) → after the chunk, send a `TriageProgress` on the mpsc sender (stop looping if the receiver was dropped, i.e. the caller disconnected) → advance `last_id` to the chunk's last message id → stop when a chunk returns fewer than `chunk_size` rows and send a final `done: true` message. - **Depends on `M4-S1`'s evaluate-and-route helper.** `M4-S1` adds a reusable "evaluate `FilterEngine` against one `Email` and apply/enqueue its matched actions + write an execution log" helper (as part of wiring live sync) — call that SAME helper here per message rather than re-implementing action-routing a second time. If `M4-S1` has not merged yet when you start this story, coordinate with the PO on its exact function signature/location (likely `crates/nunciod/src/sync.rs` or a new `crates/nunciod/src/filter_apply.rs`) so both stories converge on one implementation instead of two independently-evolving copies of the same logic. - `crates/nuncio-proto/src/client.rs` — no new `connect_*` function needed (`connect_filters`/`AuthenticatedFiltersClient` already cover the whole `Filters` service); the new `triage` method returns a `tonic::Streaming<TriageProgress>` the caller reads via repeated `.message().await` calls. - `crates/nuncio-cli/src/args.rs` — add a `FilterSubcommand::Triage { rule_id: Option<String>, chunk_size: Option<u32> }` variant, mirroring the existing subcommand style (lines 379-451). - `crates/nuncio-cli/src/runner.rs` — add a `handle_filter_triage` that connects via `self.connect_filters_client()`, calls `.triage(...)`, and loops reading the returned stream, printing/accumulating progress as it goes (this will be the CLI's first real streaming-RPC consumer — `System.Subscribe` streaming is currently only exercised by `nunciod`'s own tests, per the comment at `runner.rs` line 1878, not by the CLI itself; you are not extending that path, just establishing your own for `Triage`). - `crates/nunciod/tests/grpc_auth_test.rs` — add an unauthenticated-rejection case for `Triage` (mirroring `rejects_calls_without_a_bearer_token`); `Filters` is already mounted behind its own `BearerAuthInterceptor`, so this is a regression assertion, not new mounting code. ## Implementation steps 1. Confirm with the PO whether `M4-S1` has merged; if so, import and reuse its evaluate-and-route helper directly. If not, either wait or write the minimal version of that helper here and flag to the PO that `M4-S1`'s implementer should converge on it (do not silently let two independent copies of the same action-routing logic exist after both stories land). 2. Add `Triage`/`TriageRequest`/`TriageProgress` to `nuncio.proto`, following §5 of `STORY-WORKFLOW.md` for the golden regen. **Confirm with the PO that `M4-S3` (also `[PROTO]`, also touching `service Filters`) is not simultaneously in flight** before editing the proto — only one proto-touching story at a time. 3. Implement `FiltersGrpcService::triage`: read `System::subscribe` first (in this same file) for the trait-level shape (associated `*Stream` type, `Response<Self::TriageStream>`), then build the actual mpsc-channel-plus-spawned-task mechanism described above (`subscribe`'s own `BroadcastStream`-over-an-existing-channel approach does not apply here — there is no existing "triage progress" channel to subscribe to; you are actively producing one). 4. Inside the spawned worker: loop `db.get_message_chunk`, evaluate+route per message via the shared helper, accumulate cumulative counts, send a `TriageProgress` after each chunk, and a final `done: true` message when a chunk comes back short. 5. Add the CLI subcommand + streaming consumer. 6. Tests: an offline E2E (via `crates/nunciod/tests/`, mirroring `spine_e2e_test.rs`'s daemon-boot style) that seeds several stored messages (some matching an active rule, some not) directly through `DatabaseEngine::save_email`, calls `Triage`, drains the stream, and asserts: (a) the final cumulative `matched_count`/`actions_applied_count` are correct; (b) the matching messages show the expected side effects (read-flag flip / queued outbox row, same assertions style as `M4-S1`'s tests); (c) chunking actually occurred across multiple `TriageProgress` messages for a store larger than one `chunk_size` (prove the streaming/chunking is real, not a single batch dressed up as a stream). 7. Add the unauthenticated-rejection regression test. 8. Run `cargo verify`; regenerate the contract golden if the descriptor test fails. ## Acceptance criteria - [ ] New RPC `Triage(TriageRequest) returns (stream TriageProgress)` added to `service Filters`; new messages `TriageRequest`/`TriageProgress` added, additive within `nuncio.v1`. - [ ] Contract golden regenerated; `nuncio-proto`'s `contract_stability::descriptor_matches_committed_golden` test passes. - [ ] `Triage` scans the WHOLE store via real keyset chunking (`DatabaseEngine::get_message_chunk`), not a single unbounded query. - [ ] Matched messages get the same real side effects `M4-S1` defines for live sync (immediate local action / enqueued outbox mutation / execution log entry) — reusing that story's helper, not a second independent implementation. - [ ] Streamed `TriageProgress` messages carry real, cumulative, non-fabricated counts; a `done: true` final message is sent when the scan completes. - [ ] E2E proves multi-chunk streaming actually happens for a store bigger than one chunk (not a single disguised batch response). - [ ] An unauthenticated call to `Triage` is rejected with `Code::Unauthenticated`. - [ ] CLI `filter triage` command consumes the stream and reports progress. - [ ] `cargo verify` green (fmt + clippy `-D warnings` all-targets + test-all), 0 ignored. ## Definition of done & process Follow [`docs/STORY-WORKFLOW.md`](https://github.com/KofTwentyTwo/nuncio/blob/dev/docs/STORY-WORKFLOW.md): no fabricated data; comment convention (intent-only, no story/GH/phase refs in `.rs`/`.proto`); mock-only tests. Branch `story/m4-filters-triage` off `dev`; PR to `dev` with the gate output pasted as evidence; the PO reviews before the issue is closed. ## Dependencies / notes **[PROTO]: confirm with the PO that no other proto-touching story is in flight before starting** — `M4-S3` also touches `service Filters`; recommend it lands first (smaller, self-contained), but sequence per the PO's call. **Code-depends on `M4-S1`'s evaluate-and-route helper** (see "Files to touch" above) — this is a real dependency, not just proto-file coordination; do not start the core `triage` logic until that helper exists or you've agreed its shape with the PO. Reconciles GH #176. Size: M.

KofTwentyTwo commented

2026-07-30 00:54:32 +00:00

(Migrated from github.com)

Delivered via PR #226 (squash-merged to dev). Filters.Triage server-streaming RPC scans the whole store via real keyset chunking (get_message_chunk), evaluates the live FilterEngine per message via the shared apply_filter_actions helper (same side effects as live sync), and streams real cumulative TriageProgress with a final done:true. rule_id accepted but documented as not-yet-honored (no silent no-op). Opus adversarial review verified loop termination, empty-store, chunk_size==0, honest error path, disconnect handling, no panic in the worker; multi-chunk E2E proves real streaming. Reconciles #176. Closes out milestone M4.

Sign in to join this conversation.

1 Participants

Notifications

Due Date

No due date set.

Dependencies

No dependencies set.

Reference: KofTwentyTwo/nuncio#193