feat(assets): persisted hash modes with OFF→ON revalidation

Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>

app/assets/database/queries/asset_reference.py

@@ -907,7 +907,7 @@ class UnenrichedReferenceRow(NamedTuple):
     reference_id: str
     asset_id: str
     file_path: str
-    enrichment_level: int
+    hash_state: int
 
 
 def get_unenriched_references(
@@ -916,7 +916,7 @@ def get_unenriched_references(
     max_level: int = 0,
     limit: int = 1000,
 ) -> list[UnenrichedReferenceRow]:
-    """Get references that need enrichment (enrichment_level <= max_level).
+    """Get references that need enrichment (hash state <= max_level).
 
     Args:
         session: Database session
@@ -937,12 +937,12 @@ def get_unenriched_references(
             AssetReference.id,
             AssetReference.asset_id,
             AssetReference.file_path,
-            AssetReference.enrichment_level,
+        AssetReference.hash_state,
         )
         .where(AssetReference.file_path.isnot(None))
         .where(sa.or_(*conds))
         .where(AssetReference.is_missing == False)  # noqa: E712
-        .where(AssetReference.enrichment_level <= max_level)
+        .where(AssetReference.hash_state <= max_level)
         .order_by(AssetReference.id.asc())
         .limit(limit)
     )
@@ -953,13 +953,13 @@ def get_unenriched_references(
             reference_id=row[0],
             asset_id=row[1],
             file_path=row[2],
-            enrichment_level=row[3],
+            hash_state=row[3],
         )
         for row in rows
     ]
 
 
