from __future__ import annotations

import sqlite3
import sys
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[1]
HEALTH = ROOT / "scripts" / "health"
sys.path.insert(0, str(HEALTH))
sys.path.insert(0, str(ROOT / "tests"))

from dashboard_v5.read_api import APIError, dispatch_api  # noqa: E402
from dashboard_v5.sprint6i_a_schema import apply_schema as apply_document_schema  # noqa: E402
from dashboard_v5.sprint6i_c_schema import apply_schema as apply_reconciliation_schema  # noqa: E402
from fixtures.dashboard_v5_fixture import build_dashboard_v5_fixture  # noqa: E402
import health_dashboard_action_worker as worker  # noqa: E402


def _database(tmp_path: Path) -> Path:
    database = tmp_path / "sprint7c-e.db"
    build_dashboard_v5_fixture(database)
    connection = sqlite3.connect(database)
    connection.row_factory = sqlite3.Row
    apply_document_schema(connection)
    apply_reconciliation_schema(connection)
    connection.execute("INSERT OR IGNORE INTO dokumente_status(status) VALUES ('eingearbeitet')")
    connection.execute(
        """INSERT INTO dokumente
           (datei_name,dateipfad,daten_typ,kategorie,quelle,status,document_date,
            institution,review_status)
           VALUES('synthetic.pdf','/tmp/synthetic.pdf','pdf','LABOR','synthetic','eingearbeitet',
                  '2026-06-10','Synthetic Lab','nicht_geprueft')"""
    )
    document_id = int(connection.execute("SELECT last_insert_rowid()").fetchone()[0])
    connection.execute(
        """INSERT INTO document_processing
           (document_id,intake_id,original_status,extraction_status,content_status,
            search_status,transfer_status,sha256,page_count,engine,engine_version,
            created_at,updated_at,reconciliation_revision,original_reviewed_at)
           VALUES(?, 'doc_aaaaaaaaaaaaaaaaaaaaaaaa','available','ocr','partial',
                  'machine_searchable','candidates_available',?,1,'synthetic_ocr','1',
                  '2026-06-11','2026-06-11',4,'2026-06-11')""",
        (document_id, "a" * 64),
    )
    connection.execute(
        """INSERT INTO document_text_versions
           (id,document_id,version,source_kind,original_text,normalized_text,engine,
            engine_version,created_at)
           VALUES('text_aaaaaaaaaaaaaaaaaaaaaaaa',?,1,'ocr','synthetic text',
                  'synthetic text','synthetic_ocr','1','2026-06-11')""",
        (document_id,),
    )
    connection.execute(
        """INSERT INTO document_pages
           (id,document_id,text_version,page_number,original_text,normalized_text,confidence,
            section_hash,repetition_status,reviewed_at)
           VALUES('page_aaaaaaaaaaaaaaaaaaaaaaaa',?,1,1,'synthetic text','synthetic text',1.0,?,'first_documented','2026-06-11')""",
        (document_id,"b" * 64),
    )
    connection.execute(
        """INSERT INTO document_reconciliation
           (document_id,queue_bucket,reason_code,priority,open_decisions,
            exact_match_count,conflict_count,prepared_at)
           VALUES(?,'now_reviewable','decision_required',80,5,1,1,'2026-06-11')""",
        (document_id,),
    )
    candidates = [
        ("cand_111111111111111111111111", "CRP: <4,2", "mg/L", 0.91, "open"),
        ("cand_222222222222222222222222", "D-Dimer: 5,1", "µg/L", 0.98, "conflicting"),
        ("cand_333333333333333333333333", "Ferritin: nicht nachweisbar", None, 0.84, "open"),
        ("cand_444444444444444444444444", "Fibrinogen: 3,0", "g/L", 0.99, "already_present"),
        ("cand_555555555555555555555555", "Leukozyten: nicht lesbar", "Tsd/µL", 0.77, "open"),
    ]
    for index, (candidate_id, value, unit, confidence, status) in enumerate(candidates, 1):
        connection.execute(
            """INSERT INTO document_candidates
               (id,document_id,candidate_type,value_text,unit,page_number,section_number,
                context_text,engine,confidence,status,created_at,source_text_version,
                candidate_fingerprint,candidate_revision)
               VALUES(?,?,'laboratory_value',?,?,1,?,?,'synthetic_ocr',?,?,
                      '2026-06-11',1,?,2)""",
            (candidate_id, document_id, value, unit, index, f"Befunddatum 2026-06-10 · excerpt {index}", confidence, status, str(index) * 64),
        )
    matches = [
        (candidates[0][0], "not_present", "crp", "<4.2", "mg/l", 0),
        (candidates[1][0], "value_conflict", "d_dimer", "=5.1", "µg/l", 1),
        (candidates[2][0], "not_present", "ferritin", "nicht nachweisbar", None, 0),
        (candidates[3][0], "exact_match", "fibrinogen", "=3", "g/l", 1),
        (candidates[4][0], "ambiguous", "leukozyten", None, "tsd/µl", 0),
    ]
    for index, (candidate_id, match_status, parameter, value, unit, db_matches) in enumerate(matches, 1):
        connection.execute(
            """INSERT INTO document_candidate_matches
               (candidate_id,match_status,target_area,normalized_parameter,normalized_date,
                normalized_value,normalized_unit,reference_text,database_match_count,
                workbook_match_count,comparison_digest,compared_at)
               VALUES(?,?,'laboratory',?,'2026-06-10',?,?,?, ?,0,?,'2026-06-11')""",
            (candidate_id, match_status, parameter, value, unit, "0–5 mg/L" if index == 1 else None, db_matches, str(index) * 64),
        )
    connection.commit()
    connection.close()
    return database


