feat(ingest): add span re-anchoring and stale-span linter (Phase 5) (#450)

alembic/versions/0023_add_sourcespan_stale_column.py

@@ -0,0 +1,48 @@
+"""Add stale column to sourcespan table.
+
+Revision ID: 0023
+Revises: 0022
+Create Date: 2026-05-23
+
+Adds a boolean ``stale`` column to the ``sourcespan`` table so that spans
+whose content no longer matches after a source re-ingestion can be flagged.
+Claims referencing stale spans surface as linter warnings. See issue #450,
+Phase 5.
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from sqlalchemy import inspect as sa_inspect
+
+from alembic import op
+
+revision: str = "0023"
+down_revision: str = "0022"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa_inspect(conn)
+    existing = inspector.get_table_names()
+
+    if "sourcespan" in existing:
+        columns = [c["name"] for c in inspector.get_columns("sourcespan")]
+        if "stale" not in columns:
+            op.add_column(
+                "sourcespan",
+                sa.Column("stale", sa.Boolean(), nullable=False, server_default=sa.text("false")),
+            )
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa_inspect(conn)
+    existing = inspector.get_table_names()
+
+    if "sourcespan" in existing:
+        columns = [c["name"] for c in inspector.get_columns("sourcespan")]
+        if "stale" in columns:
+            op.drop_column("sourcespan", "stale")

src/wikimind/engine/linter/__init__.py

@@ -6,5 +6,6 @@
 from wikimind.engine.linter.contradictions import detect_contradictions
 from wikimind.engine.linter.orphans import detect_orphans
 from wikimind.engine.linter.runner import run_lint
+from wikimind.engine.linter.stale_spans import detect_stale_spans
 
-__all__ = ["detect_contradictions", "detect_orphans", "run_lint"]
+__all__ = ["detect_contradictions", "detect_orphans", "detect_stale_spans", "run_lint"]

src/wikimind/engine/linter/runner.py

@@ -26,6 +26,7 @@
 from wikimind.engine.backlink_enforcer import enforce_backlinks
 from wikimind.engine.linter.contradictions import detect_contradictions
 from wikimind.engine.linter.orphans import detect_orphans
+from wikimind.engine.linter.stale_spans import detect_stale_spans
 from wikimind.engine.linter.staleness import detect_stale_articles
 from wikimind.engine.llm_router import get_llm_router
 from wikimind.models import (
@@ -379,6 +380,10 @@ async def run_lint(
         stale_findings = await detect_stale_articles(session, settings, report.id, user_id=user_id)
         structurals.extend(stale_findings)
 
+        # Phase 5: Stale source-span detection (issue #450)
+        stale_span_findings = await detect_stale_spans(session, report.id, user_id=user_id)
+        structurals.extend(stale_span_findings)
+
         # Apply dismiss suppression
         await _apply_dismiss_suppression(session, contradictions, orphans, structurals)
 

src/wikimind/engine/linter/stale_spans.py

@@ -0,0 +1,122 @@
+"""Stale-span detection — surface articles with claims pointing to stale source spans.
+
+When a source is re-ingested and some paragraphs no longer match, the
+corresponding ``SourceSpan`` rows are marked ``stale=True``.  Claims that
+still reference those stale spans lose their citation anchor.  This check
+generates :class:`StructuralFinding` lint warnings so users know which
+articles need attention.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from typing import TYPE_CHECKING
+
+import structlog
+from sqlmodel import select
+
+from wikimind.models import (
+    Article,
+    CompiledClaim,
+    LintFindingKind,
+    LintSeverity,
+    SourceSpan,
+    StructuralFinding,
+)
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+log = structlog.get_logger()
+
+VIOLATION_TYPE = "stale_source_spans"
+
+
+def _content_hash(article_id: str) -> str:
+    """Compute a stable sha256 for cross-run dedup of stale-span findings."""
+    raw = f"{LintFindingKind.STRUCTURAL}|{article_id}|{VIOLATION_TYPE}"
+    return hashlib.sha256(raw.encode()).hexdigest()
+
+
+async def detect_stale_spans(
+    session: AsyncSession,
+    report_id: str,
+    user_id: str,
+) -> list[StructuralFinding]:
+    """Find articles with claims pointing to stale source spans.
+
+    For each article, loads its compiled claims and checks whether any
+    ``source_span_ids`` reference a span whose ``stale`` flag is True.
+
+    Args:
+        session: Async database session.
+        report_id: The parent LintReport ID.
+        user_id: User ID for data isolation.
+
+    Returns:
+        List of StructuralFinding instances for articles with stale span refs.
+    """
+    # Load all stale span IDs for this user in one query
+    stale_stmt = select(SourceSpan.id).where(
+        SourceSpan.user_id == user_id,
+        SourceSpan.stale.is_(True),  # type: ignore[attr-defined]
+    )
+    stale_result = await session.execute(stale_stmt)
+    stale_span_ids: set[str] = {row[0] for row in stale_result.all()}
+
+    if not stale_span_ids:
+        log.info("Stale-span detection: no stale spans found")
+        return []
+
+    # Load all claims for this user that have span references
+    claim_stmt = (
+        select(CompiledClaim.article_id, CompiledClaim.source_span_ids)
+        .where(CompiledClaim.user_id == user_id)
+        .where(CompiledClaim.source_span_ids != "[]")
+    )
+    claim_result = await session.execute(claim_stmt)
+    claim_rows = claim_result.all()
+
+    # Group stale-span-referencing claims by article
+    articles_with_stale: dict[str, int] = {}
+    for article_id, span_ids_json in claim_rows:
+        try:
+            span_ids = json.loads(span_ids_json)
+        except (json.JSONDecodeError, TypeError):
+            continue
+        stale_count = sum(1 for sid in span_ids if sid in stale_span_ids)
+        if stale_count > 0:
+            articles_with_stale[article_id] = articles_with_stale.get(article_id, 0) + stale_count
+
+    if not articles_with_stale:
+        log.info("Stale-span detection: no claims reference stale spans")
+        return []
+
+    # Look up article titles for readable descriptions
+    article_stmt = select(Article.id, Article.title).where(
+        Article.id.in_(list(articles_with_stale.keys())),  # type: ignore[attr-defined]
+    )
+    article_result = await session.execute(article_stmt)
+    article_titles: dict[str, str] = {row[0]: row[1] for row in article_result.all()}
+
+    findings: list[StructuralFinding] = []
+    for article_id, stale_count in articles_with_stale.items():
+        title = article_titles.get(article_id, article_id)
+        desc = f"Article '{title}' has {stale_count} claim(s) referencing stale source spans"
+        findings.append(
+            StructuralFinding(
+                report_id=report_id,
+                severity=LintSeverity.WARN,
+                description=desc,
+                content_hash=_content_hash(article_id),
+                article_id=article_id,
+                violation_type=VIOLATION_TYPE,
+                auto_repaired=False,
+                detail=f"stale_span_references={stale_count}",
+                user_id=user_id,
+            )
+        )
+
+    log.info("Stale-span detection complete", articles_affected=len(findings))
+    return findings

src/wikimind/ingest/spans.py

@@ -12,6 +12,7 @@
 from typing import TYPE_CHECKING
 
 import structlog
+from sqlmodel import select
 
 from wikimind.models.enums import LocatorKind
 from wikimind.models.tables.wiki import SourceSpan
@@ -245,13 +246,113 @@ async def persist_spans(
 ) -> None:
     """Persist a batch of SourceSpan instances to the database.
 
+    If the source already has spans in the database, delegates to
+    :func:`reanchor_spans` for fingerprint-based matching so that
+    existing span IDs (and thus claim references) are preserved.
+
     Args:
         spans: List of SourceSpan instances to save.
         session: Async database session.
     """
     if not spans:
         return
+
+    # Check if this source already has spans — if so, re-anchor
+    source_id = spans[0].source_id
+    existing_stmt = select(SourceSpan).where(SourceSpan.source_id == source_id)
+    existing_result = await session.execute(existing_stmt)
+    existing_spans = list(existing_result.scalars().all())
+
+    if existing_spans:
+        await reanchor_spans(source_id, spans, session, existing_spans=existing_spans)
+        return
+
     for span in spans:
         session.add(span)
     await session.flush()
-    log.info("Persisted source spans", count=len(spans), source_id=spans[0].source_id)
+    log.info("Persisted source spans", count=len(spans), source_id=source_id)
+
+
+# ---------------------------------------------------------------------------
+# Re-anchoring on source update
+# ---------------------------------------------------------------------------
+
+
+async def reanchor_spans(
+    source_id: str,
+    new_spans: list[SourceSpan],
+    session: AsyncSession,
+    *,
+    existing_spans: list[SourceSpan] | None = None,
+) -> list[SourceSpan]:
+    """Re-anchor existing spans after a source is re-ingested.
+
+    Matches old spans to new spans by fingerprint so that claim references
+    (via ``CompiledClaim.source_span_ids``) remain valid after content
+    updates.  Unmatched old spans are marked stale; genuinely new spans
+    are created normally.
+
+    Args:
+        source_id: The source whose spans are being refreshed.
+        new_spans: Freshly extracted spans from the updated content.
+        session: Async database session.
+        existing_spans: Pre-loaded existing spans to avoid a redundant
+            database query when the caller already has them.
+
+    Returns:
+        The final list of spans (updated + new) that were persisted.
+    """
+    if existing_spans is not None:
+        old_spans = existing_spans
+    else:
+        # Load existing spans for this source
+        stmt = select(SourceSpan).where(SourceSpan.source_id == source_id)
+        result = await session.execute(stmt)
+        old_spans = list(result.scalars().all())
+
+    if not old_spans:
+        # No existing spans — just persist the new ones directly
+        await persist_spans(new_spans, session)
+        return new_spans
+
+    # Build lookup from fingerprint -> old span (first match wins)
+    old_by_fingerprint: dict[str, SourceSpan] = {}
+    for span in old_spans:
+        if span.fingerprint not in old_by_fingerprint:
+            old_by_fingerprint[span.fingerprint] = span
+
+    matched_old_ids: set[str] = set()
+    final_spans: list[SourceSpan] = []
+
+    for new_span in new_spans:
+        old_span = old_by_fingerprint.get(new_span.fingerprint)
+        if old_span and old_span.id not in matched_old_ids:
+            # Matched: update locator to new position, keep same ID
+            old_span.locator = new_span.locator
+            old_span.text = new_span.text
+            old_span.stale = False
+            session.add(old_span)
+            matched_old_ids.add(old_span.id)
+            final_spans.append(old_span)
+        else:
+            # Genuinely new span — persist it
+            session.add(new_span)
+            final_spans.append(new_span)
+
+    # Mark unmatched old spans as stale
+    stale_count = 0
+    for old_span in old_spans:
+        if old_span.id not in matched_old_ids:
+            old_span.stale = True
+            session.add(old_span)
+            stale_count += 1
+
+    await session.flush()
+    log.info(
+        "Re-anchored source spans",
+        source_id=source_id,
+        matched=len(matched_old_ids),
+        new=len(final_spans) - len(matched_old_ids),
+        stale=stale_count,
+    )
+    return final_spans

src/wikimind/models/dto/wiki.py

@@ -405,6 +405,7 @@ class SourceSpanResponse(BaseModel):
     locator: dict
     text: str
     fingerprint: str
+    stale: bool = False
     created_at: datetime
 
 

src/wikimind/models/tables/wiki.py

@@ -28,6 +28,7 @@ class SourceSpan(SQLModel, table=True):
     locator: dict = Field(sa_column=Column(JSON, nullable=False))  # adapter-specific anchor
     text: str = Field(sa_type=Text)  # verbatim quoted text
     fingerprint: str = Field(index=True)  # SHA-256 of normalized text for re-anchoring
+    stale: bool = Field(default=False)  # True when source re-ingested and span no longer matches
     created_at: datetime = Field(default_factory=utcnow_naive)