update

backend/board_inventory.py

@@ -104,6 +104,20 @@ def _naming_schema(data, count: int) -> tuple[int | None, str | None, bool] | No
     return n, name, False
 
 
+def _sanierung_schema(data) -> tuple[str, str] | None:
+    """{"title": …, "description": …} → (title, description), beides getrimmt; Titel über
+    80 Zeichen verfällt (leerer String = kein Vorschlag, Feld bleibt wie es ist)."""
+    if not isinstance(data, dict):
+        return None
+    t = str(data.get("title") or "").strip()
+    d = str(data.get("description") or "").strip()
+    if not t and not d:
+        return None
+    if len(t) > 80:
+        t = ""
+    return t, d
+
+
 def _name_verankert(name: str, rows: list[dict], ctoks: set[str] | None) -> bool:
     """Abstraction guard: a free-formed title must be anchored — in the corpus (uni/projekt)
     or in the members' own words (thema). Unanchored names drift to textbook canon
@@ -470,6 +484,16 @@ def _hat_anker(title: str, ctoks: set[str]) -> bool:
     return False
 
 
+def _sanierung_noetig(p: dict, ctoks: set[str] | None) -> bool:
+    """QA-messbare Befund-Formen am Entstehungsort: Titel ohne Korpus-Anker (QA: fremd —
+    misst Token-Anker, nicht Semantik) oder leere Beschreibung (QA: hygiene). Gilt auch für
+    Singleton-Cluster, die das Naming sonst überspringen — Reader-Rohtitel gingen wörtlich
+    bis done_block durch (gemessen: 'k-Coloring' statt Korpus-Form 'k-Color')."""
+    if not (p.get("description") or "").strip():
+        return True
+    return bool(ctoks) and not _hat_anker(p.get("title", ""), ctoks)
+
+
 async def _anker_beleg(ctx: GenContext, flow: Flow, kandidaten: list[tuple[str, dict]]) -> set[str]:
     """Evidence judge for quorum titles WITHOUT any corpus anchor — two readers naming the
     same famous canon independently beat the quorum although the material never mentions it
@@ -703,9 +727,59 @@ async def _name_one(ctx: GenContext, flow: Flow, c):
             p["title"] = custom or w["title"]
             p["description"] = w.get("description") or p.get("description", "")
             await db.kanban_set_payload(topic, BOARD, cid, p)
+    await _saniere_one(ctx, flow, cid, p)
     await db.kanban_advance(topic, BOARD, cid, "naming_check")
 
 
+async def _saniere_one(ctx: GenContext, flow: Flow, cid: str, p: dict) -> None:
+    """Fremd-/Hygiene-Sanierung am Entstehungsort (nur Korpus-Quellen): Titel ohne
+    Korpus-Anker auf die Oberflächenform des Materials umschreiben, leere Beschreibung
+    belegt nachfassen. Läuft NACH der Member-Wahl auf dem finalen Payload — greift damit
+    auch für Singleton-Cluster und unverankerte Naming-Gewinner. Vorschläge werden
+    deterministisch validiert (_hat_anker, dieselbe Messlatte wie QA-fremd); fail-open:
+    unverankerter Vorschlag oder Judge-Ausfall lässt die Karte unverändert weiter."""
+    from qa import _distinctive
+    topic = flow.topic
+    folder = source_folder(topic)
+    if folder is None:
+        return  # thema: kein Korpus, keine Oberflächenform zum Verankern
+    ctoks = flow.state.get("korpus_tokens")
+    if ctoks is None:  # Resume: das Konsens-Gate dieses Flows lief nie
+        ctoks = flow.state["korpus_tokens"] = await asyncio.to_thread(_korpus_tokens, folder)
+    if not _sanierung_noetig(p, ctoks):
+        return
+    titel, beschr = p.get("title", ""), (p.get("description") or "").strip()
+    toks = _distinctive(titel) | _distinctive(beschr)
+    ev = _evidence_pack(folder, p.get("sources") or None, [" ".join(sorted(toks))],
+                        budget=4000) if toks else ""
+    if not ev:
+        return  # ohne Auszüge kein Anker-Rewrite — Anker-Beleg hat Echtheit schon entschieden
+    h = _h(titel, beschr, "sanierung")
+    path = flow.work_dir / f"sanierung-{cid}-{h}.json"
+    verdict = _sanierung_schema(_json_file(path))
+    if verdict is None:
+        status, verdict = await run_single_slot(
+            ctx, f"Sanierung {cid}", key=f"blocks-{topic}-sanierung-{cid}-{h}",
+            prompt=_prompt("Blocks-Sanierung", topic=topic, title=titel,
+                           description=beschr or "(leer)", excerpts=ev),
+            role="judge", capabilities="none",
+            payload=lambda result, p2=path: _sink_json(result, p2, _sanierung_schema),
+            timeout=_timeout("selection_mapping", 1))
+        if status != OK or verdict is None:
+            return
+    neu_titel, neu_beschr = verdict
+    changed = False
+    if (neu_titel and _norm_title(neu_titel) != _norm_title(titel)
+            and _hat_anker(neu_titel, ctoks)):
+        p["title"] = clean_title(neu_titel)
+        changed = True
+    if not beschr and neu_beschr:
+        p["description"] = neu_beschr
+        changed = True
+    if changed:
+        await db.kanban_set_payload(topic, BOARD, cid, p)
+
+
 async def _proc_naming_check(ctx: GenContext, flow: Flow, cards):
     results = await asyncio.gather(*[_namecheck_one(ctx, flow, c) for c in cards], return_exceptions=True)
     errs = [r for r in results if isinstance(r, Exception)]