def _payload(database: Path, query: str = "") -> dict:
    return dispatch_api(database, "/api/v1/lab-review", query)


def test_review_states_keep_match_confidence_separate(tmp_path: Path) -> None:
    payload = _payload(_database(tmp_path), "status=all")
    states = {item["raw_label"]: item["review_status"] for item in payload["candidates"]}
    assert states == {
        "CRP": "suggestion_available",
        "D-Dimer": "conflict",
        "Ferritin": "suggestion_available",
        "Fibrinogen": "possible_duplicate",
        "Leukozyten": "incomplete",
    }
    assert all(item["match_confidence"] is not None for item in payload["candidates"])
    assert not any(item["review_status"] == "confirmed" for item in payload["candidates"])


def test_operator_qualitative_and_unreadable_values_are_never_invented(tmp_path: Path) -> None:
    payload = _payload(_database(tmp_path))
    values = {item["raw_label"]: item["value"] for item in payload["candidates"]}
    assert values["CRP"] == {"raw": "<4,2", "kind": "numeric", "operator": "<", "normalized": "4.2"}
    assert values["Ferritin"] == {"raw": "nicht nachweisbar", "kind": "qualitative", "operator": None, "normalized": None}
    assert values["Leukozyten"] == {"raw": "nicht lesbar", "kind": "unreadable", "operator": None, "normalized": None}


def test_preview_is_revision_bound_and_contains_exact_single_target(tmp_path: Path) -> None:
    candidate = next(item for item in _payload(_database(tmp_path))["candidates"] if item["raw_label"] == "CRP")
    preview = candidate["preview"]
    assert preview["target_count"] == 1
    assert preview["target_parameter"] == "CRP"
    assert preview["value"] == "<4,2"
    assert preview["unit"] == "mg/L"
    assert preview["observation_date"] == "2026-06-10"
    assert preview["reference_range"] == "0–5 mg/L"
    assert preview["candidate_revision"] == 2
    assert len(preview["revision"]) == 64
    assert preview["existing_comparisons"] == 0


def test_filters_are_server_side_and_bounded(tmp_path: Path) -> None:
    database = _database(tmp_path)
    assert [item["raw_label"] for item in _payload(database, "status=conflict")["candidates"]] == ["D-Dimer"]
    assert [item["raw_label"] for item in _payload(database, "q=ferritin")["candidates"]] == ["Ferritin"]
    assert len(_payload(database, "status=all&from=2026-06-10&to=2026-06-10")["candidates"]) == 5
    assert len(_payload(database, "from=2026-06-11&to=2026-06-12")["candidates"]) == 0
    with pytest.raises(APIError):
        _payload(database, "status=confirmed&status=open")


