feat(onboard): rejection ledger so re-runs don't re-classify rejected sentences (#801)

CHANGELOG.md

@@ -38,6 +38,8 @@ installable release; see the roadmap in [README.md](README.md).
 
 ### Fixed
 
+- **`aelf onboard <path>` no longer re-classifies previously rejected sentences** ([#801](https://github.com/robotrocketscience/aelfrice/issues/801)). Host `persist=False` verdicts were dropped on the floor: rejected sentences are never stored as beliefs, so the dedup-by-id filter in `start_onboard_session` had nothing to match them against, and every re-run re-emitted the same ~700 rejects to be re-classified (recorded repro: pass 1 → 2276 candidates / 723 rejected, pass 2 → 726 candidates ≈ pass-1 rejects + churn, pass 3 → 496 candidates, all costing one classification round trip per pass). New `onboard_rejections` table (per-project brain.db, keyed on the same SHA used by `beliefs.id`) is written by `accept_classifications` whenever the host returns `persist=False`; `start_onboard_session` and `check_onboard_candidates` skip ledger-resident ids the same way they already skip ids that resolved to a stored belief. `--emit-candidates` JSON gains `n_already_rejected`; `--check` reports the new bucket. `aelf onboard <path> --force` bypasses the ledger so the operator can re-roll a prior pass they think was too aggressive — already-present beliefs stay filtered under `--force` (it only re-opens the noise lane, not duplicate-stored content). A previously rejected sentence that is then accepted under `--force` (persist=True on the re-roll) is removed from the ledger so the post-accept state is a true no-op. Forward-compat: `CREATE TABLE IF NOT EXISTS` leaves prior schemas untouched and the table starts empty on existing stores; the slash command's `n_new == 0` short-circuit (`onboard.md` step 2) keeps working because rejected sentences move out of `n_new` into the new bucket.
+
 - **Retype legacy `belief_corroborations.belief_id` INTEGER → TEXT** ([#762](https://github.com/robotrocketscience/aelfrice/issues/762)). Some DBs created against the released v1.5.0 package shipped with `belief_id INTEGER` even though `beliefs.id` is `TEXT PRIMARY KEY`. Because all DDL in `store.py` uses `CREATE TABLE IF NOT EXISTS`, the corrected canonical shape never reached pre-existing stores; the FK silently missed on every hex-id insert and `aelf lock <statement>` on any text Jaccard-close enough to an existing belief raised `sqlite3.IntegrityError: FOREIGN KEY constraint failed` at `record_corroboration`. Symptoms were silent until the dedup path triggered, so users likely worked around by adding noise text rather than reporting. New `MemoryStore._maybe_retype_belief_corroborations_belief_id()` runs the SQLite-recommended column-retype recipe (CREATE new → COPY → DROP old → RENAME) under `PRAGMA foreign_keys=OFF`, gated on a `PRAGMA table_info` check that fires only when `belief_id` reports type `INTEGER`. Inserted in the migration ladder *before* `_maybe_consolidate_content_hash_duplicates` so its synthetic `consolidation_migration` rows write against the fixed shape. Idempotent via `SCHEMA_META_CORROBORATIONS_BELIEF_ID_RETYPED`; fresh stores stamp the marker without doing work. Hex belief ids stored under the broken INTEGER affinity round-trip as TEXT through the straight copy (digit-only literals would have coerced; hex strings are not valid INTEGER literals so the original values survived). Five new tests cover legacy-DB retype, idempotency across re-opens, fresh-store no-op, the `record_corroboration` success path on a previously-failing DB, and pre-existing row carry-over.
 
 - **Filter harness-wrapper prompts at the transcript-write path** ([#747](https://github.com/robotrocketscience/aelfrice/issues/747)). `transcript_logger.py` now drops prompts rejected by `noise_filter.is_transcript_noise` (synthetic `<task-notification>`, `<summary>Monitor`, `<tool-result>`, etc.) before append to `<git-common-dir>/aelfrice/transcripts/turns.jsonl`. Companion broadening: `_TRANSCRIPT_XML_PREFIXES` now matches `<summary>Monitor` in addition to `<summary>Background`, closing the gap that let Monitor stream-end renderings reach `ingest.py` and land as `agent_inferred` beliefs. Session-id detection and upstream fire counters are unaffected; only the JSONL append is gated. Cleanup of the existing noise rows in `turns.jsonl` and noise beliefs in the store is out of scope here (issue notes a follow-up).

src/aelfrice/classification.py

@@ -119,12 +119,16 @@ class StartOnboardResult:
     counts candidates that were dropped before the host saw them because
     a belief with the deterministic id already exists — re-running
     onboard on a tree the brain has already seen does not re-ask the
-    host to classify the same content.
+    host to classify the same content. `n_already_rejected` (#801)
+    counts candidates the host previously rejected with persist=False;
+    they sit in the rejection ledger and are bypassed unless the caller
+    passes `force=True`.
     """
 
     session_id: str
     sentences: list[OnboardSentence]
     n_already_present: int
+    n_already_rejected: int = 0
 
 
 @dataclass
@@ -151,11 +155,16 @@ class OnboardCheckResult:
     classifier, without writing an onboard_sessions row or inserting
     any beliefs. Lets callers decide whether re-onboard is worth the
     LLM/CPU cost before dispatching classification (#761).
+
+    `n_already_rejected` (#801) counts candidates currently in the
+    rejection ledger (host previously verdict persist=False); they are
+    excluded from `n_new` unless the caller passes `force=True`.
     """
 
     n_already_present: int
     n_new: int
     repo_path: str
+    n_already_rejected: int = 0
 
 
 @dataclass
@@ -195,16 +204,24 @@ def start_onboard_session(
     repo_path: Path,
     *,
     now: str | None = None,
+    force: bool = False,
 ) -> StartOnboardResult:
     """Run the three scanner extractors against `repo_path`, filter out
-    candidates whose deterministic belief id is already in the store,
-    persist the rest as a pending onboard_sessions row, and return the
-    payload the host should classify.
+    candidates whose deterministic belief id is already in the store
+    (or in the rejection ledger when `force=False`), persist the rest
+    as a pending onboard_sessions row, and return the payload the host
+    should classify.
 
     Idempotent: re-calling against the same tree returns a fresh
-    session_id whose `sentences` list excludes anything already present.
+    session_id whose `sentences` list excludes anything already present
+    *and* anything the host previously rejected with persist=False.
     The host can answer with an empty list of classifications and the
     session will close cleanly.
+
+    `force=True` (#801) bypasses the rejection ledger so previously
+    rejected candidates are re-emitted for the host to re-classify.
+    Already-present beliefs are still filtered — `force` only opts back
+    in to noise the host already saw, not to duplicate-stored content.
     """
     # Lazy import: scanner imports classification at module-load (it
     # calls `classify_sentence`); importing scanner at top-level here
@@ -223,13 +240,21 @@ def start_onboard_session(
         + extract_ast(repo_path)
     )
 
+    rejected_ids: set[str] = (
+        set() if force else store.list_onboard_rejection_ids()
+    )
+
     pending_sentences: list[OnboardSentence] = []
     n_already_present = 0
+    n_already_rejected = 0
     for c in candidates:
         bid = _derive_belief_id(c.text, c.source)
         if store.get_belief(bid) is not None:
             n_already_present += 1
             continue
+        if bid in rejected_ids:
+            n_already_rejected += 1
+            continue
         pending_sentences.append(
             OnboardSentence(
                 index=len(pending_sentences),
@@ -259,12 +284,15 @@ def start_onboard_session(
         session_id=session_id,
         sentences=pending_sentences,
         n_already_present=n_already_present,
+        n_already_rejected=n_already_rejected,
     )
 
 
 def check_onboard_candidates(
     store: "MemoryStore",
     repo_path: Path,
+    *,
+    force: bool = False,
 ) -> OnboardCheckResult:
     """Pre-scan a repo without persisting a session or inserting beliefs.
 
@@ -277,6 +305,9 @@ def check_onboard_candidates(
     handshake exposes via `--emit-candidates`, but at the human-facing
     `aelf onboard <path> --check` entry — letting callers see what a
     re-onboard would do before paying the classification cost (#761).
+
+    `n_already_rejected` (#801) reports candidates currently in the
+    rejection ledger; they are excluded from `n_new` unless `force=True`.
     """
     from aelfrice.scanner import (
         extract_ast,
@@ -290,19 +321,27 @@ def check_onboard_candidates(
         + extract_ast(repo_path)
     )
 
+    rejected_ids: set[str] = (
+        set() if force else store.list_onboard_rejection_ids()
+    )
+
     n_already_present = 0
+    n_already_rejected = 0
     n_new = 0
     for c in candidates:
         bid = _derive_belief_id(c.text, c.source)
         if store.get_belief(bid) is not None:
             n_already_present += 1
+        elif bid in rejected_ids:
+            n_already_rejected += 1
         else:
             n_new += 1
 
     return OnboardCheckResult(
         n_already_present=n_already_present,
         n_new=n_new,
         repo_path=str(repo_path),
+        n_already_rejected=n_already_rejected,
     )
 
 
@@ -370,8 +409,17 @@ def accept_classifications(
             skipped_unclassified += 1
             continue
         if not c.persist:
+            # #801: record the rejection so the next `aelf onboard` pass
+            # filters this candidate out instead of re-classifying it.
+            store.insert_onboard_rejection(
+                _derive_belief_id(text, source), text, source, timestamp,
+            )
             skipped_non_persisting += 1
             continue
+        # #801: a persisted sentence may have been previously rejected
+        # and surfaced again via --force. Drop the stale ledger entry so
+        # the ledger only carries currently-rejected candidates.
+        store.delete_onboard_rejection(_derive_belief_id(text, source))
         log_id = store.record_ingest(
             source_kind=INGEST_SOURCE_FILESYSTEM,
             source_path=source,

src/aelfrice/cli.py

@@ -345,14 +345,16 @@ def _cmd_onboard_emit_candidates(args: argparse.Namespace, out: object) -> int:
         )
         return 2
     repo_path = Path(args.path)
+    force = bool(getattr(args, "force", False))
     store = _open_store()
     try:
-        result = start_onboard_session(store, repo_path)
+        result = start_onboard_session(store, repo_path, force=force)
     finally:
         store.close()
     payload = {
         "session_id": result.session_id,
         "n_already_present": result.n_already_present,
+        "n_already_rejected": result.n_already_rejected,
         "sentences": [
             {"index": s.index, "text": s.text, "source": s.source}
             for s in result.sentences
@@ -476,19 +478,25 @@ def _cmd_onboard_check(args: argparse.Namespace, out: object) -> int:
         )
         return 2
     repo_path = Path(args.path)
+    force = bool(getattr(args, "force", False))
     store = _open_store()
     try:
-        result = check_onboard_candidates(store, repo_path)
+        result = check_onboard_candidates(store, repo_path, force=force)
     finally:
         store.close()
-    total = result.n_already_present + result.n_new
+    total = (
+        result.n_already_present + result.n_already_rejected + result.n_new
+    )
     pct_present = (
         (result.n_already_present * 100) // total if total > 0 else 0
     )
+    force_note = " (--force: ledger bypassed)" if force else ""
     print(
         f"path: {result.repo_path}\n"
         f"already present: {result.n_already_present} candidates "
         f"({pct_present}% of {total})\n"
+        f"already rejected: {result.n_already_rejected} candidates"
+        f"{force_note}\n"
         f"new since last onboard: {result.n_new} candidates\n"
         f"(read-only pre-scan; no beliefs inserted, no session persisted)",
         file=out,  # type: ignore[arg-type]
@@ -4899,6 +4907,18 @@ def build_parser(*, show_advanced: bool = False) -> argparse.ArgumentParser:
             "without contacting the network. Implies --llm-classify."
         ),
     )
+    p_onboard.add_argument(
+        "--force",
+        dest="force",
+        action="store_true",
+        help=(
+            "bypass the rejection ledger: re-emit candidates the host "
+            "previously classified persist=False so a fresh classification "
+            "pass can re-roll them. Has no effect on candidates already "
+            "stored as beliefs — those stay filtered. Apply on the "
+            "--emit-candidates or --check entry; default is ledger-on (#801)."
+        ),
+    )
     p_onboard.add_argument(
         "--revoke-consent",
         dest="revoke_consent",

src/aelfrice/store.py

@@ -463,6 +463,22 @@ def _check_insert_belief_authority() -> None:
     "CREATE INDEX IF NOT EXISTS idx_injection_events_pending "
     "ON injection_events(session_id, referenced) "
     "WHERE referenced IS NULL",
+    # #801: onboard rejection ledger. Records (text, source) pairs the
+    # host classifier rejected with persist=False so re-running
+    # `aelf onboard <path>` does not re-emit and re-classify the same
+    # noise on every pass. belief_id is the same SHA used by `beliefs.id`
+    # (sha256(source\\0text)[:16]); rejections and accepted beliefs share
+    # a key space and a rejection-then-accept moves the row from this
+    # table into `beliefs`. Forward-compat: empty on existing stores
+    # because IF NOT EXISTS leaves prior schemas untouched.
+    """
+    CREATE TABLE IF NOT EXISTS onboard_rejections (
+        belief_id   TEXT PRIMARY KEY,
+        text        TEXT NOT NULL,
+        source      TEXT NOT NULL,
+        rejected_at TEXT NOT NULL
+    )
+    """,
 )
 
 # Marker key for the entity-index one-shot backfill. Empty value =
@@ -3905,6 +3921,58 @@ def count_onboard_sessions(self, state: str | None = None) -> int:
         row = cur.fetchone()
         return int(row["n"]) if row else 0
 
+    def insert_onboard_rejection(
+        self, belief_id: str, text: str, source: str, rejected_at: str,
+    ) -> None:
+        """Record a host classifier `persist=False` verdict (#801).
+
+        Idempotent on `belief_id`: subsequent rejections of the same
+        `(text, source)` pair refresh `rejected_at` to the latest run.
+        """
+        self._conn.execute(
+            """
+            INSERT INTO onboard_rejections (belief_id, text, source, rejected_at)
+            VALUES (?, ?, ?, ?)
+            ON CONFLICT(belief_id) DO UPDATE SET rejected_at = excluded.rejected_at
+            """,
+            (belief_id, text, source, rejected_at),
+        )
+        self._conn.commit()
+
+    def delete_onboard_rejection(self, belief_id: str) -> bool:
+        """Drop a rejection ledger entry. Returns True if a row was
+        removed. Called when a previously-rejected sentence is accepted
+        on a `--force` re-run so the ledger only carries currently-
+        rejected entries.
+        """
+        cur = self._conn.execute(
+            "DELETE FROM onboard_rejections WHERE belief_id = ?",
+            (belief_id,),
+        )
+        self._conn.commit()
+        return cur.rowcount > 0
+
+    def is_onboard_rejected(self, belief_id: str) -> bool:
+        cur = self._conn.execute(
+            "SELECT 1 FROM onboard_rejections WHERE belief_id = ? LIMIT 1",
+            (belief_id,),
+        )
+        return cur.fetchone() is not None
+
+    def list_onboard_rejection_ids(self) -> set[str]:
+        """Set of all rejected belief_ids. Used to bulk-filter
+        candidates without one SELECT per row.
+        """
+        cur = self._conn.execute("SELECT belief_id FROM onboard_rejections")
+        return {row["belief_id"] for row in cur.fetchall()}
+
+    def count_onboard_rejections(self) -> int:
+        cur = self._conn.execute(
+            "SELECT COUNT(*) AS n FROM onboard_rejections"
+        )
+        row = cur.fetchone()
+        return int(row["n"]) if row else 0
+
     def list_pending_onboard_sessions(self) -> list[OnboardSession]:
         """All sessions in `pending` state, oldest first.