fix(pipeline): stop incremental context runs gutting updated sources

In default new-content-only mode, add_context_to_nodes derived
updated_sources from the new docs and removed EVERY existing pkl node
whose source was in that set, re-adding only the new docs' nodes: a
source with 1,000 indexed pages and 1 new page kept context for 1 page.
create_vector_stores then deleted the other 999 pages' embeddings as
stale and the gutted bundle uploaded to HF. The next run regenerated
the 999 at Gemini cost while dropping the 1 (never converging).

- Replace the untestable inline python -c heredoc with an in-process,
unit-tested merge: merge_contextual_nodes keeps every existing node
except those whose doc_id is being (re)processed this run; nodes with
undeterminable doc_ids are kept. Removal by source no longer happens.
- add_course_workflow now shares the same function instead of carrying
a byte-for-byte duplicate of the broken block.
- Close the change-detection gap the doc_id-membership filter left:
content edits keep a stable doc_id, so edited docs were never
re-contextualized. Nodes now record the source row's content_hash
(excluded from embedded text), and the incremental filter selects new
docs plus docs whose stored hash differs. Legacy unhashed nodes are
treated as unchanged (no surprise Gemini reprocess); one
--process-all-context run baselines hashes corpus-wide.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

data/scraping_scripts/add_context_to_nodes.py

@@ -36,6 +36,13 @@ CONTEXT_RETRY_ATTEMPTS = int(os.getenv("GEMINI_CONTEXT_RETRY_ATTEMPTS", "8"))
 DEFAULT_SEMAPHORE_LIMIT = int(os.getenv("GEMINI_CONTEXT_CONCURRENCY", "50"))
 MAX_DOCUMENT_TOKENS = 120_000
 RETRYABLE_GENAI_STATUS_CODES = {408, 429, 500, 502, 503, 504}
+# Node-metadata key recording the source document's JSONL ``content_hash`` on
+# every contextual chunk. update_docs_workflow reads it back from the PKL to
+# detect docs whose content changed in place (doc_ids are path-based and stable
+# across edits, so id membership alone cannot catch this). update_docs_workflow
+# keeps its own copy of this constant so it does not have to import this
+# (Gemini/tiktoken/llama_index-heavy) module; tests assert the two stay in sync.
+DOC_CONTENT_HASH_METADATA_KEY = "doc_content_hash"
 _genai_client: genai.Client | None = None
 _token_encoding = tiktoken.get_encoding("cl100k_base")
 
@@ -83,28 +90,36 @@ def create_docs(input_file: str) -> List[Document]:
         documents: list[Document] = []
         for line in f:
             data = json.loads(line)
+            metadata = {
+                "url": data["url"],
+                "title": data["name"],
+                "tokens": data["tokens"],
+                "retrieve_doc": data["retrieve_doc"],
+                "source": data["source"],
+            }
+            # Rows written by process_md_files carry a doc-level content hash;
+            # older JSONLs may not, so only record it when present.
+            content_hash = data.get("content_hash")
+            if content_hash:
+                metadata["content_hash"] = content_hash
             documents.append(
                 Document(
                     doc_id=data["doc_id"],
                     text=data["content"],
-                    metadata={  # type: ignore
-                        "url": data["url"],
-                        "title": data["name"],
-                        "tokens": data["tokens"],
-                        "retrieve_doc": data["retrieve_doc"],
-                        "source": data["source"],
-                    },
+                    metadata=metadata,  # type: ignore
                     excluded_llm_metadata_keys=[
                         "title",
                         "tokens",
                         "retrieve_doc",
                         "source",
+                        "content_hash",
                     ],
                     excluded_embed_metadata_keys=[
                         "url",
                         "tokens",
                         "retrieve_doc",
                         "source",
+                        "content_hash",
                     ],
                 )
             )
@@ -242,7 +257,7 @@ Return a title for the document and the succinct context.
 
 
 def document_to_row(document: Document) -> dict:
-    return {
+    row = {
         "doc_id": document.doc_id,
         "content": document.get_content(),
         "name": document.metadata["title"],
@@ -251,6 +266,10 @@ def document_to_row(document: Document) -> dict:
         "retrieve_doc": document.metadata["retrieve_doc"],
         "tokens": document.metadata["tokens"],
     }
+    content_hash = document.metadata.get("content_hash")
+    if content_hash:
+        row["content_hash"] = content_hash
+    return row
 
 
 async def process_chunk(
@@ -278,6 +297,15 @@ async def process_chunk(
     context = await situate_context(doc.get_content(), chunk_record.text)
     metadata = dict(chunk_record.metadata)
     metadata["raw_text"] = chunk_record.text
+    # Stamp the document-level content hash on every chunk so incremental
+    # workflow runs can detect in-place content changes
+    # (update_docs_workflow.select_docs_to_process). Every contextualization
+    # path — full rebuild (main), incremental (update_docs_workflow), and
+    # create_vector_stores' non-pickle path — funnels through process_chunk,
+    # so this covers all of them. Absent for legacy rows without a hash.
+    doc_content_hash = doc.metadata.get("content_hash")
+    if doc_content_hash:
+        metadata[DOC_CONTENT_HASH_METADATA_KEY] = doc_content_hash
     contextual_text = (
         f"{format_chunk_for_retrieval(chunk_record.text, metadata)}"
         f"\n\nContext: {context}"

data/scraping_scripts/add_course_workflow.py

@@ -37,7 +37,7 @@ import os
 import pickle
 import subprocess
 import sys
-from typing import Dict, List, Set
+from typing import Dict, List
 
 from dotenv import load_dotenv
 from huggingface_hub import hf_hub_download
@@ -52,7 +52,7 @@ from data.scraping_scripts.source_registry import (
     required_data_files,
     source_output_files,
 )
-from app.chroma_rag import get_chunk_record_doc_id
+from data.scraping_scripts.update_docs_workflow import add_context_to_nodes
 
 # Load environment variables from .env file
 load_dotenv()
@@ -144,14 +144,6 @@ def load_jsonl(file_path: str) -> List[Dict]:
     return data
 
 
-def save_jsonl(data: List[Dict], file_path: str) -> None:
-    """Save data to a JSONL file."""
-    with open(file_path, "w", encoding="utf-8") as f:
-        for item in data:
-            json.dump(item, f, ensure_ascii=False)
-            f.write("\n")
-
-
 def process_markdown_files(course_name: str) -> str:
     """Process markdown files for a specific course. Returns path to output JSONL."""
     logger.info(f"Processing markdown files for course: {course_name}")
@@ -255,118 +247,9 @@ def purge_sources_from_pkl(sources_to_purge: List[str]) -> None:
     )
 
 
-def get_processed_doc_ids() -> Set[str]:
-    """Get set of doc_ids that have already been processed with context."""
-    if not os.path.exists("data/all_sources_contextual_nodes.pkl"):
-        return set()
-
-    try:
-        with open("data/all_sources_contextual_nodes.pkl", "rb") as f:
-            nodes = pickle.load(f)
-            return {get_chunk_record_doc_id(node) for node in nodes}
-    except Exception as e:
-        logger.error(f"Error loading processed doc_ids: {e}")
-        return set()
-
-
-def add_context_to_nodes(new_only: bool = False) -> None:
-    """Add context to document nodes, optionally processing only new content."""
-    logger.info("Adding context to document nodes")
-
-    if new_only:
-        # Load all documents
-        all_docs = load_jsonl("data/all_sources_data.jsonl")
-        processed_ids = get_processed_doc_ids()
-
-        # Filter for unprocessed documents
-        new_docs = [doc for doc in all_docs if doc["doc_id"] not in processed_ids]
-
-        if not new_docs:
-            logger.info("No new documents to process")
-            return
-
-        # Save temporary JSONL with only new documents
-        temp_file = "data/new_docs_temp.jsonl"
-        save_jsonl(new_docs, temp_file)
-
-        # Temporarily modify the add_context_to_nodes.py script to use the temp file
-        cmd = [
-            sys.executable,
-            "-c",
-            f"""
-import asyncio
-import os
-import pickle
-import json
-from data.scraping_scripts.add_context_to_nodes import create_docs, process
-from app.chroma_rag import get_chunk_record_source
-
-async def main():
-    # First, get the list of sources being updated from the temp file
-    updated_sources = set()
-    with open("{temp_file}", "r") as f:
-        for line in f:
-            data = json.loads(line)
-            updated_sources.add(data["source"])
-    
-    print(f"Updating nodes for sources: {{updated_sources}}")
-    
-    # Process new documents
-    documents = create_docs("{temp_file}")
-    enhanced_nodes = await process(documents)
-    print(f"Generated context for {{len(enhanced_nodes)}} new nodes")
-    
-    # Load existing nodes if they exist
-    existing_nodes = []
-    if os.path.exists("data/all_sources_contextual_nodes.pkl"):
-        with open("data/all_sources_contextual_nodes.pkl", "rb") as f:
-            existing_nodes = pickle.load(f)
-        
-        # Filter out existing nodes for sources we're updating
-        filtered_nodes = []
-        removed_count = 0
-        
-        for node in existing_nodes:
-            try:
-                source = get_chunk_record_source(node)
-                if source not in updated_sources:
-                    filtered_nodes.append(node)
-                else:
-                    removed_count += 1
-            except Exception:
-                # Keep nodes where we can't determine the source
-                filtered_nodes.append(node)
-        
-        print(f"Removed {{removed_count}} existing nodes for updated sources")
-        existing_nodes = filtered_nodes
-    
-    # Combine filtered existing nodes with new nodes
-    all_nodes = existing_nodes + enhanced_nodes
-    
-    # Save all nodes
-    with open("data/all_sources_contextual_nodes.pkl", "wb") as f:
-        pickle.dump(all_nodes, f)
-    
-    print(f"Total nodes in updated file: {{len(all_nodes)}}")
-
-asyncio.run(main())
-            """,
-        ]
-    else:
-        # Process all documents
-        cmd = [sys.executable, "-m", "data.scraping_scripts.add_context_to_nodes"]
-
-    result = subprocess.run(cmd)
-
-    if result.returncode != 0:
-        logger.error("Error adding context to nodes - check output above")
-        sys.exit(1)
-
-    logger.info("Successfully added context to nodes")
-
-    # Clean up temp file if it exists
-    if new_only and os.path.exists("data/new_docs_temp.jsonl"):
-        os.remove("data/new_docs_temp.jsonl")
+# The context-addition step (incl. the incremental merge that preserves
+# existing contextual nodes) is shared with the docs workflow:
+# data.scraping_scripts.update_docs_workflow.add_context_to_nodes.
 
 
 def create_vector_stores() -> None:

data/scraping_scripts/update_docs_workflow.py

@@ -18,7 +18,7 @@ Usage:
     Additional flags to run specific steps (if you want to restart from a specific point):
     --skip-download         Skip the GitHub download step
     --skip-process          Skip the markdown processing step
-    --new-context-only      Only process new content when adding context
+    --process-all-context   Regenerate context for every doc (default: only new or changed content)
     --skip-context          Skip the context addition step entirely
     --skip-vectors          Skip vector store creation
     --skip-upload           Skip uploading to HuggingFace
@@ -47,7 +47,7 @@ from data.scraping_scripts.source_registry import (
     required_data_files,
     source_output_files,
 )
-from app.chroma_rag import get_chunk_record_doc_id
+from app.chroma_rag import get_chunk_record_doc_id, get_chunk_record_metadata
 
 # Load environment variables from .env file
 load_dotenv()
@@ -135,6 +135,14 @@ DOC_SOURCES = list(DOC_SOURCE_KEYS)
 GITHUB_SOURCES = list(GITHUB_SOURCE_KEYS)
 LLMS_TXT_SOURCES = list(LLMS_TXT_SOURCE_KEYS)
 
+# Node-metadata key stamped by add_context_to_nodes.process_chunk with the
+# source document's JSONL ``content_hash``. Deliberately duplicated here (not
+# imported) so this module keeps its light import footprint —
+# add_context_to_nodes pulls in Gemini/tiktoken/llama_index, which this
+# module's importers and tests should not need.
+# tests/test_incremental_context.py asserts the two constants stay in sync.
+DOC_CONTENT_HASH_METADATA_KEY = "doc_content_hash"
+
 
 def load_jsonl(file_path: str) -> List[Dict]:
     """Load data from a JSONL file."""
@@ -247,109 +255,225 @@ def process_markdown_files(sources: List[str]) -> None:
     logger.info("Successfully processed markdown files")
 
 
-def get_processed_doc_ids() -> Set[str]:
-    """Get set of doc_ids that have already been processed with context."""
+def build_doc_hash_map(nodes: List) -> Dict[str, str | None]:
+    """Map doc_id -> stored ``doc_content_hash`` for a list of contextual nodes.
+
+    Membership means the doc already has contextual nodes; the value is the
+    document-level content hash stamped at context-generation time, or ``None``
+    for legacy nodes written before hashes existed. If a doc has a mix of
+    hashed and unhashed nodes (shouldn't happen, but resolve it sanely), any
+    stored hash for the doc counts. Nodes whose doc_id can't be determined are
+    skipped.
+    """
+    doc_hashes: Dict[str, str | None] = {}
+    for node in nodes:
+        try:
+            doc_id = get_chunk_record_doc_id(node)
+        except Exception:
+            continue
+        try:
+            metadata = get_chunk_record_metadata(node)
+        except Exception:
+            metadata = {}
+        stored_hash = metadata.get(DOC_CONTENT_HASH_METADATA_KEY)
+        if doc_id not in doc_hashes:
+            doc_hashes[doc_id] = stored_hash
+        elif doc_hashes[doc_id] is None and stored_hash is not None:
+            doc_hashes[doc_id] = stored_hash
+    return doc_hashes
+
+
+def get_processed_doc_hashes() -> Dict[str, str | None]:
+    """Get doc_id -> stored content hash for docs already processed with context.
+
+    Key membership carries the old ``get_processed_doc_ids`` contract (the doc
+    has contextual nodes in the PKL); the value adds the stored
+    ``doc_content_hash`` (``None`` for legacy nodes), enabling changed-content
+    detection. See ``build_doc_hash_map`` for resolution rules.
+    """
     if not os.path.exists("data/all_sources_contextual_nodes.pkl"):
-        return set()
+        return {}
 
     try:
         with open("data/all_sources_contextual_nodes.pkl", "rb") as f:
             nodes = pickle.load(f)
-            return {get_chunk_record_doc_id(node) for node in nodes}
     except Exception as e:
-        logger.error(f"Error loading processed doc_ids: {e}")
-        return set()
+        logger.error(f"Error loading processed doc hashes: {e}")
+        return {}
+
+    return build_doc_hash_map(nodes)
+
+
+def select_docs_to_process(
+    all_docs: List[Dict],
+    stored_hashes: Dict[str, str | None],
+) -> tuple[List[Dict], Dict[str, int]]:
+    """Pick the docs needing (re)contextualization, with selection stats.
+
+    A doc is selected when:
+
+    - its doc_id has no contextual nodes yet (**new**), or
+    - its stored ``doc_content_hash`` differs from the JSONL row's current
+      ``content_hash`` (**changed** in place; doc_ids are path-based and stable
+      across content edits, so id membership alone can never catch this).
+
+    Docs whose pkl nodes lack the hash field (**legacy**, written before this
+    fix) are treated as UNCHANGED so shipping hash-forward detection does not
+    trigger a surprise full-corpus Gemini reprocess; a one-time
+    ``--process-all-context`` run baselines hashes for the whole corpus, after
+    which in-place edits are picked up automatically. Rows without a
+    ``content_hash`` (older JSONLs) are likewise treated as unchanged.
+
+    Returns ``(docs_to_process, stats)`` with stats keys ``new``, ``changed``,
+    and ``legacy_unhashed``.
+    """
+    docs_to_process: List[Dict] = []
+    stats = {"new": 0, "changed": 0, "legacy_unhashed": 0}
+
+    for doc in all_docs:
+        doc_id = doc["doc_id"]
+        if doc_id not in stored_hashes:
+            stats["new"] += 1
+            docs_to_process.append(doc)
+            continue
+
+        stored_hash = stored_hashes[doc_id]
+        if stored_hash is None:
+            stats["legacy_unhashed"] += 1
+            continue
+
+        row_hash = doc.get("content_hash")
+        if row_hash is not None and row_hash != stored_hash:
+            stats["changed"] += 1
+            docs_to_process.append(doc)
+
+    return docs_to_process, stats
+
+
+def merge_contextual_nodes(
+    existing_nodes: List,
+    new_nodes: List,
+    reprocessed_doc_ids: Set[str],
+) -> List:
+    """Merge freshly contextualized nodes into an existing contextual-node list.
+
+    Every existing node is preserved except those whose doc_id is in
+    ``reprocessed_doc_ids`` (the docs being (re)processed in this run); their
+    old nodes are superseded by ``new_nodes``, which are appended. Nodes whose
+    doc_id cannot be determined are kept. Nodes are never removed by source:
+    an incremental run that adds one new page to a source must keep every
+    other already-indexed page of that source intact.
+    """
+    merged: List = []
+    replaced_count = 0
+    unknown_count = 0
+
+    for node in existing_nodes:
+        try:
+            doc_id = get_chunk_record_doc_id(node)
+        except Exception:
+            # Keep nodes whose doc_id can't be determined
+            merged.append(node)
+            unknown_count += 1
+            continue
+
+        if doc_id in reprocessed_doc_ids:
+            replaced_count += 1
+        else:
+            merged.append(node)
+
+    logger.info(
+        "Merging contextual nodes: kept %s existing nodes "
+        "(%s with undetermined doc_id), replaced %s nodes for reprocessed "
+        "doc_ids, appended %s new nodes",
+        len(merged),
+        unknown_count,
+        replaced_count,
+        len(new_nodes),
+    )
+    return merged + list(new_nodes)
+
+
+def _add_context_for_new_docs(temp_file: str, reprocessed_doc_ids: Set[str]) -> None:
+    """Generate context for the docs in ``temp_file`` and merge into the PKL."""
+    # Imported lazily: pulls in Gemini/tiktoken/llama_index, which the rest of
+    # this module (and its importers/tests) should not need.
+    import asyncio
+
+    from data.scraping_scripts.add_context_to_nodes import create_docs, process
+
+    documents = create_docs(temp_file)
+    enhanced_nodes = asyncio.run(process(documents))
+    logger.info("Generated context for %s new nodes", len(enhanced_nodes))
+
+    pkl_path = "data/all_sources_contextual_nodes.pkl"
+    existing_nodes: List = []
+    if os.path.exists(pkl_path):
+        with open(pkl_path, "rb") as f:
+            existing_nodes = pickle.load(f)
+
+    all_nodes = merge_contextual_nodes(
+        existing_nodes, enhanced_nodes, reprocessed_doc_ids
+    )
+
+    with open(pkl_path, "wb") as f:
+        pickle.dump(all_nodes, f)
+
+    logger.info("Total nodes in updated file: %s", len(all_nodes))
 
 
 def add_context_to_nodes(new_only: bool = False) -> None:
-    """Add context to document nodes, optionally processing only new content."""
+    """Add context to document nodes, optionally only new/changed content."""
     logger.info("Adding context to document nodes")
 
     if new_only:
         # Load all documents
         all_docs = load_jsonl("data/all_sources_data.jsonl")
-        processed_ids = get_processed_doc_ids()
-
-        # Filter for unprocessed documents
-        new_docs = [doc for doc in all_docs if doc["doc_id"] not in processed_ids]
+        stored_hashes = get_processed_doc_hashes()
+
+        # Select docs with no contextual nodes yet (new) plus docs whose
+        # content hash differs from the one stamped in the pkl (changed).
+        docs_to_process, stats = select_docs_to_process(all_docs, stored_hashes)
+        logger.info(
+            "Context selection: %s new docs, %s changed docs, %s legacy docs "
+            "without a stored content hash (treated as unchanged; a one-time "
+            "--process-all-context run baselines hashes for the whole corpus)",
+            stats["new"],
+            stats["changed"],
+            stats["legacy_unhashed"],
+        )
 
-        if not new_docs:
-            logger.info("No new documents to process")
+        if not docs_to_process:
+            logger.info("No new or changed documents to process")
             return
 
-        # Save temporary JSONL with only new documents
+        # Save temporary JSONL with only the documents to (re)process
         temp_file = "data/new_docs_temp.jsonl"
-        save_jsonl(new_docs, temp_file)
-
-        # Temporarily modify the add_context_to_nodes.py script to use the temp file
-        cmd = [
-            sys.executable,
-            "-c",
-            f"""
-import asyncio
-import os
-import pickle
-import json
-from data.scraping_scripts.add_context_to_nodes import create_docs, process
-from app.chroma_rag import get_chunk_record_source
+        save_jsonl(docs_to_process, temp_file)
 
-async def main():
-    # First, get the list of sources being updated from the temp file
-    updated_sources = set()
-    with open("{temp_file}", "r") as f:
-        for line in f:
-            data = json.loads(line)
-            updated_sources.add(data["source"])
-    
-    print(f"Updating nodes for sources: {{updated_sources}}")
-    
-    # Process new documents
-    documents = create_docs("{temp_file}")
-    enhanced_nodes = await process(documents)
-    print(f"Generated context for {{len(enhanced_nodes)}} new nodes")
-    
-    # Load existing nodes if they exist
-    existing_nodes = []
-    if os.path.exists("data/all_sources_contextual_nodes.pkl"):
-        with open("data/all_sources_contextual_nodes.pkl", "rb") as f:
-            existing_nodes = pickle.load(f)
-        
-        # Filter out existing nodes for sources we're updating
-        filtered_nodes = []
-        removed_count = 0
-        
-        for node in existing_nodes:
-            try:
-                source = get_chunk_record_source(node)
-                if source not in updated_sources:
-                    filtered_nodes.append(node)
-                else:
-                    removed_count += 1
-            except Exception:
-                # Keep nodes where we can't determine the source
-                filtered_nodes.append(node)
-        
-        print(f"Removed {{removed_count}} existing nodes for updated sources")
-        existing_nodes = filtered_nodes
-    
-    # Combine filtered existing nodes with new nodes
-    all_nodes = existing_nodes + enhanced_nodes
-    
-    # Save all nodes
-    with open("data/all_sources_contextual_nodes.pkl", "wb") as f:
-        pickle.dump(all_nodes, f)
-    
-    print(f"Total nodes in updated file: {{len(all_nodes)}}")
+        try:
+            # merge_contextual_nodes replaces every existing node whose doc_id
+            # is in this set, so a changed doc's old nodes are dropped in
+            # favor of the freshly contextualized ones.
+            _add_context_for_new_docs(
+                temp_file, {doc["doc_id"] for doc in docs_to_process}
+            )
+        except Exception:
+            logger.exception("Error adding context to nodes")
+            sys.exit(1)
 
-asyncio.run(main())
-            """,
-        ]
-    else:
-        # Process all documents
-        logger.info("Adding context to all nodes")
-        cmd = [sys.executable, "-m", "data.scraping_scripts.add_context_to_nodes"]
+        logger.info("Successfully added context to nodes")
 
-    result = subprocess.run(cmd)
+        # Clean up temp file (kept on failure to help debugging)
+        if os.path.exists(temp_file):
+            os.remove(temp_file)
+        return
+
+    # Process all documents
+    logger.info("Adding context to all nodes")
+    result = subprocess.run(
+        [sys.executable, "-m", "data.scraping_scripts.add_context_to_nodes"]
+    )
 
     if result.returncode != 0:
         logger.error("Error adding context to nodes - check output above")
@@ -357,10 +481,6 @@ asyncio.run(main())
 
     logger.info("Successfully added context to nodes")
 
-    # Clean up temp file if it exists
-    if new_only and os.path.exists("data/new_docs_temp.jsonl"):
-        os.remove("data/new_docs_temp.jsonl")
-
 
 def create_vector_stores() -> None:
     """Create vector stores from processed documents."""
@@ -443,7 +563,11 @@ def main():
     parser.add_argument(
         "--process-all-context",
         action="store_true",
-        help="Process all content when adding context (default: only process new content)",
+        help=(
+            "Process all content when adding context (default: only process "
+            "new or changed content; also baselines doc content hashes for "
+            "legacy nodes written before hashes were stamped)"
+        ),
     )
     parser.add_argument(
         "--skip-context",

tests/test_incremental_context.py

@@ -0,0 +1,304 @@
+"""Tests for the incremental contextual-node selection/merge in the workflows.
+
+Regression coverage for two data bugs in new-content-only mode:
+
+1. Data destruction: the workflows used to drop every existing node whose
+   *source* had any new doc, gutting incrementally updated sources. The merge
+   must instead preserve all existing nodes except those for doc_ids being
+   (re)processed in the run.
+2. Stale context: doc_ids are path-based and stable across content edits, so a
+   membership-only reprocess filter never re-contextualized a doc whose
+   content changed in place — its pkl nodes kept the old text forever. The
+   selection must compare the row's ``content_hash`` against the
+   ``doc_content_hash`` stamped in node metadata, while treating legacy nodes
+   without the hash as unchanged (no surprise full-corpus Gemini reprocess).
+"""
+
+import asyncio
+import json
+from unittest.mock import patch
+
+from app.chroma_rag import ChunkRecord
+from data.scraping_scripts.update_docs_workflow import (
+    DOC_CONTENT_HASH_METADATA_KEY,
+    build_doc_hash_map,
+    merge_contextual_nodes,
+    select_docs_to_process,
+)
+
+
+def make_node(
+    chunk_id: str, doc_id: str, source: str, content_hash: str | None = None
+) -> ChunkRecord:
+    metadata = {"doc_id": doc_id, "source": source}
+    if content_hash is not None:
+        metadata[DOC_CONTENT_HASH_METADATA_KEY] = content_hash
+    return ChunkRecord(
+        chunk_id=chunk_id,
+        doc_id=doc_id,
+        text=f"text for {chunk_id}",
+        metadata=metadata,
+    )
+
+
+def make_doc(doc_id: str, content_hash: str | None = "sha256:aaa") -> dict:
+    doc = {
+        "doc_id": doc_id,
+        "content": f"content for {doc_id}",
+        "name": doc_id,
+        "url": f"https://example.com/{doc_id}",
+        "source": doc_id.split(":")[0],
+        "retrieve_doc": False,
+        "tokens": 10,
+    }
+    if content_hash is not None:
+        doc["content_hash"] = content_hash
+    return doc
+
+
+def test_new_doc_in_existing_source_keeps_all_existing_nodes():
+    # A source with already-indexed docs gains one new doc: every existing
+    # node survives (same source or not) and the new doc's nodes are appended.
+    existing = [
+        make_node("t-p1-c1", "transformers:page-1", "transformers"),
+        make_node("t-p1-c2", "transformers:page-1", "transformers"),
+        make_node("t-p2-c1", "transformers:page-2", "transformers"),
+        make_node("l-p1-c1", "langchain:page-1", "langchain"),
+    ]
+    new = [
+        make_node("t-p3-c1", "transformers:page-3", "transformers"),
+        make_node("t-p3-c2", "transformers:page-3", "transformers"),
+    ]
+
+    merged = merge_contextual_nodes(existing, new, {"transformers:page-3"})
+
+    assert merged == existing + new
+
+
+def test_reprocessed_doc_id_replaces_only_its_own_nodes():
+    # If a doc_id in the new batch already has nodes in the pkl, those old
+    # nodes are replaced by the freshly generated ones; sibling docs from the
+    # same source are untouched.
+    stale = [
+        make_node("a-c1", "course:lesson-a", "course"),
+        make_node("a-c2", "course:lesson-a", "course"),
+    ]
+    sibling = make_node("b-c1", "course:lesson-b", "course")
+    regenerated = [make_node("a-c1-new", "course:lesson-a", "course")]
+
+    merged = merge_contextual_nodes(stale + [sibling], regenerated, {"course:lesson-a"})
+
+    assert merged == [sibling] + regenerated
+
+
+def test_nodes_with_undetermined_doc_id_are_kept():
+    class OpaqueNode:
+        """No doc_id, metadata, or source_node: doc_id lookup raises."""
+
+    opaque = OpaqueNode()
+    parseable = make_node("x-c1", "src:doc-x", "src")
+    new = [make_node("y-c1", "src:doc-y", "src")]
+
+    merged = merge_contextual_nodes([opaque, parseable], new, {"src:doc-y"})
+
+    assert merged == [opaque, parseable] + new
+
+
+def test_empty_existing_pkl_yields_only_new_nodes():
+    new = [make_node("n-c1", "src:doc-n", "src")]
+
+    assert merge_contextual_nodes([], new, {"src:doc-n"}) == new
+
+
+# --- Hash-forward selection: new + changed docs, legacy nodes untouched ---
+
+
+def test_changed_content_hash_is_reprocessed_and_old_nodes_replaced():
+    # A doc whose content changed in place (same path-stable doc_id, new
+    # content_hash) must be selected for reprocessing, and its old nodes must
+    # be replaced by the regenerated ones when merged back.
+    existing = [
+        make_node("d1-c1", "src:doc-1", "src", content_hash="sha256:old"),
+        make_node("d1-c2", "src:doc-1", "src", content_hash="sha256:old"),
+        make_node("d2-c1", "src:doc-2", "src", content_hash="sha256:same"),
+    ]
+    all_docs = [
+        make_doc("src:doc-1", "sha256:new"),
+        make_doc("src:doc-2", "sha256:same"),
+    ]
+
+    selected, stats = select_docs_to_process(all_docs, build_doc_hash_map(existing))
+
+    assert [doc["doc_id"] for doc in selected] == ["src:doc-1"]
+    assert stats == {"new": 0, "changed": 1, "legacy_unhashed": 0}
+
+    regenerated = [
+        make_node("d1-c1-new", "src:doc-1", "src", content_hash="sha256:new")
+    ]
+    merged = merge_contextual_nodes(
+        existing, regenerated, {doc["doc_id"] for doc in selected}
+    )
+
+    assert merged == [existing[2]] + regenerated
+
+
+def test_unchanged_content_hash_is_not_reprocessed():
+    existing = [make_node("d1-c1", "src:doc-1", "src", content_hash="sha256:same")]
+
+    selected, stats = select_docs_to_process(
+        [make_doc("src:doc-1", "sha256:same")], build_doc_hash_map(existing)
+    )
+
+    assert selected == []
+    assert stats == {"new": 0, "changed": 0, "legacy_unhashed": 0}
+
+
+def test_legacy_nodes_without_hash_are_treated_as_unchanged():
+    # Nodes written before hashes were stamped must NOT be reprocessed, even
+    # if the row's hash cannot be matched: --process-all-context is the
+    # explicit one-time baseline, not an implicit mass Gemini run.
+    existing = [make_node("d1-c1", "src:doc-1", "src")]
+    stored_hashes = build_doc_hash_map(existing)
+
+    assert stored_hashes == {"src:doc-1": None}
+
+    selected, stats = select_docs_to_process(
+        [make_doc("src:doc-1", "sha256:new")], stored_hashes
+    )
+
+    assert selected == []
+    assert stats == {"new": 0, "changed": 0, "legacy_unhashed": 1}
+
+
+def test_new_doc_is_still_detected():
+    existing = [make_node("d1-c1", "src:doc-1", "src", content_hash="sha256:a")]
+
+    selected, stats = select_docs_to_process(
+        [make_doc("src:doc-1", "sha256:a"), make_doc("src:doc-2", "sha256:b")],
+        build_doc_hash_map(existing),
+    )
+
+    assert [doc["doc_id"] for doc in selected] == ["src:doc-2"]
+    assert stats == {"new": 1, "changed": 0, "legacy_unhashed": 0}
+
+
+def test_row_without_content_hash_is_treated_as_unchanged():
+    # Older JSONL rows may lack content_hash entirely; with no basis for
+    # comparison the doc must not be reprocessed.
+    selected, stats = select_docs_to_process(
+        [make_doc("src:doc-1", content_hash=None)], {"src:doc-1": "sha256:a"}
+    )
+
+    assert selected == []
+    assert stats == {"new": 0, "changed": 0, "legacy_unhashed": 0}
+
+
+def test_mixed_hash_doc_uses_any_stored_hash():
+    # A doc with some hashed and some legacy-unhashed nodes (shouldn't exist,
+    # but must resolve sanely): any stored hash counts, in either node order.
+    hashed_first = [
+        make_node("d1-c1", "src:doc-1", "src", content_hash="sha256:a"),
+        make_node("d1-c2", "src:doc-1", "src"),
+    ]
+    unhashed_first = list(reversed(hashed_first))
+
+    assert build_doc_hash_map(hashed_first) == {"src:doc-1": "sha256:a"}
+    assert build_doc_hash_map(unhashed_first) == {"src:doc-1": "sha256:a"}
+
+    selected, stats = select_docs_to_process(
+        [make_doc("src:doc-1", "sha256:b")], build_doc_hash_map(unhashed_first)
+    )
+
+    assert [doc["doc_id"] for doc in selected] == ["src:doc-1"]
+    assert stats == {"new": 0, "changed": 1, "legacy_unhashed": 0}
+
+
+def test_nodes_with_undetermined_doc_id_are_skipped_in_hash_map():
+    class OpaqueNode:
+        """No doc_id, metadata, or source_node: doc_id lookup raises."""
+
+    assert build_doc_hash_map(
+        [OpaqueNode(), make_node("d1-c1", "src:doc-1", "src", content_hash="sha256:a")]
+    ) == {"src:doc-1": "sha256:a"}
+
+
+# --- Producer side: the context step must stamp the hash it reads back ---
+
+
+def test_metadata_key_constant_matches_producer():
+    # update_docs_workflow duplicates the key to stay import-light; the copies
+    # must never drift from the producer's.
+    from data.scraping_scripts.add_context_to_nodes import (
+        DOC_CONTENT_HASH_METADATA_KEY as producer_key,
+    )
+
+    assert producer_key == DOC_CONTENT_HASH_METADATA_KEY
+
+
+def test_process_stamps_doc_content_hash_on_every_chunk(tmp_path):
+    # create_docs -> process is the single funnel for the full-rebuild path,
+    # the incremental path, and create_vector_stores' non-pickle path; every
+    # resulting chunk must carry the row's content_hash. No Gemini call: the
+    # context generator is stubbed out.
+    from data.scraping_scripts.add_context_to_nodes import create_docs, process
+
+    row = make_doc("src:doc-1", "sha256:abc123")
+    row["content"] = "# Doc One\n\nSome content about incremental context."
+    jsonl_path = tmp_path / "docs.jsonl"
+    jsonl_path.write_text(json.dumps(row) + "\n", encoding="utf-8")
+
+    documents = create_docs(str(jsonl_path))
+    assert documents[0].metadata["content_hash"] == "sha256:abc123"
+
+    async def fake_situate_context(_doc: str, _chunk: str) -> str:
+        return "situated"
+
+    with patch(
+        "data.scraping_scripts.add_context_to_nodes.situate_context",
+        fake_situate_context,
+    ):
+        records = asyncio.run(process(documents, semaphore_limit=1))
+
+    assert records
+    assert all(
+        record.metadata[DOC_CONTENT_HASH_METADATA_KEY] == "sha256:abc123"
+        for record in records
+    )
+    # The selection round-trip closes: unchanged hash -> no reprocess,
+    # changed hash -> reprocess.
+    stored_hashes = build_doc_hash_map(records)
+    unchanged, _ = select_docs_to_process([row], stored_hashes)
+    changed, _ = select_docs_to_process(
+        [make_doc("src:doc-1", "sha256:def456")], stored_hashes
+    )
+    assert unchanged == []
+    assert [doc["doc_id"] for doc in changed] == ["src:doc-1"]
+
+
+def test_legacy_row_without_hash_produces_unhashed_nodes(tmp_path):
+    # Rows without content_hash (older JSONLs) must not stamp the field at
+    # all, so those docs land in the legacy-unhashed bucket downstream.
+    from data.scraping_scripts.add_context_to_nodes import create_docs, process
+
+    row = make_doc("src:doc-1", content_hash=None)
+    row["content"] = "# Doc One\n\nLegacy row without a content hash."
+    jsonl_path = tmp_path / "docs.jsonl"
+    jsonl_path.write_text(json.dumps(row) + "\n", encoding="utf-8")
+
+    documents = create_docs(str(jsonl_path))
+    assert "content_hash" not in documents[0].metadata
+
+    async def fake_situate_context(_doc: str, _chunk: str) -> str:
+        return "situated"
+
+    with patch(
+        "data.scraping_scripts.add_context_to_nodes.situate_context",
+        fake_situate_context,
+    ):
+        records = asyncio.run(process(documents, semaphore_limit=1))
+
+    assert records
+    assert all(
+        DOC_CONTENT_HASH_METADATA_KEY not in record.metadata for record in records
+    )
+    assert build_doc_hash_map(records) == {"src:doc-1": None}