def test_confirmed_coverage_uses_only_canonical_verified_rows(tmp_path: Path) -> None:
    payload = _payload(_database(tmp_path))
    coverage = payload["coverage"]
    assert coverage["title"] == "Bestätigte Laborabdeckung"
    assert coverage["confirmed_values"] == 6
    assert coverage["confirmed_parameters"] == 2
    assert coverage["earliest_date"] == "2024-01-10"
    assert coverage["latest_date"] == "2026-05-01"
    assert coverage["parameters_multiple_dates"] == 1
    assert coverage["parameters_single_date"] == 1
    assert coverage["open_candidates"] == 4
    assert coverage["medical_completeness_claim"] is False
    assert set(coverage["groups"]) == {"Weitere / nicht klassifiziert"}


def test_payload_contains_no_paths_or_internal_filenames(tmp_path: Path) -> None:
    database=_database(tmp_path)
    connection=sqlite3.connect(database);connection.execute("UPDATE document_candidates SET engine='/home/private/secret-ocr' ");connection.commit();connection.close()
    payload = _payload(database)
    serialized = str(payload)
    assert "/tmp/" not in serialized
    assert ".pdf" not in serialized
    assert "synthetic.pdf" not in serialized
    assert "/home/private" not in serialized
    assert all(item["source"]["document_id"].startswith("api-document-") for item in payload["candidates"])


def _worker_action(database: Path, monkeypatch: pytest.MonkeyPatch, candidate: str, operation: str, *, value: str = "", unit: str = "", decision: str = "", action_id: str, expected_revision: int | None = None, preview_revision: str | None = None, bind: bool = True) -> None:
    monkeypatch.setattr(worker, "DASHBOARD_DB", database)
    monkeypatch.setattr(worker, "DASHBOARD_V5_FILE", None)
    payload = {
        "version": 1,
        "action": "document_review",
        "operation": operation,
        "document_id": "doc_aaaaaaaaaaaaaaaaaaaaaaaa",
        "target_id": candidate,
        "value": value,
        "unit": unit,
        "decision": decision,
        "metadata": {},
    }
    if bind and operation in {"candidate_decision", "candidate_remap", "candidate_transfer"} and expected_revision is None:
        projected = next(item for item in _payload(database, "status=all")["candidates"] if item["id"] == candidate)
        expected_revision = projected["revision_number"]
        preview_revision = projected["revision"]
    if expected_revision is not None:
        payload["expected_candidate_revision"] = expected_revision
        payload["preview_revision"] = preview_revision or "f" * 64
    worker.apply_document_review(worker.validate_document_action(payload), action_id)