-def bulk_update_enrichment_level(
+def bulk_update_hash_state(
     session: Session,
     reference_ids: list[str],
     level: int,
@@ -973,7 +973,7 @@ def bulk_update_enrichment_level(
     result = session.execute(
         sa.update(AssetReference)
         .where(AssetReference.id.in_(reference_ids))
-        .values(enrichment_level=level)
+        .values(hash_state=level)
     )
     return result.rowcount
 

app/assets/lifecycle.py

@@ -15,6 +15,7 @@ from app.assets.services.lookup import is_temp_path
 from app.database.db import create_session, init_db
 
 _excluded_scan_roots: set[str] = set()
+_hash_mode_transition: str | None = None
 
 
 def get_excluded_scan_roots() -> frozenset[str]:
@@ -22,11 +23,20 @@ def get_excluded_scan_roots() -> frozenset[str]:
 
 
 def record_hash_mode_transition_intent() -> None:
-    """Compare stored vs runtime hash mode; record intent only (todo 21 enqueues work)."""
+    global _hash_mode_transition
+    from app.assets.services.hash_mode_state import record_transition_intent
+
+    with create_session() as session:
+        _hash_mode_transition = record_transition_intent(session)
+        session.commit()
 
 
 def enqueue_mode_transition_work() -> None:
-    """Enqueue hash transition work over surviving rows after temp wipe (todo 21)."""
+    from app.assets.services.hash_mode_state import enqueue_transition_work
+
+    with create_session() as session:
+        enqueue_transition_work(session, _hash_mode_transition)
+        session.commit()
 
 
 def init_db_and_state() -> None:

app/assets/scanner.py

@@ -8,7 +8,7 @@ import sqlalchemy as sa
 from sqlalchemy.orm import Session
 from app.assets import mode
 from app.assets.database.queries import (
-    bulk_update_enrichment_level,
+    bulk_update_hash_state,
     bulk_update_needs_verify,
     delete_orphaned_seed_asset,
     get_asset_by_hash,
@@ -514,7 +514,7 @@ def enrich_asset(
     elif mime_type:
         update_asset_hash_and_mime(session, asset_id, mime_type=mime_type)
 
-    bulk_update_enrichment_level(session, [reference_id], new_level)
+    bulk_update_hash_state(session, [reference_id], new_level)
     session.commit()
 
     return new_level
@@ -564,7 +564,7 @@ def enrich_assets_batch(
                     interrupt_check=interrupt_check,
                     hash_checkpoints=hash_checkpoints,
                 )
-                if new_level > row.enrichment_level:
+                if new_level > row.hash_state:
                     enriched += 1
                 else:
                     failed_ids.append(row.reference_id)

app/assets/services/hash_mode_state.py

@@ -0,0 +1,92 @@
+"""Hash-mode persistence and OFF→ON transition logic."""
+
+from __future__ import annotations
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.assets import mode as _mode
+from app.assets.database.models import AssetContent, AssetSystemState
+from app.assets.database.queries.records import mark_content_missing
+from app.assets.services.snapshot_hash import snapshot_hash
+
+_KEY = "hash_mode"
+_PENDING_QUEUE: list[str] = []
+
+
+def clear_transition_queue() -> None:
+    _PENDING_QUEUE.clear()
+
+
+def pending_transition_count() -> int:
+    return len(_PENDING_QUEUE)
+
+
+def read_stored_mode(session: Session) -> str | None:
+    row = session.get(AssetSystemState, _KEY)
+    return row.value if row else None
+
+
+def write_stored_mode(session: Session, value: str) -> None:
+    row = session.get(AssetSystemState, _KEY)
+    if row is None:
+        session.add(AssetSystemState(key=_KEY, value=value))
+    else:
+        row.value = value
+    session.flush()
+
+
+def record_transition_intent(session: Session) -> str | None:
+    stored = read_stored_mode(session)
+    runtime = "on" if _mode.hashing_enabled() else "off"
+    if stored is None:
+        write_stored_mode(session, runtime)
+        return None
+    if stored == "off" and runtime == "on":
+        return "off_to_on"
+    if stored == "on" and runtime == "off":
+        write_stored_mode(session, "off")
+        return "on_to_off"
+    return None
+
+
+def enqueue_transition_work(session: Session, transition: str | None) -> None:
+    if transition != "off_to_on":
+        return
+    rows = session.scalars(
+        select(AssetContent).where(AssetContent.is_missing.is_(False))
+    )
+    for row in rows:
+        if row.path not in _PENDING_QUEUE:
+            _PENDING_QUEUE.append(row.path)
+
+
+def drain_transition_queue(session: Session) -> None:
+    pending_count = len(_PENDING_QUEUE)
+    for _ in range(pending_count):
+        path = _PENDING_QUEUE.pop(0)
+        digest = snapshot_hash(path)
+        if digest is None:
+            _PENDING_QUEUE.append(path)
+            continue
+        content = session.scalars(
+            select(AssetContent).where(
+                AssetContent.path == path, AssetContent.is_missing.is_(False)
+            )
+        ).first()
+        if content is None:
+            continue
+        current_hash = f"blake3:{digest}"
+        if content.hash is None:
+            content.hash = current_hash
+        elif content.hash != current_hash:
+            mark_content_missing(session, content.id)
+            replacement = AssetContent(
+                path=path,
+                hash=current_hash,
+                size_bytes=content.size_bytes,
+                mtime_ns=content.mtime_ns,
+            )
+            session.add(replacement)
+    if not _PENDING_QUEUE:
+        write_stored_mode(session, "on")

app/assets/services/ingest.py

@@ -217,7 +217,7 @@ def ingest_existing_file(
             existing_ref.job_id = job_id
             existing_ref.is_missing = False
             existing_ref.updated_at = now
-            existing_ref.enrichment_level = 0
+            existing_ref.hash_state = 0
 
             asset = existing_ref.asset
             if asset:

tests-unit/assets_test/services/test_hash_mode_state.py

@@ -0,0 +1,48 @@
+from unittest.mock import patch
+
+from app.assets.database.models import AssetContent
+from app.assets.services.hash_mode_state import (
+    clear_transition_queue,
+    enqueue_transition_work,
+    pending_transition_count,
+    read_stored_mode,
+    record_transition_intent,
+    write_stored_mode,
+)
+
+
+def test_absent_row_off_mode_no_transition(session):
+    with patch("app.assets.services.hash_mode_state._mode.hashing_enabled", return_value=False):
+        assert record_transition_intent(session) is None
+    assert read_stored_mode(session) == "off"
+
+
+def test_off_to_on_enqueues_null_rows(session):
+    session.add(AssetContent(path="/tmp/null", hash=None))
+    session.add(AssetContent(path="/tmp/hashed", hash="blake3:abc"))
+    write_stored_mode(session, "off")
+    with patch("app.assets.services.hash_mode_state._mode.hashing_enabled", return_value=True):
+        transition = record_transition_intent(session)
+    enqueue_transition_work(session, transition)
+    assert transition == "off_to_on"
+    assert pending_transition_count() == 2
+    assert read_stored_mode(session) == "off"
+    clear_transition_queue()
+
+
+def test_on_to_off_freezes(session):
+    write_stored_mode(session, "on")
+    with patch("app.assets.services.hash_mode_state._mode.hashing_enabled", return_value=False):
+        transition = record_transition_intent(session)
+    assert transition == "on_to_off"
+    assert read_stored_mode(session) == "off"
+    assert pending_transition_count() == 0
+
+
+def test_mode_stays_off_during_transition(session):
+    session.add(AssetContent(path="/tmp/null", hash=None))
+    write_stored_mode(session, "off")
+    with patch("app.assets.services.hash_mode_state._mode.hashing_enabled", return_value=True):
+        transition = record_transition_intent(session)
+    enqueue_transition_work(session, transition)
+    assert read_stored_mode(session) == "off"