fix: separate schema constructor drift contexts

desloppify/app/commands/scan/plan_reconcile.py

@@ -87,6 +87,7 @@ def _plan_has_user_content(plan: dict[str, object]) -> bool:
         or plan.get("overrides")
         or plan.get("clusters")
         or plan.get("skipped")
+        or plan.get("uncommitted_issues")
     )
 
 

desloppify/app/commands/scan/workflow.py

@@ -470,6 +470,11 @@ def merge_scan_results(
             subjective_integrity_target=target_score,
             project_root=str(get_project_root()),
             zone_map=runtime.lang.zone_map if runtime.lang else None,
+            semantic_corrections=(
+                runtime.lang.semantic_corrections
+                if runtime.lang is not None and scan_path_rel == "."
+                else None
+            ),
         ),
     )
 

desloppify/engine/_plan/scan_issue_reconcile.py

@@ -367,6 +367,35 @@ def _sync_skipped_issue_statuses(plan: PlanModel, state: StateModel) -> None:
             issue["status"] = target_status
 
 
+def _purge_semantic_correction_uncommitted(
+    plan: PlanModel,
+    state: StateModel,
+    *,
+    result: ReconcileResult,
+) -> None:
+    """Drop system-invalidated findings from the pending commit ledger."""
+
+    issues = state.get("work_items") or state.get("issues", {})
+    uncommitted = plan.get("uncommitted_issues", [])
+    retained: list[str] = []
+    removed = 0
+    for issue_id in uncommitted:
+        issue = issues.get(issue_id)
+        attestation = issue.get("resolution_attestation", {}) if issue else {}
+        if (
+            issue
+            and issue.get("status") == "false_positive"
+            and isinstance(attestation, dict)
+            and attestation.get("kind") == "detector_semantic_correction"
+        ):
+            removed += 1
+            continue
+        retained.append(issue_id)
+    if removed:
+        plan["uncommitted_issues"] = retained
+        result.changes += removed
+
+
 def reconcile_plan_after_scan(
     plan: PlanModel,
     state: StateModel,
@@ -397,6 +426,7 @@ def reconcile_plan_after_scan(
     # Sync state status for issues in plan.skipped that are still "open" in state.
     # This migrates existing data: temporary skips → deferred, triaged_out skips → triaged_out.
     _sync_skipped_issue_statuses(plan, state)
+    _purge_semantic_correction_uncommitted(plan, state, result=result)
 
     _supersede_dead_references(
         plan,

desloppify/engine/_state/merge.py

@@ -20,6 +20,7 @@
     _record_scan_metadata,
 )
 from desloppify.engine._state.merge_issues import (
+    apply_semantic_corrections,
     verify_disappeared,
     find_suspect_detectors,
     upsert_issues,
@@ -149,6 +150,7 @@ class MergeScanOptions:
     subjective_integrity_target: float | None = None
     project_root: str | None = None
     zone_map: Any | None = None
+    semantic_corrections: dict[str, dict[str, str]] | None = None
 
 
 def merge_scan(
@@ -219,6 +221,16 @@ def merge_scan(
         resolved_options.force_resolve,
         ran_detectors,
     )
+    semantic_correction_ids = apply_semantic_corrections(
+        existing,
+        current_ids,
+        resolved_options.semantic_corrections,
+        now,
+        lang=resolved_options.lang,
+        scan_path=resolved_options.scan_path,
+        confirmed_detectors=confirmed_detectors,
+        suspect_detectors=suspect_detectors,
+    )
     auto_resolved, skipped_other_lang, resolved_out_of_scope, resolve_changed = verify_disappeared(
         existing,
         current_ids,
@@ -230,6 +242,7 @@ def merge_scan(
         project_root=resolved_options.project_root,
         zone_map=resolved_options.zone_map,
         confirmed_detectors=confirmed_detectors,
+        semantic_correction_ids=semantic_correction_ids,
     )
 
     # Mark subjective assessments stale when mechanical issues changed.

desloppify/engine/_state/merge_issues.py

@@ -79,6 +79,80 @@ def _mark_scan_verified(
     existing["scan_verification_text"] = attestation_text
 
 
+def apply_semantic_corrections(
+    existing: dict,
+    current_ids: set[str],
+    corrections: Mapping[str, Mapping[str, str]] | None,
+    now: str,
+    *,
+    lang: str | None,
+    scan_path: str | None,
+    confirmed_detectors: set[str],
+    suspect_detectors: set[str],
+) -> set[str]:
+    """Apply detector-proven false-positive corrections from a full scan.
+
+    Corrections are scanner-owned evidence, not a generic replacement for a
+    user disposition: the item must still be open, absent from the current
+    detector output, and produced by a confirmed non-suspect detector.
+    """
+
+    if scan_path != "." or not corrections:
+        return set()
+
+    corrected_issue_ids: set[str] = set()
+    for issue_id, correction in corrections.items():
+        previous = existing.get(issue_id)
+        if not isinstance(previous, dict) or issue_id in current_ids:
+            continue
+        if previous.get("status") != "open" or previous.get("suppressed"):
+            continue
+        detector = str(correction.get("detector", ""))
+        correction_kind = str(correction.get("kind", "")).strip()
+        correction_file = str(correction.get("file", "")).strip()
+        correction_key = str(correction.get("key", "")).strip()
+        correction_line = str(correction.get("line", "")).strip()
+        detail = previous.get("detail")
+        if (
+            not detector
+            or not correction_kind
+            or not correction_file
+            or not correction_key
+            or not correction_line
+            or previous.get("detector") != detector
+            or previous.get("file") != correction_file
+            or detector not in confirmed_detectors
+            or detector in suspect_detectors
+            or not isinstance(detail, dict)
+            or detail.get("kind") != correction_kind
+            or str(detail.get("key", "")) != correction_key
+            or str(detail.get("line", "")) != correction_line
+        ):
+            continue
+        if lang and previous.get("lang") not in (None, lang):
+            continue
+        rule = str(correction.get("rule", "")).strip()
+        evidence = str(correction.get("evidence", "")).strip()
+        if not rule or not evidence:
+            continue
+
+        previous.update(
+            status="false_positive",
+            resolved_at=now,
+            note=f"Detector semantic correction ({rule}): {evidence}",
+            resolution_attestation={
+                "kind": "detector_semantic_correction",
+                "rule": rule,
+                "scan_verified": True,
+                "scan_verified_at": now,
+                "scan_verification_text": evidence,
+            },
+        )
+        corrected_issue_ids.add(issue_id)
+
+    return corrected_issue_ids
+
+
 def verify_disappeared(
     existing: dict,
     current_ids: set[str],
@@ -91,6 +165,7 @@ def verify_disappeared(
     project_root: str | None = None,
     zone_map=None,
     confirmed_detectors: set[str] | None = None,
+    semantic_correction_ids: set[str] | None = None,
 ) -> tuple[int, int, int, set[str]]:
     """Update scan corroboration for issues absent from scan.
 
@@ -103,6 +178,8 @@ def verify_disappeared(
     resolved_detectors: set[str] = set()
 
     for issue_id, previous in existing.items():
+        if semantic_correction_ids and issue_id in semantic_correction_ids:
+            continue
         previous_status = previous.get("status")
         if issue_id in current_ids or previous_status not in (
             "open",
@@ -325,6 +402,7 @@ def _suppression_metadata_from_state(
 
 
 __all__ = [
+    "apply_semantic_corrections",
     "verify_disappeared",
     "find_suspect_detectors",
     "upsert_issues",

desloppify/languages/_framework/base/types.py

@@ -88,6 +88,7 @@ class LangRuntimeContract(Protocol):
     subjective_assessments: dict[str, Any]
     detector_coverage: dict[str, DetectorCoverageRecord]
     coverage_warnings: list[DetectorCoverageRecord]
+    semantic_corrections: dict[str, dict[str, str]]
 
     def runtime_setting(self, key: str, default: Any = None) -> Any: ...
 

desloppify/languages/_framework/runtime_support/accessors.py

@@ -118,6 +118,16 @@ def coverage_warnings(self) -> list[DetectorCoverageRecord]:
     def coverage_warnings(self, value: list[DetectorCoverageRecord]) -> None:
         self.state.coverage_warnings = value
 
+    @property
+    def semantic_corrections(self) -> dict[str, dict[str, str]]:
+        """Detector-proven historical findings invalidated during this scan."""
+
+        return self.state.semantic_corrections
+
+    @semantic_corrections.setter
+    def semantic_corrections(self, value: dict[str, dict[str, str]]) -> None:
+        self.state.semantic_corrections = value
+
     @property
     def large_threshold(self) -> int:
         override = self.state.large_threshold_override

desloppify/languages/_framework/runtime_support/runtime.py

@@ -36,6 +36,7 @@ class LangRuntimeState:
     props_threshold_override: int = 0
     detector_coverage: dict[str, DetectorCoverageRecord] = field(default_factory=dict)
     coverage_warnings: list[DetectorCoverageRecord] = field(default_factory=list)
+    semantic_corrections: dict[str, dict[str, str]] = field(default_factory=dict)
 
 
 @dataclass