def test_manual_remap_is_unit_safe_and_audited(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    database = _database(tmp_path)
    candidate = "cand_555555555555555555555555"
    _worker_action(database, monkeypatch, candidate, "candidate_remap", value="lab.platelets", action_id="a" * 32)
    connection = sqlite3.connect(database)
    match = connection.execute("SELECT normalized_parameter,normalized_unit FROM document_candidate_matches WHERE candidate_id=?", (candidate,)).fetchone()
    event = connection.execute("SELECT decision,corrected_value,corrected_unit FROM document_candidate_review_events WHERE action_id=?", ("a" * 32,)).fetchone()
    connection.close()
    assert match[0] == "thrombozyten" and match[1] in {"tsd/µl", "tsd/μl"}
    assert event[0:2] == ("corrected", "Thrombozyten") and event[2] in {"Tsd/µL", "Tsd/μL"}
    with pytest.raises(RuntimeError, match="unit mismatch"):
        _worker_action(database, monkeypatch, "cand_111111111111111111111111", "candidate_remap", value="lab.d_dimer", action_id="b" * 32)


def test_mark_incomplete_is_audited_without_canonical_write(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    database = _database(tmp_path)
    before = sqlite3.connect(database).execute("SELECT COUNT(*) FROM laborwerte").fetchone()[0]
    candidate = "cand_111111111111111111111111"
    _worker_action(database, monkeypatch, candidate, "candidate_decision", value="CRP: <4,2", unit="mg/L", decision="conflicting", action_id="c" * 32)
    payload = _payload(database, "status=incomplete")
    assert candidate in {item["id"] for item in payload["candidates"]}
    connection = sqlite3.connect(database)
    after = connection.execute("SELECT COUNT(*) FROM laborwerte").fetchone()[0]
    marker = connection.execute("SELECT corrected_value FROM document_candidate_review_events WHERE action_id=?", ("c" * 32,)).fetchone()[0]
    connection.close()
    assert before == after
    assert marker == "__incomplete__"


def test_preview_bound_action_fails_closed_after_candidate_change(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    database = _database(tmp_path)
    candidate = "cand_111111111111111111111111"
    connection = sqlite3.connect(database)
    connection.execute("UPDATE document_candidates SET candidate_revision=2 WHERE id=?", (candidate,))
    connection.commit()
    before = connection.execute("SELECT COUNT(*) FROM document_candidate_review_events").fetchone()[0]
    connection.close()
    with pytest.raises(RuntimeError, match="candidate revision conflict"):
        _worker_action(database, monkeypatch, candidate, "candidate_decision", value="CRP: <4,2", unit="mg/L", decision="confirmed", action_id="9" * 32, expected_revision=1)
    connection = sqlite3.connect(database)
    assert connection.execute("SELECT COUNT(*) FROM document_candidate_review_events").fetchone()[0] == before
    assert connection.execute("SELECT COUNT(*) FROM document_transfer_staging").fetchone()[0] == 0
    connection.close()


def test_preview_hash_fails_closed_after_match_change(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    database = _database(tmp_path)
    candidate = "cand_111111111111111111111111"
    item = next(item for item in _payload(database)["candidates"] if item["id"] == candidate)
    connection = sqlite3.connect(database)
    connection.execute(
        "UPDATE document_candidate_matches SET reference_text='0–6 mg/L',comparison_digest=? WHERE candidate_id=?",
        ("8" * 64,candidate),
    )
    connection.commit()
    connection.close()
    with pytest.raises(RuntimeError, match="preview revision conflict"):
        _worker_action(
            database,monkeypatch,candidate,"candidate_decision",value="CRP: <4,2",unit="mg/L",
            decision="confirmed",action_id="7" * 32,expected_revision=item["revision_number"],
            preview_revision=item["revision"],
        )


def test_laboratory_confirmation_cannot_omit_preview_binding(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    database = _database(tmp_path)
    with pytest.raises(RuntimeError, match="preview binding required"):
        _worker_action(
            database, monkeypatch, "cand_111111111111111111111111", "candidate_decision",
            value="CRP: <4,2", unit="mg/L", decision="confirmed", action_id="8" * 32, bind=False,
        )
    connection = sqlite3.connect(database)
    assert connection.execute("SELECT COUNT(*) FROM document_candidate_review_events").fetchone()[0] == 0
    assert connection.execute("SELECT COUNT(*) FROM document_transfer_staging").fetchone()[0] == 0
    connection.close()


def test_laboratory_corrected_confirmation_requires_a_regenerated_preview(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    database=_database(tmp_path);candidate="cand_111111111111111111111111"
    connection=sqlite3.connect(database)
    before=(connection.execute("SELECT COUNT(*) FROM document_candidate_review_events").fetchone()[0],connection.execute("SELECT COUNT(*) FROM document_transfer_staging").fetchone()[0])
    connection.close()
    with pytest.raises(RuntimeError,match="regenerated preview"):
        _worker_action(database,monkeypatch,candidate,"candidate_decision",value="CRP: 99",unit="mg/L",decision="corrected_confirmed",action_id="d1"*16)
    connection=sqlite3.connect(database)
    after=(connection.execute("SELECT COUNT(*) FROM document_candidate_review_events").fetchone()[0],connection.execute("SELECT COUNT(*) FROM document_transfer_staging").fetchone()[0])
    connection.close()
    assert after==before


def test_reviewed_but_currently_unavailable_original_fails_closed(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    database=_database(tmp_path);candidate="cand_111111111111111111111111"
    connection=sqlite3.connect(database)
    connection.execute("UPDATE document_processing SET original_status='missing' WHERE document_id=1")
    connection.commit();connection.close()
    with pytest.raises(RuntimeError,match="source review required"):
        _worker_action(database,monkeypatch,candidate,"candidate_decision",value="CRP: <4,2",unit="mg/L",decision="confirmed",action_id="d2"*16)
    connection=sqlite3.connect(database)
    assert connection.execute("SELECT COUNT(*) FROM document_candidate_review_events").fetchone()[0]==0
    assert connection.execute("SELECT COUNT(*) FROM document_transfer_staging").fetchone()[0]==0
    connection.close()


def test_preview_hash_fails_closed_after_source_review_revision_change(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    database = _database(tmp_path)
    candidate = "cand_111111111111111111111111"
    item = next(item for item in _payload(database)["candidates"] if item["id"] == candidate)
    connection = sqlite3.connect(database)
    connection.execute("UPDATE document_processing SET original_reviewed_at='2026-06-12' WHERE intake_id='doc_aaaaaaaaaaaaaaaaaaaaaaaa'")
    connection.commit(); connection.close()
    with pytest.raises(RuntimeError, match="preview revision conflict"):
        _worker_action(
            database, monkeypatch, candidate, "candidate_decision", value="CRP: <4,2", unit="mg/L",
            decision="confirmed", action_id="b" * 32, expected_revision=item["revision_number"],
            preview_revision=item["revision"],
        )


def test_operator_confirmation_is_explicit_idempotent_and_visible(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    database = _database(tmp_path)
    candidate = "cand_111111111111111111111111"
    _worker_action(database, monkeypatch, candidate, "candidate_decision", value="CRP: <4,2", unit="mg/L", decision="confirmed", action_id="d" * 32)
    staged = next(item for item in _payload(database, "status=needs_review")["candidates"] if item["id"] == candidate)
    assert staged["transfer_preview"]["status"] == "reviewed_pending_preview"
    assert set(staged["transfer_preview"]["new"]) <= {"parameter","value","unit","date","reference"}
    assert "document_id" not in staged["transfer_preview"]["new"]
    document_review=dispatch_api(database,f"/api/v1/documents/{staged['source']['document_id']}/review","")
    document_candidate=next(item for item in document_review["candidates"] if item["id"]==candidate)
    assert set(document_candidate["transfer_preview"]["new"]) <= {"parameter","value","unit","date","reference"}
    assert "document_id" not in document_candidate["transfer_preview"]["new"]
    _worker_action(database, monkeypatch, candidate, "candidate_transfer", decision="transfer", action_id="e" * 32)
    _worker_action(database, monkeypatch, candidate, "candidate_transfer", decision="transfer", action_id="f" * 32)
    labs = dispatch_api(database, "/api/v1/record-labs", "from=2026-06-10&to=2026-06-10")
    inserted = [item for item in labs["observations"] if item["parameter"] == "CRP" and item["date"] == "2026-06-10"]
    assert len(inserted) == 1
    assert inserted[0]["operator"] == "<"
    assert inserted[0]["display_value"] == "<4.2"
    assert inserted[0]["reference"]["raw"] == "0–5 mg/L"
    connection = sqlite3.connect(database)
    stored = connection.execute(
        "SELECT abnahme_datum,befund_datum FROM laborwerte WHERE lower(parameter_name)='crp' AND befund_datum='2026-06-10'"
    ).fetchone()
    connection.close()
    assert stored == (None,"2026-06-10")
    assert candidate in {item["id"] for item in _payload(database, "status=confirmed")["candidates"]}


def test_qualitative_confirmation_preserves_text_without_unit_or_number(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    database = _database(tmp_path)
    candidate = "cand_333333333333333333333333"
    _worker_action(database,monkeypatch,candidate,"candidate_decision",value="Ferritin: nicht nachweisbar",decision="confirmed",action_id="6" * 32)
    _worker_action(database,monkeypatch,candidate,"candidate_transfer",decision="transfer",action_id="5" * 32)
    labs=dispatch_api(database,"/api/v1/record-labs","from=2026-06-10&to=2026-06-10")
    rows=[item for item in labs["observations"] if item["parameter"]=="Ferritin"]
    assert len(rows)==1
    assert rows[0]["value_kind"]=="qualitative"
    assert rows[0]["display_value"]=="nicht nachweisbar"
    assert rows[0]["unit"]==""


def test_transfer_requires_reviewed_original_and_source_page(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    database=_database(tmp_path)
    candidate="cand_111111111111111111111111"
    _worker_action(database,monkeypatch,candidate,"candidate_decision",value="CRP: <4,2",unit="mg/L",decision="confirmed",action_id="4" * 32)
    connection=sqlite3.connect(database)
    connection.execute("UPDATE document_processing SET original_reviewed_at=NULL WHERE intake_id='doc_aaaaaaaaaaaaaaaaaaaaaaaa'")
    connection.commit(); connection.close()
    with pytest.raises(RuntimeError,match="source review required"):
        _worker_action(database,monkeypatch,candidate,"candidate_transfer",decision="transfer",action_id="3" * 32)


def test_censored_confirmed_value_stays_visible_but_out_of_exact_series(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    database=_database(tmp_path);candidate="cand_111111111111111111111111"
    _worker_action(database,monkeypatch,candidate,"candidate_decision",value="CRP: <4,2",unit="mg/L",decision="confirmed",action_id="c1" * 16)
    _worker_action(database,monkeypatch,candidate,"candidate_transfer",decision="transfer",action_id="c2" * 16)
    labs=dispatch_api(database,"/api/v1/record-labs","from=2026-06-10&to=2026-06-10")
    assert any(item["parameter"]=="CRP" and item["display_value"]=="<4.2" for item in labs["observations"])
    series=dispatch_api(database,"/api/v1/series","metric=lab.crp&from=2026-06-10&to=2026-06-10&resolution=day")
    assert not any(point["date"]=="2026-06-10" for point in series["points"])


def test_equal_same_day_values_from_distinct_sources_are_not_collapsed(tmp_path: Path) -> None:
    database=_database(tmp_path);connection=sqlite3.connect(database)
    for marker in ("source-a","source-b"):
        connection.execute(
            """INSERT INTO laborwerte(parameter_name,wert,einheit,befund_datum,bemerking,reference_min,reference_max,
                      quelle,validierungsstatus,source_type,reference_range_source,verified_against_original)
                 VALUES('CRP','2.0','mg/L','2026-06-09','Quell-Referenzbereich: 0–5 mg/L','0','5',?,
                        'validiert','document_candidate','scanned_original',1)""",
            (marker,),
        )
    connection.commit();connection.close()
    labs=dispatch_api(database,"/api/v1/record-labs","from=2026-06-09&to=2026-06-09")
    rows=[item for item in labs["observations"] if item["parameter"]=="CRP" and item["date"]=="2026-06-09"]
    assert len(rows)==2
    assert len({item["id"] for item in rows})==2


def test_reference_parser_preserves_signs_and_detects_units() -> None:
    assert worker._reference_bounds("0-5 mg/L")==('0','5')
    assert worker._reference_bounds(">5 mg/L")==('5',None)
    assert worker._reference_unit("0–5 mg/dL")=="mg/dL"


def test_reference_unit_conflict_blocks_canonical_write(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    database=_database(tmp_path);candidate="cand_111111111111111111111111"
    connection=sqlite3.connect(database);connection.execute("UPDATE document_candidate_matches SET reference_text='0–5 mg/dL' WHERE candidate_id=?",(candidate,));connection.commit();connection.close()
    _worker_action(database,monkeypatch,candidate,"candidate_decision",value="CRP: <4,2",unit="mg/L",decision="confirmed",action_id="0" * 32)
    with pytest.raises(RuntimeError,match="reference unit conflict"):
        _worker_action(database,monkeypatch,candidate,"candidate_transfer",decision="transfer",action_id="a0" * 16)


def test_conflicting_transfer_rolls_back_without_partial_confirmation(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    database = _database(tmp_path)
    connection = sqlite3.connect(database)
    connection.execute(
        """INSERT INTO laborwerte(parameter_name,wert,einheit,abnahme_datum,befund_datum,
                  validierungsstatus,verified_against_original,reference_range_source)
             VALUES('CRP','9.9','mg/L','2026-06-10','2026-06-10','validiert',1,'scanned_original')"""
    )
    connection.commit()
    before = connection.execute("SELECT COUNT(*) FROM laborwerte").fetchone()[0]
    connection.close()
    candidate = "cand_111111111111111111111111"
    _worker_action(database, monkeypatch, candidate, "candidate_decision", value="CRP: <4,2", unit="mg/L", decision="confirmed", action_id="1" * 32)
    with pytest.raises(RuntimeError, match="transfer conflict"):
        _worker_action(database, monkeypatch, candidate, "candidate_transfer", decision="transfer", action_id="2" * 32)
    connection = sqlite3.connect(database)
    after = connection.execute("SELECT COUNT(*) FROM laborwerte").fetchone()[0]
    stage = connection.execute("SELECT status,canonical_row_id FROM document_transfer_staging WHERE candidate_id=?", (candidate,)).fetchone()
    connection.close()
    assert after == before
    assert stage == ("reviewed_pending_preview", None)