"""Creating, listing and deciding proposals. The interesting parts are deduplication (a queue that shows the same suggestion five times is a queue nobody opens) and what "accept" actually does — which is delegated to the corpus adapter, because only it knows how to write to its own corpus. """ from __future__ import annotations import hashlib import json import logging import uuid from datetime import UTC, datetime from typing import Any from sqlalchemy import func, select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import ValidationError from app.core.knowledge_loop.constants import ( KINDS, MAX_PROPOSALS_PER_INGEST, OVERRIDABLE_PAYLOAD_KEYS, SOURCES, STATUS_ACCEPTED, STATUS_FAILED, STATUS_OPEN, STATUS_REJECTED, STATUS_SUPERSEDED, ) from app.core.knowledge_loop.models import KnowledgeProposal from app.core.knowledge_loop.registry import ProposalApplyError, adapter_for from app.core.txn import best_effort logger = logging.getLogger(__name__) def _naive_now() -> datetime: return datetime.now(UTC).replace(tzinfo=None) def fingerprint_for( *, corpus: str, kind: str, target_ref: str | None, related_ref: str | None, payload: dict[str, Any] | None, ) -> str: """A stable hash of *what this proposal asks for*. Deliberately excludes the rationale, the evidence and the source: the same alias suggested by the weekly lint and by a ticket close is one suggestion, and showing it twice because two different things noticed it is exactly the noise that makes a review queue unusable. Payload keys are sorted and their values normalised, so a model that emits ``["笔记本", "手提电脑"]`` one week and ``["手提电脑", "笔记本"]`` the next does not produce a second row. """ normalised = _normalise(payload or {}) material = json.dumps( { "corpus": corpus, "kind": kind, "target": target_ref or "", "related": related_ref or "", "payload": normalised, }, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ) return hashlib.sha256(material.encode("utf-8")).hexdigest()[:64] def _normalise(value: Any) -> Any: if isinstance(value, dict): return {str(k): _normalise(v) for k, v in sorted(value.items())} if isinstance(value, (list, tuple)): # Sorted: a list of aliases is a set, and its order carries no meaning. return sorted((_normalise(v) for v in value), key=lambda item: json.dumps( item, ensure_ascii=False, sort_keys=True, )) if isinstance(value, str): return " ".join(value.split()).casefold() return value async def propose( db: AsyncSession, *, corpus: str, kind: str, title: str, source: str, target_ref: str | None = None, related_ref: str | None = None, payload: dict[str, Any] | None = None, rationale: str | None = None, evidence: dict[str, Any] | None = None, source_ref: str | None = None, ) -> KnowledgeProposal | None: """File a proposal, or return ``None`` when an identical one is already open. Deduplication is enforced by the database (see the unique constraint on ``(corpus, open_fingerprint)``) rather than by a read-then-write, because the read-then-write loses: the weekly lint and a ticket close can reach the same conclusion inside the same second. """ if kind not in KINDS: raise ValueError(f"unknown proposal kind: {kind!r}") if source not in SOURCES: raise ValueError(f"unknown proposal source: {source!r}") digest = fingerprint_for( corpus=corpus, kind=kind, target_ref=target_ref, related_ref=related_ref, payload=payload, ) row = KnowledgeProposal( corpus=corpus, kind=kind, target_ref=target_ref, related_ref=related_ref, title=(title or "").strip()[:255] or kind, payload=payload, rationale=(rationale or None), evidence=evidence, source=source, source_ref=source_ref, fingerprint=digest, open_fingerprint=digest, status=STATUS_OPEN, ) try: # A SAVEPOINT, so the duplicate case costs one savepoint rollback rather # than poisoning the caller's whole session — an ingest run proposes # many things and the fifth being a duplicate must not discard the # first four. # # The ``add`` belongs INSIDE the block: an instance added outside it # survives the savepoint rollback, sits in ``session.new``, and is # re-attempted by the very next flush — so the caller's next unrelated # write dies with the same IntegrityError, several statements away from # anything that explains it. async with db.begin_nested(): db.add(row) await db.flush() except IntegrityError: logger.debug( "knowledge_loop: %s/%s already open (fingerprint=%s)", corpus, kind, digest[:12], ) # Belt and braces: the savepoint rollback detaches it, but an instance # that somehow remained pending would be a landmine in the caller's # session rather than a returned ``None``. if row in db: db.expunge(row) return None return row async def propose_many( db: AsyncSession, *, items: list[dict[str, Any]], ) -> dict[str, Any]: """File a batch, bounded, reporting how many were already open. The bound is not politeness: a misconfigured importer turning a 3,000-row spreadsheet into 3,000 proposals produces a queue that gets abandoned rather than reviewed, and the corpus is then worse off than before anyone tried to improve it. """ created = 0 duplicates = 0 refused = 0 for item in items: if created >= MAX_PROPOSALS_PER_INGEST: refused += 1 continue row = await propose(db, **item) if row is None: duplicates += 1 else: created += 1 return { "created": created, "duplicates": duplicates, "over_limit": refused, "limit": MAX_PROPOSALS_PER_INGEST, } def serialize(proposal: KnowledgeProposal) -> dict[str, Any]: return { "id": str(proposal.id), "corpus": proposal.corpus, "kind": proposal.kind, "target_ref": proposal.target_ref, "related_ref": proposal.related_ref, "title": proposal.title, "payload": proposal.payload or {}, "rationale": proposal.rationale, "evidence": proposal.evidence or {}, "source": proposal.source, "source_ref": proposal.source_ref, "status": proposal.status, "applied_ref": proposal.applied_ref, "error": proposal.error, "decided_at": proposal.decided_at.isoformat() if proposal.decided_at else None, "decision_note": proposal.decision_note, "created_at": proposal.created_at.isoformat() if proposal.created_at else None, } async def list_proposals( db: AsyncSession, *, corpus: str | None = None, status: str | None = STATUS_OPEN, kind: str | None = None, source: str | None = None, target_ref: str | None = None, page: int = 1, page_size: int = 50, ) -> tuple[list[dict[str, Any]], int]: stmt = select(KnowledgeProposal) count_stmt = select(func.count()).select_from(KnowledgeProposal) conditions = [] if corpus: conditions.append(KnowledgeProposal.corpus == corpus) if status: conditions.append(KnowledgeProposal.status == status) if kind: conditions.append(KnowledgeProposal.kind == kind) if source: conditions.append(KnowledgeProposal.source == source) if target_ref: conditions.append(KnowledgeProposal.target_ref == target_ref) for condition in conditions: stmt = stmt.where(condition) count_stmt = count_stmt.where(condition) total = (await db.execute(count_stmt)).scalar() or 0 rows = ( ( await db.execute( stmt.order_by(KnowledgeProposal.created_at.desc()) .offset(max(0, (page - 1) * page_size)) .limit(page_size) ) ) .scalars() .all() ) return [serialize(row) for row in rows], total async def queue_summary(db: AsyncSession) -> dict[str, Any]: """Open counts per corpus and per kind — what the badge and tabs read.""" rows = ( await db.execute( select( KnowledgeProposal.corpus, KnowledgeProposal.kind, func.count(), ) .where(KnowledgeProposal.status == STATUS_OPEN) .group_by(KnowledgeProposal.corpus, KnowledgeProposal.kind) ) ).all() by_corpus: dict[str, dict[str, int]] = {} total = 0 for corpus, kind, count in rows: bucket = by_corpus.setdefault(corpus, {}) bucket[kind] = int(count) total += int(count) return {"open_total": total, "by_corpus": by_corpus} async def decide( db: AsyncSession, *, proposal: KnowledgeProposal, accept: bool, actor_person_id: uuid.UUID | None, note: str | None = None, trace_id: str = "", payload_override: dict[str, Any] | None = None, ) -> dict[str, Any]: """Accept (and apply) or reject one proposal. Applying is the adapter's job. Three outcomes are distinguished on purpose: * ``accepted`` — applied, and ``applied_ref`` says what it produced; * ``superseded`` — the page moved out from under the proposal. Nobody decided anything; the question stopped being askable. Closing it as a rejection would record an opinion that was never held; * ``failed`` — the write blew up. Stays visible with its error rather than silently returning to the queue, so a broken adapter looks broken. ``payload_override`` is the reviewer filling in a value the machine deliberately refused to invent. Today that is one thing: the body of a ``new_article`` proposal raised from an evaluation failure, which is filed with ``body: ""`` and ``needs_author: True`` because a GENERATED answer is exactly the confident, unsupported prose this whole system exists to keep out of the corpus. Without it those proposals could only ever be rejected — accepting one hit the service's "标题和正文不能为空" validation and failed. ⚠️ **The overridable keys are whitelisted per kind** (``OVERRIDABLE_PAYLOAD_KEYS``), not merely "keys the proposal already declares". That weaker rule was the first shape here and it was wrong: for ``dead_link`` the declared key ``link_id`` IS the target — the adapter resolves it to a row and deletes it — so a reviewer could name any link in the corpus while the decision audit named a different one. A key outside the whitelist is REFUSED, not dropped: a reviewer who typed a value is entitled to know it went nowhere. ⚠️ **Every terminal status clears ``open_fingerprint``, ``failed`` included.** The fingerprint answers one question — "is this same suggestion already OPEN?" — and a decided proposal is not open whatever its outcome. Keeping it on a failure made a transient error permanently silence the suggestion: nothing anywhere moves a proposal back to ``open``, so the next lint pass hit the unique constraint, ``propose`` returned ``None``, and the finding was dropped without a word. The failed row is still visible under the ``failed`` filter with its error, so nothing is lost by letting the suggestion be raised again — and if the cause is systematic, a row per run is how that becomes noticeable rather than invisible. """ if proposal.status != STATUS_OPEN: return {"status": proposal.status, "changed": False} if not accept: proposal.status = STATUS_REJECTED proposal.open_fingerprint = None proposal.decided_by_person_id = actor_person_id proposal.decided_at = _naive_now() proposal.decision_note = note await db.flush() return {"status": STATUS_REJECTED, "changed": True} adapter = adapter_for(proposal.corpus) if adapter is None: proposal.status = STATUS_FAILED proposal.open_fingerprint = None proposal.error = ( f"没有为语料 {proposal.corpus} 注册处理器 —— 这条提案没法落地。" ) proposal.decided_by_person_id = actor_person_id proposal.decided_at = _naive_now() await db.flush() return {"status": STATUS_FAILED, "changed": True, "error": proposal.error} applied_override: list[str] = [] if payload_override: base = dict(proposal.payload or {}) allowed = OVERRIDABLE_PAYLOAD_KEYS.get(proposal.kind, frozenset()) merged = { key: value for key, value in payload_override.items() if key in allowed and key in base } refused = sorted(set(payload_override) - set(merged)) if refused: # Refused loudly rather than dropped. A reviewer who typed a value # is entitled to know it went nowhere — silently discarding it and # then failing the accept on 「标题和正文不能为空」 tells them # nothing about which of the two things went wrong. raise ValidationError( "这些字段不能由审阅人填写:" + "、".join(refused), ) if merged: proposal.payload = {**base, **merged} applied_override = sorted(merged) await db.flush() # A SAVEPOINT around the apply. # # Applying writes — a merge alone touches three tables — and a failed flush # deactivates the WHOLE transaction. Without isolation the handler below # would set ``status = failed`` and flush it, and THAT flush would raise # ``PendingRollbackError``, taking the caller down and leaving the proposal # stuck at ``open``. That is the 2026-08-06 Zoom-seat bug class exactly: a # swallowed error whose recovery bookkeeping is the thing that dies. proposal_id = proposal.id result: dict[str, Any] = {} apply_error: str | None = None superseded = False async with best_effort(db, label="knowledge_loop.apply") as step: result = await adapter.apply( db, proposal=proposal, actor_person_id=actor_person_id, trace_id=trace_id, ) if not step.ok: failure = step.error if isinstance(failure, ProposalApplyError): superseded = failure.superseded apply_error = str(failure) else: logger.exception( "knowledge_loop: applying %s failed", proposal_id, exc_info=failure, ) apply_error = ( f"{type(failure).__name__}: {failure}"[:2000] if failure is not None else "apply failed" ) # The savepoint rollback EXPIRES instances in this session, so the # proposal has to be re-read before anything touches its columns — # otherwise the first attribute access is lazy IO and raises # MissingGreenlet from inside the error path. refreshed = await db.get(KnowledgeProposal, proposal_id) if refreshed is None: return {"status": STATUS_FAILED, "changed": False, "error": apply_error} refreshed.status = STATUS_SUPERSEDED if superseded else STATUS_FAILED # Cleared for BOTH outcomes — see the docstring. A failure is a decision # about this row, not a reason to silence the finding for good. refreshed.open_fingerprint = None refreshed.error = apply_error refreshed.decided_by_person_id = actor_person_id refreshed.decided_at = _naive_now() refreshed.decision_note = note await db.flush() return { "status": refreshed.status, "changed": True, "error": apply_error, } proposal.status = STATUS_ACCEPTED proposal.open_fingerprint = None result.setdefault("authored_fields", applied_override or None) proposal.applied_ref = str(result.get("applied_ref") or "") or None proposal.decided_by_person_id = actor_person_id proposal.decided_at = _naive_now() proposal.decision_note = note proposal.error = None await db.flush() return {"status": STATUS_ACCEPTED, "changed": True, **result} async def supersede_for_target( db: AsyncSession, *, corpus: str, target_ref: str, reason: str, ) -> int: """Close every open proposal against a page that has just gone away. Called by the corpus when it retires or deletes something. Without it a reviewer opens a proposal, reads a careful rationale, clicks accept, and is told the page does not exist — having spent the attention that the queue's credibility is made of. """ rows = ( ( await db.execute( select(KnowledgeProposal).where( KnowledgeProposal.corpus == corpus, KnowledgeProposal.status == STATUS_OPEN, (KnowledgeProposal.target_ref == target_ref) | (KnowledgeProposal.related_ref == target_ref), ) ) ) .scalars() .all() ) for row in rows: row.status = STATUS_SUPERSEDED row.open_fingerprint = None row.decided_at = _naive_now() row.error = reason if rows: await db.flush() return len(rows) __all__ = [ "decide", "fingerprint_for", "list_proposals", "propose", "propose_many", "queue_summary", "serialize", "supersede_for_target", ]