truf-server/app/keycheckers/keycheck_common.py

import hashlib
import json
import os
import re
import shutil
import sqlite3
import sys
import time
import uuid
from collections import OrderedDict
from contextlib import contextmanager
from datetime import datetime, timezone
from itertools import cycle

import requests

PROJECT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if PROJECT_DIR not in sys.path:
    sys.path.append(PROJECT_DIR)

from paths import apply_path_config, default_project_paths
from scanner_db import ScannerDB, extract_raw_secret, json_dumps, safe_json_loads, sha256_text
from lifecycle_authority import LifecycleAuthorityError, require_active_supervisor_child
from runtime_security import (
    canonical_path,
    durable_replace,
    durable_unlink,
    PrivateFileLock,
    harden_private_file,
    read_private_json,
    reject_reparse_components,
    require_private_directory,
    require_private_file,
)


STATUS_TRANSACTION_JOURNAL_FILENAME = ".status-transaction.pending.json"
STATUS_TRANSACTION_LOCK_FILENAME = ".status-transaction.lock"
STATUS_TRANSACTION_MAX_KEY_BYTES = 8192
STATUS_TRANSACTION_MAX_LINE_BYTES = 8192
STATUS_TRANSACTION_MAX_JOURNAL_BYTES = 64 * 1024
GCP_STATUS_MAX_LINE_BYTES = 256 * 1024
GCP_STATUS_MAX_JOURNAL_BYTES = 1024 * 1024
GENERIC_SK_PROVIDERS = frozenset(("qwen", "deepseek", "kimi", "zai"))
QWEN_DEEPSEEK_PROVIDERS = frozenset(("qwen", "deepseek"))
AMBIGUOUS_QWEN_DEEPSEEK_HINT = "ambiguous_qwen_deepseek"
AMBIGUOUS_GENERIC_SK_HINT = "ambiguous_generic_sk"
EXPLICIT_ASSIGNMENT_PROVIDER_HINT_SOURCE = "explicit_assignment"
QWEN_EXPLICIT_ROUTING_DETECTORS = frozenset(("qwendashscope", "qwen_dashscope"))
DEEPSEEK_EXPLICIT_ROUTING_DETECTORS = frozenset(("deepseekapikey", "deepseek_api_key"))
KIMI_EXPLICIT_ROUTING_DETECTORS = frozenset(("kimimoonshot", "moonshotai"))
ZAI_EXPLICIT_ROUTING_DETECTORS = frozenset(("zaiglm",))
GENERIC_SK_NON_EXPLICIT_ROUTING_DETECTORS = frozenset(("qwen", "dashscope", "deepseek"))
PROVIDER_ROUTING_EVIDENCE_CACHE = OrderedDict()
PROVIDER_ROUTING_EVIDENCE_CACHE_AUTHORITY = None
PROVIDER_ROUTING_EVIDENCE_DB_FAILED = False
REVIEWED_CORRUPTION_MAX_ROWS = 100000
KEYCHECK_CAPACITY_BLOCKED_EXIT = 76
REVIEWED_CORRUPTION_LEDGER_MAX_BYTES = 512 * 1024 * 1024
REVIEWED_CORRUPTION_CLASSIFICATIONS = frozenset((
    'invalid_json', 'legacy_numeric_prefix_corrupt_json', 'invalid_utf8',
    'utf8_bom_prefix', 'unterminated_record', 'invalid_error_projection',
    'oversized_record',
))
_POSTGRES_DB = None
_ACTIVE_DB_CANDIDATE = None


def keycheck_input_mode():
    default = 'postgres' if os.getenv('SCANNER_SUPERVISED') == '1' else 'jsonl'
    return str(os.getenv('KEYCHECK_INPUT_MODE') or default).strip().lower()


def _postgres_candidate_db():
    global _POSTGRES_DB
    if _POSTGRES_DB is not None and _POSTGRES_DB.enabled:
        return _POSTGRES_DB
    db_url = str(os.getenv('KEYCHECK_DB_URL') or os.getenv('SCANNER_DB_URL') or '').strip()
    if not db_url:
        raise RuntimeError('PostgreSQL keycheck input mode requires KEYCHECK_DB_URL')
    _POSTGRES_DB = ScannerDB(db_url=db_url, initialize=False)
    if not _POSTGRES_DB.enabled:
        raise RuntimeError('PostgreSQL keycheck candidate database is unavailable')
    _POSTGRES_DB.set_application_name(f'truf-keycheck-provider:{keycheck_service_name() or "unknown"}')
    _POSTGRES_DB.require_runtime_safety_schema()
    _POSTGRES_DB.require_final_cutover()
    return _POSTGRES_DB


def now_iso():
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def env_int(name, default):
    try:
        return int(os.getenv(name, default))
    except (TypeError, ValueError):
        return default


def require_provider_authority(service):
    """Authenticate a leaf checker before it parses arguments or opens findings."""
    metadata = require_active_supervisor_child(
        child_kind='keycheck-provider',
        require_dsn=True,
        handshake_timeout_retries=1,
    )
    expected_service = str(service or '').strip().lower()
    if not expected_service or str(os.getenv('KEYCHECK_SERVICE') or '').strip().lower() != expected_service:
        raise LifecycleAuthorityError('keycheck provider capability does not match this service')
    canonical_dsn = os.getenv('TRUF_MANAGED_POSTGRES_DSN') or ''
    if os.getenv('KEYCHECK_DB_URL') != canonical_dsn:
        raise LifecycleAuthorityError('keycheck provider database URL is not supervisor-managed')
    if keycheck_input_mode() == 'postgres' and any(
        str(value).lower() in ('--input', '--plain')
        or str(value).lower().startswith(('--input=', '--plain='))
        for value in sys.argv[1:]
    ):
        raise LifecycleAuthorityError(
            'PostgreSQL keycheck providers forbid compatibility input/plain arguments'
        )
    try:
        import yaml

        with open(metadata['config_path'], 'r', encoding='utf-8') as handle:
            config = apply_path_config(yaml.safe_load(handle) or {}, metadata['config_path'])
        expected_output = os.path.join((config.get('global') or {})['keycheck_dir'], expected_service)
    except Exception as exc:
        raise LifecycleAuthorityError('keycheck provider could not validate its canonical output path') from exc
    output_dir = os.getenv('KEYCHECK_OUTPUT_DIR') or ''
    state_dir = os.getenv('KEYCHECK_STATE_DIR') or ''
    if canonical_path(output_dir) != canonical_path(expected_output) or canonical_path(state_dir) != canonical_path(expected_output):
        raise LifecycleAuthorityError('keycheck provider output capability is outside its canonical service directory')
    return metadata


def load_keycheck_layout(config_path=None):
    if config_path:
        try:
            import yaml
            with open(config_path, "r", encoding="utf-8") as f:
                config = apply_path_config(yaml.safe_load(f) or {}, config_path)
            return config.get("global") or {}
        except Exception:
            pass
    return default_project_paths()


def service_output_dir(service, config_path=None, output_dir=None):
    if output_dir:
        return output_dir
    layout = load_keycheck_layout(config_path)
    return os.path.join(layout["keycheck_dir"], service)


def default_input_file(config_path=None):
    layout = load_keycheck_layout(config_path)
    return os.path.join(layout["results_dir"], "found_secrets.jsonl")


def default_proxy_file(config_path=None):
    layout = load_keycheck_layout(config_path)
    return layout["proxy_file"]


def truthy_env(name, default=False):
    value = os.getenv(name)
    if value is None:
        return default
    return str(value).strip().lower() in ("1", "true", "yes", "on")


def keycheck_service_name():
    service = os.getenv("KEYCHECK_SERVICE")
    if service:
        return service.strip().lower()
    output_dir = os.getenv("KEYCHECK_OUTPUT_DIR")
    if output_dir:
        return os.path.basename(os.path.normpath(output_dir)).lower()
    return ""


def status_projection_line_max_bytes():
    if os.getenv("KEYCHECK_INPUT_LIST_MAX_LINE_BYTES") is not None:
        return max(1, env_int("KEYCHECK_INPUT_LIST_MAX_LINE_BYTES", STATUS_TRANSACTION_MAX_LINE_BYTES))
    if keycheck_service_name() == "gcp":
        return GCP_STATUS_MAX_LINE_BYTES
    return STATUS_TRANSACTION_MAX_LINE_BYTES


def status_transaction_journal_max_bytes():
    if keycheck_service_name() == "gcp":
        return GCP_STATUS_MAX_JOURNAL_BYTES
    return STATUS_TRANSACTION_MAX_JOURNAL_BYTES


def keycheck_state_path(service=None):
    service = service or keycheck_service_name() or "default"
    state_dir = os.getenv("KEYCHECK_STATE_DIR")
    if not state_dir:
        output_dir = os.getenv("KEYCHECK_OUTPUT_DIR")
        state_dir = output_dir or service_output_dir(service)
    require_private_directory(state_dir, create=True)
    return os.path.join(state_dir, "input_state.json")


def read_json_file(path, default=None):
    try:
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    except (OSError, ValueError):
        return default


def write_json_file(path, data):
    with private_atomic_writer(path) as f:
        json.dump(data, f, ensure_ascii=False, indent=2, sort_keys=True)


def input_file_signature(path, handle=None):
    details = os.fstat(handle.fileno()) if handle is not None else os.stat(path)
    return {
        "path": os.path.abspath(path),
        "device": int(getattr(details, "st_dev", 0) or 0),
        "size": int(details.st_size),
        "mtime_ns": int(getattr(details, "st_mtime_ns", int(details.st_mtime * 1_000_000_000))),
        "inode": int(getattr(details, "st_ino", 0) or 0),
    }


def jsonl_manifest_path(path):
    base, _ = os.path.splitext(path)
    return f"{base}.manifest.json"


def load_jsonl_manifest(path):
    try:
        if os.path.getsize(jsonl_manifest_path(path)) > 1024 * 1024:
            raise RuntimeError(f"keycheck JSONL manifest exceeds bounded size: {jsonl_manifest_path(path)}")
        with open(jsonl_manifest_path(path), "r", encoding="utf-8") as f:
            data = json.load(f)
            return data if isinstance(data, dict) else {}
    except FileNotFoundError:
        return {}


def write_jsonl_manifest(path, manifest):
    manifest_path = jsonl_manifest_path(path)
    with private_atomic_writer(manifest_path) as f:
        json.dump(manifest, f, ensure_ascii=False, indent=2, sort_keys=True)


@contextmanager
def private_atomic_writer(path, binary=False, suffix=".tmp"):
    """Publish a same-directory file whose ACL is exact-private before payload writes."""
    path = os.path.abspath(path)
    require_private_directory(os.path.dirname(path), create=True)
    if os.path.lexists(path):
        require_private_file(path)
    temporary = f"{path}.{os.getpid()}.{uuid.uuid4().hex}{suffix}"
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
    if hasattr(os, "O_BINARY"):
        flags |= os.O_BINARY
    descriptor = os.open(temporary, flags, 0o600)
    published = False
    try:
        harden_private_file(temporary)
        require_private_file(temporary)
        mode = "wb" if binary else "w"
        options = {} if binary else {"encoding": "utf-8", "newline": ""}
        with os.fdopen(descriptor, mode, **options) as handle:
            descriptor = None
            yield handle
            handle.flush()
            os.fsync(handle.fileno())
        require_private_file(temporary)
        durable_replace(temporary, path)
        published = True
        require_private_file(path)
    finally:
        if descriptor is not None:
            os.close(descriptor)
        if not published and os.path.exists(temporary):
            os.remove(temporary)


@contextmanager
def private_append_writer(path, binary=False):
    """Open an exact-private output for append, rejecting unsafe existing files."""
    path = os.path.abspath(path)
    ensure_output_files([path])
    require_private_file(path)
    mode = "ab" if binary else "a"
    options = {} if binary else {"encoding": "utf-8", "newline": ""}
    with open(path, mode, **options) as handle:
        yield handle
        handle.flush()
        os.fsync(handle.fileno())
    require_private_file(path)


def acquire_file_lock(lock_path, stale_sec=300, timeout_sec=30):
    require_private_directory(os.path.dirname(os.path.abspath(lock_path)), create=True)
    reject_reparse_components(os.path.dirname(os.path.abspath(lock_path)))
    deadline = time.monotonic() + max(1, timeout_sec)
    while True:
        lock = PrivateFileLock(lock_path)
        try:
            return lock.acquire()
        except BlockingIOError:
            if time.monotonic() >= deadline:
                raise TimeoutError(f"timed out acquiring lock {lock_path}")
            time.sleep(0.05)


def release_file_lock(lock, lock_path):
    try:
        lock.release()
    except (AttributeError, OSError):
        return


def next_jsonl_segment_path(path, manifest):
    base, ext = os.path.splitext(path)
    seq = int(manifest.get("next_sequence") or 1)
    while True:
        segment = f"{base}.{seq:06d}{ext or '.jsonl'}"
        if not os.path.exists(segment):
            return segment, seq
        seq += 1


def physical_jsonl_segments(path):
    root = os.path.dirname(os.path.abspath(path))
    base, extension = os.path.splitext(os.path.basename(path))
    pattern = re.compile(rf"^{re.escape(base)}\.(\d{{6}}){re.escape(extension)}$")
    output = []
    inspect_limit = max(2, env_int("KEYCHECK_RESULTS_MAX_SEGMENTS", 16) + 1)
    try:
        with os.scandir(root) as entries:
            for entry in entries:
                match = pattern.fullmatch(entry.name)
                if not match:
                    continue
                if entry.is_symlink() or not entry.is_file(follow_symlinks=False):
                    raise RuntimeError(f"unsafe keycheck JSONL segment: {entry.path}")
                output.append((int(match.group(1)), os.path.abspath(entry.path)))
                if len(output) > inspect_limit:
                    raise RuntimeError(f"keycheck JSONL segment count exceeds its bound: {path}")
    except FileNotFoundError:
        return []
    return sorted(output)


def _keycheck_files_share_prefix(segment_path, current_path, size):
    if size <= 0 or not os.path.isfile(current_path) or os.path.getsize(current_path) < size:
        return False
    remaining = size
    with open(segment_path, "rb") as segment, open(current_path, "rb") as current:
        while remaining:
            amount = min(1024 * 1024, remaining)
            left = segment.read(amount)
            right = current.read(amount)
            if not left or left != right:
                return False
            remaining -= len(left)
    return remaining == 0


def _publish_empty_keycheck_generation(path):
    with private_atomic_writer(path, binary=True, suffix=".empty.tmp"):
        pass


def _keycheck_remove_prefix(path, size):
    current_size = os.path.getsize(path)
    if current_size == size:
        _publish_empty_keycheck_generation(path)
        return
    require_private_file(path)
    with open(path, "rb") as source, private_atomic_writer(path, binary=True, suffix=".prefix.tmp") as destination:
        source.seek(size)
        shutil.copyfileobj(source, destination, 1024 * 1024)


def _keycheck_file_generation(path):
    details = os.stat(path, follow_symlinks=False)
    return {
        "device": int(getattr(details, "st_dev", 0) or 0),
        "size": int(details.st_size),
        "mtime_ns": int(getattr(details, "st_mtime_ns", int(details.st_mtime * 1_000_000_000))),
        "inode": int(getattr(details, "st_ino", 0) or 0),
    }


def _keycheck_manifest_skip_valid(path, manifest, skip):
    signature = manifest.get("current_skip_signature") if isinstance(manifest, dict) else None
    if not isinstance(signature, dict) or not os.path.isfile(path):
        return False
    try:
        current = _keycheck_file_generation(path)
        return int(skip) <= current["size"] and all(
            int(current[key]) == int(signature.get(key, -1))
            for key in ("device", "size", "mtime_ns", "inode")
        )
    except (OSError, TypeError, ValueError):
        return False


def repair_keycheck_jsonl_tail(path):
    if not os.path.exists(path) or os.path.getsize(path) == 0:
        return 0
    require_private_file(path)
    scan_limit = max(1, env_int("KEYCHECK_JSONL_TAIL_SCAN_MAX_BYTES", 8 * 1024 * 1024))
    quarantine_limit = max(1, env_int("KEYCHECK_JSONL_TORN_QUARANTINE_MAX_BYTES", 64 * 1024))
    size = os.path.getsize(path)
    with open(path, "r+b") as handle:
        handle.seek(-1, os.SEEK_END)
        if handle.read(1) == b"\n":
            return 0
        start = max(0, size - scan_limit)
        handle.seek(start)
        tail = handle.read(size - start)
        newline = tail.rfind(b"\n")
        if newline < 0 and start:
            raise RuntimeError(f"keycheck JSONL tail exceeds bounded repair window: {path}")
        truncate_at = start + newline + 1 if newline >= 0 else 0
        torn = tail[newline + 1:] if newline >= 0 else tail
        quarantine = f"{path}.torn-tail.bin"
        with private_atomic_writer(quarantine, binary=True) as output:
            output.write(torn[:quarantine_limit])
        handle.truncate(truncate_at)
        handle.flush()
        os.fsync(handle.fileno())
    return size - truncate_at


def reconcile_keycheck_jsonl_segments(path):
    manifest = load_jsonl_manifest(path)
    physical = physical_jsonl_segments(path)
    listed = {
        str(item.get("name") or os.path.basename(str(item.get("path") or ""))): item
        for item in (manifest.get("segments") or []) if isinstance(item, dict)
    }
    segments = []
    missing = []
    for sequence, segment_path in physical:
        item = listed.get(os.path.basename(segment_path))
        if item is None:
            item = {
                "name": os.path.basename(segment_path), "path": segment_path,
                "bytes": os.path.getsize(segment_path), "closed_at": now_iso(),
                "sequence": sequence,
            }
            missing.append(segment_path)
        else:
            item = dict(item, name=os.path.basename(segment_path), path=segment_path, sequence=sequence)
        segments.append(item)
    skip = max(0, int(manifest.get("current_skip_bytes") or 0))
    duplicate = None
    for segment_path in list(reversed(missing)) + ([physical[-1][1]] if skip and physical else []):
        size = os.path.getsize(segment_path)
        if _keycheck_files_share_prefix(segment_path, path, size):
            duplicate = (segment_path, size)
            break
    effective_skip = duplicate[1] if duplicate else 0
    if missing or segments != (manifest.get("segments") or []) or skip or manifest.get("current_skip_signature"):
        manifest.update({
            "current": os.path.basename(path), "current_path": os.path.abspath(path),
            "next_sequence": max([item[0] for item in physical] or [0]) + 1,
            "segments": segments,
            "current_skip_bytes": effective_skip,
            "current_skip_signature": _keycheck_file_generation(path) if effective_skip and os.path.isfile(path) else None,
            "updated_at": now_iso(),
        })
        write_jsonl_manifest(path, manifest)
        if duplicate:
            _keycheck_remove_prefix(path, duplicate[1])
            manifest["current_skip_bytes"] = 0
            manifest["current_skip_signature"] = None
            manifest["updated_at"] = now_iso()
            write_jsonl_manifest(path, manifest)
    return manifest


def rotate_jsonl_if_needed(path, max_bytes):
    if not max_bytes or max_bytes <= 0 or not os.path.lexists(path):
        return
    reject_reparse_components(path)
    repair_keycheck_jsonl_tail(path)
    size = os.stat(path, follow_symlinks=False).st_size
    if size < max_bytes:
        return
    manifest = reconcile_keycheck_jsonl_segments(path)
    max_segments = max(1, env_int("KEYCHECK_RESULTS_MAX_SEGMENTS", 16))
    if len(physical_jsonl_segments(path)) >= max_segments:
        raise RuntimeError(
            f"keycheck JSONL segment bound reached for {path}; ingest and retire segments offline"
        )
    segment_path, seq = next_jsonl_segment_path(path, manifest)
    require_private_file(path)
    with open(path, "rb") as source, private_atomic_writer(segment_path, binary=True) as destination:
            shutil.copyfileobj(source, destination, 1024 * 1024)
    segments = manifest.get("segments") if isinstance(manifest.get("segments"), list) else []
    segments.append({
        "name": os.path.basename(segment_path),
        "path": segment_path,
        "bytes": int(size),
        "closed_at": now_iso(),
        "sequence": seq,
    })
    manifest.update({
        "current": os.path.basename(path),
        "current_path": path,
        "next_sequence": seq + 1,
        "max_bytes": int(max_bytes),
        "segments": segments,
        "current_skip_bytes": int(size),
        "current_skip_signature": _keycheck_file_generation(path),
        "updated_at": now_iso(),
    })
    write_jsonl_manifest(path, manifest)
    _publish_empty_keycheck_generation(path)
    manifest["current_skip_bytes"] = 0
    manifest["current_skip_signature"] = None
    manifest["updated_at"] = now_iso()
    write_jsonl_manifest(path, manifest)


def should_rotate_jsonl(path):
    return os.path.basename(str(path or "")).lower().endswith("results.jsonl")


def tail_fallback_bytes():
    tail_mb = os.getenv("KEYCHECK_INPUT_TAIL_MB") or os.getenv("KEYCHECK_TAIL_MB")
    try:
        return int(float(tail_mb) * 1024 * 1024) if tail_mb else int(os.getenv("KEYCHECK_INPUT_TAIL_BYTES", "0") or 0)
    except (TypeError, ValueError):
        return 0


def keycheck_input_max_line_bytes():
    return max(1024, env_int("KEYCHECK_INPUT_MAX_LINE_BYTES", 16 * 1024 * 1024))


def keycheck_reconciliation_ledger_path(input_file):
    base, _ = os.path.splitext(os.path.abspath(input_file))
    return base + '.publication-ledger.sqlite3'


def load_reviewed_keycheck_corruptions(input_file):
    ledger_path = keycheck_reconciliation_ledger_path(input_file)
    if not os.path.lexists(ledger_path):
        return {}
    require_private_file(ledger_path)
    max_bytes = max(
        1024 * 1024,
        env_int('KEYCHECK_RECONCILIATION_LEDGER_MAX_BYTES', REVIEWED_CORRUPTION_LEDGER_MAX_BYTES),
    )
    if os.path.getsize(ledger_path) > max_bytes:
        raise RuntimeError('keycheck reconciliation ledger exceeds its byte bound')
    uri = 'file:' + ledger_path.replace('\\', '/') + '?mode=ro'
    connection = sqlite3.connect(uri, uri=True, timeout=5)
    connection.row_factory = sqlite3.Row
    try:
        connection.execute('PRAGMA query_only=ON')
        connection.execute('PRAGMA busy_timeout=5000')
        columns = {
            row['name'] for row in connection.execute('PRAGMA table_info(reconciliation_issue)').fetchall()
        }
        required = {
            'identity_key', 'file_name', 'file_device', 'file_inode', 'file_size',
            'file_mtime_ns', 'byte_offset', 'byte_length', 'record_sha256',
            'classification', 'status', 'resolved_at',
        }
        if not required.issubset(columns):
            raise RuntimeError('keycheck reconciliation ledger schema is incomplete')
        marker = connection.execute(
            'SELECT value FROM publication_meta WHERE key = ?', ('bootstrapped:finding_uid',),
        ).fetchone()
        if not marker or str(marker['value'] or '') != '1':
            raise RuntimeError('keycheck reconciliation ledger is not complete')
        max_rows = min(
            REVIEWED_CORRUPTION_MAX_ROWS,
            max(1, env_int('KEYCHECK_REVIEWED_CORRUPTION_MAX_ROWS', REVIEWED_CORRUPTION_MAX_ROWS)),
        )
        rows = connection.execute(
            '''SELECT file_name, file_device, file_inode, file_size, file_mtime_ns,
                      byte_offset, byte_length, record_sha256, classification
               FROM reconciliation_issue
               WHERE identity_key = 'finding_uid' AND status = 'resolved' AND resolved_at IS NOT NULL
               ORDER BY file_name, byte_offset LIMIT ?''',
            (max_rows + 1,),
        ).fetchall()
    finally:
        connection.close()
    if len(rows) > max_rows:
        raise RuntimeError('reviewed keycheck corruption count exceeds its bound')
    reviewed = {}
    for row in rows:
        file_name = str(row['file_name'] or '')
        offset = int(row['byte_offset'])
        length = int(row['byte_length'])
        digest = str(row['record_sha256'] or '').lower()
        classification = str(row['classification'] or '')
        if (
            not file_name or os.path.basename(file_name) != file_name
            or offset < 0 or length <= 0
            or not re.fullmatch(r'[a-f0-9]{64}', digest)
            or classification not in REVIEWED_CORRUPTION_CLASSIFICATIONS
        ):
            raise RuntimeError('reviewed keycheck corruption metadata is invalid')
        file_rows = reviewed.setdefault(file_name, {})
        if offset in file_rows:
            raise RuntimeError('reviewed keycheck corruption offsets are ambiguous')
        file_rows[offset] = {
            'device': str(row['file_device']),
            'inode': str(row['file_inode']),
            'size': int(row['file_size']),
            'mtime_ns': str(row['file_mtime_ns']),
            'length': length,
            'sha256': digest,
            'classification': classification,
        }
    return reviewed


def reviewed_corruptions_for_file(reviewed, path, signature):
    rows = (reviewed or {}).get(os.path.basename(path), {})
    if not rows:
        return {}
    actual = {
        'device': str(signature.get('device', 0)),
        'inode': str(signature.get('inode', 0)),
        'size': int(signature.get('size', -1)),
        'mtime_ns': str(signature.get('mtime_ns', -1)),
    }
    for issue in rows.values():
        expected = {key: issue[key] for key in ('device', 'inode', 'size', 'mtime_ns')}
        if actual != expected:
            raise RuntimeError(f'reviewed keycheck corruption file identity changed: {path}')
    return rows


def skip_reviewed_corruption(path, offset, length, digest, reviewed):
    issue = (reviewed or {}).get(int(offset))
    if issue is None:
        return False
    if int(issue['length']) != int(length) or issue['sha256'] != str(digest or '').lower():
        raise RuntimeError(f'reviewed keycheck corruption record changed: {path}:byte:{offset}')
    print(
        f"Warning: skipped exact reviewed keycheck corruption at {path}:byte:{offset}; "
        f"classification={issue['classification']}",
        flush=True,
    )
    return True


def _drain_oversized_jsonl_identity(handle, first_chunk):
    digest = hashlib.sha256(first_chunk)
    length = len(first_chunk)
    if first_chunk.endswith(b"\n"):
        return True, length, digest.hexdigest()
    while True:
        chunk = handle.readline(64 * 1024)
        if not chunk:
            return False, length, digest.hexdigest()
        digest.update(chunk)
        length += len(chunk)
        if chunk.endswith(b"\n"):
            return True, length, digest.hexdigest()


def _drain_oversized_jsonl_record(handle, first_chunk):
    return _drain_oversized_jsonl_identity(handle, first_chunk)[0]


def _warn_skipped_oversized(path, line_offset, max_line_bytes):
    print(
        f"Warning: skipped oversized committed keycheck input at {path}:byte:{line_offset}; "
        f"line limit={max_line_bytes}",
        flush=True,
    )


def checkpoint_matches_signature(path, state, signature):
    if not isinstance(state, dict) or state.get("path") != os.path.abspath(path):
        return False
    state_inode = int(state.get("inode", 0) or 0)
    signature_inode = int(signature.get("inode", 0) or 0)
    if not state_inode or not signature_inode or state_inode != signature_inode:
        return False
    state_device = int(state.get("device", 0) or 0)
    signature_device = int(signature.get("device", 0) or 0)
    if (state_device or signature_device) and (
        not state_device or not signature_device or state_device != signature_device
    ):
        return False
    state_size = int(state.get("size", -1))
    signature_size = int(signature.get("size", 0))
    if state_size < 0 or state_size > signature_size:
        return False
    if state_size == signature_size:
        state_mtime = int(state.get("mtime_ns", -1))
        signature_mtime = int(signature.get("mtime_ns", -2))
        if state_mtime < 0 or state_mtime != signature_mtime:
            return False
    return True


def input_start_offset(input_file, state, signature):
    if truthy_env("KEYCHECK_DISABLE_HIGH_WATERMARK"):
        tail_bytes = tail_fallback_bytes()
        return max(0, signature["size"] - tail_bytes) if tail_bytes > 0 else 0, "tail_disabled" if tail_bytes > 0 else "full_disabled"

    state = state or {}
    if checkpoint_matches_signature(input_file, state, signature):
        offset = int(state.get("offset", 0) or 0)
        if 0 <= offset <= signature["size"]:
            return offset, "high_watermark"
    return 0, "full_replay" if state else "full_initial"


def iter_jsonl_input(input_file):
    physical_segments = physical_jsonl_segments(input_file)
    if not os.path.exists(input_file) and not physical_segments:
        print(f"Warning: {input_file} not found. Nothing to read.")
        return
    manifest_path = jsonl_manifest_path(input_file)
    if os.path.exists(manifest_path) or physical_segments:
        yield from iter_segmented_jsonl_input(input_file, manifest_path)
        return
    service = keycheck_service_name()
    state_file = keycheck_state_path(service)
    state = read_json_file(state_file, {}) or {}
    mode = "unopened"
    final_offset = 0
    processed = 0
    skipped_oversized = 0
    previous_skipped_oversized = 0
    completed_read = False
    checkpoint_signature = None
    max_line_bytes = keycheck_input_max_line_bytes()
    reviewed = load_reviewed_keycheck_corruptions(input_file)
    skipped_reviewed = 0
    try:
        with open(input_file, "rb") as f:
            signature = input_file_signature(input_file, f)
            file_reviews = reviewed_corruptions_for_file(reviewed, input_file, signature)
            start_offset, mode = input_start_offset(input_file, state, signature)
            previous_skipped_oversized = (
                int(state.get("skipped_oversized", 0) or 0)
                if mode == "high_watermark" else 0
            )
            final_offset = start_offset
            print(
                f"Input reader: service={service or 'unknown'} mode={mode} "
                f"offset={start_offset} size={signature['size']} state={state_file}",
                flush=True,
            )
            if start_offset > 0:
                f.seek(start_offset)
                if mode.startswith("tail"):
                    boundary_offset = f.tell()
                    skipped = f.readline(max_line_bytes + 1)
                    if len(skipped) > max_line_bytes:
                        if not _drain_oversized_jsonl_record(f, skipped):
                            raise RuntimeError(f"torn oversized keycheck tail boundary record in {input_file}")
                        _warn_skipped_oversized(input_file, boundary_offset, max_line_bytes)
                        skipped_oversized += 1
                    elif skipped and not skipped.endswith(b"\n"):
                        raise RuntimeError(f"torn keycheck tail boundary record in {input_file}")
                    final_offset = f.tell()
            while True:
                line_offset = f.tell()
                raw_line = f.readline(max_line_bytes + 1)
                if not raw_line:
                    checkpoint_signature = input_file_signature(input_file, f)
                    completed_read = True
                    break
                final_offset = f.tell()
                if len(raw_line) > max_line_bytes:
                    complete, record_length, record_digest = _drain_oversized_jsonl_identity(f, raw_line)
                    if not complete:
                        final_offset = line_offset
                        completed_read = False
                        raise RuntimeError(f"torn oversized committed keycheck input at {input_file}:{line_offset}")
                    final_offset = f.tell()
                    if skip_reviewed_corruption(
                        input_file, line_offset, record_length, record_digest, file_reviews,
                    ):
                        skipped_oversized += 1
                        skipped_reviewed += 1
                        continue
                    skipped_oversized += 1
                    _warn_skipped_oversized(input_file, line_offset, max_line_bytes)
                    continue
                if not raw_line.endswith(b"\n"):
                    if skip_reviewed_corruption(
                        input_file, line_offset, len(raw_line), hashlib.sha256(raw_line).hexdigest(), file_reviews,
                    ):
                        skipped_reviewed += 1
                        continue
                    final_offset = line_offset
                    completed_read = False
                    raise RuntimeError(f"torn committed keycheck input at {input_file}:{line_offset}")
                try:
                    line = raw_line.decode("utf-8")
                    data = json.loads(line)
                except (UnicodeDecodeError, ValueError) as exc:
                    if skip_reviewed_corruption(
                        input_file, line_offset, len(raw_line), hashlib.sha256(raw_line).hexdigest(), file_reviews,
                    ):
                        skipped_reviewed += 1
                        continue
                    final_offset = line_offset
                    completed_read = False
                    kind = "torn" if not raw_line.endswith(b"\n") else "invalid"
                    raise RuntimeError(f"{kind} committed keycheck input at {input_file}:{line_offset}") from exc
                processed += 1
                yield {
                    "source": f"{input_file}:byte:{line_offset}",
                    "line_offset": line_offset,
                    "data": data,
                }
    finally:
        if completed_read and checkpoint_signature is not None:
            new_state = {
                **checkpoint_signature,
                "offset": int(final_offset),
                "processed": int(processed),
                "skipped_oversized": previous_skipped_oversized + int(skipped_oversized),
                "skipped_reviewed_corrupt": int(state.get('skipped_reviewed_corrupt', 0) or 0) + skipped_reviewed,
                "mode": mode,
                "updated_at": now_iso(),
            }
            write_json_file(state_file, new_state)
            print(
                f"Input reader done: service={service or 'unknown'} processed={processed} "
                f"skipped_oversized={skipped_oversized} new_offset={final_offset}",
                flush=True,
            )
        else:
            print(
                f"Input reader stopped before EOF: service={service or 'unknown'} processed={processed}; "
                f"state offset unchanged",
                flush=True,
            )


def manifest_input_files(input_file, manifest):
    root = os.path.dirname(os.path.abspath(input_file))
    physical = physical_jsonl_segments(input_file)
    files = [path for _, path in physical]
    listed_names = {
        os.path.basename(str(segment.get("path") or segment.get("name") or ""))
        for segment in (manifest.get("segments") or []) if isinstance(segment, dict)
    }
    missing = sorted(name for name in listed_names if name and name not in {os.path.basename(path) for path in files})
    if missing:
        raise RuntimeError(f"manifest-listed keycheck segment is missing: {missing[0]}")
    current = os.path.abspath(input_file)
    if current and os.path.exists(current):
        files.append(os.path.abspath(current))
    seen = set()
    out = []
    for path in files:
        if path not in seen:
            seen.add(path)
            out.append(path)
    return out


def iter_segmented_jsonl_input(input_file, manifest_path):
    service = keycheck_service_name()
    state_file = keycheck_state_path(service)
    state = read_json_file(state_file, {}) or {}
    manifest = read_json_file(manifest_path, {}) or {}
    files_state = state.get("files") if isinstance(state.get("files"), dict) else {}
    files = manifest_input_files(input_file, manifest)
    first_manifest_run = not bool(files_state)
    force_replay = truthy_env("KEYCHECK_DISABLE_HIGH_WATERMARK")
    current_path = os.path.abspath(manifest.get("current_path") or input_file)
    if current_path != os.path.abspath(input_file):
        raise RuntimeError("keycheck manifest current path does not match the configured input")
    # Physical segments are always discoverable and downstream publication is
    # idempotent. Never seek past current-file bytes based on mutable manifest
    # state; a writer could truncate and append after any generation check.
    skip_current = 0
    physical = physical_jsonl_segments(input_file)
    total_processed = 0
    total_skipped_oversized = 0
    max_line_bytes = keycheck_input_max_line_bytes()
    reviewed = load_reviewed_keycheck_corruptions(input_file)
    total_skipped_reviewed = 0
    print(
        f"Input reader: service={service or 'unknown'} mode=manifest files={len(files)} "
        f"state={state_file}",
        flush=True,
    )
    try:
        for path in files:
            file_state = files_state.get(os.path.abspath(path)) or files_state.get(path) or {}
            processed = 0
            skipped_oversized = 0
            skipped_reviewed = 0
            with open(path, "rb") as f:
                signature = input_file_signature(path, f)
                file_reviews = reviewed_corruptions_for_file(reviewed, path, signature)
                if force_replay:
                    offset = 0
                    previous_skipped_oversized = 0
                    previous_skipped_reviewed = 0
                elif checkpoint_matches_signature(path, file_state, signature):
                    offset = int(file_state.get("offset", 0) or 0)
                    previous_skipped_oversized = int(file_state.get("skipped_oversized", 0) or 0)
                    previous_skipped_reviewed = int(file_state.get('skipped_reviewed_corrupt', 0) or 0)
                else:
                    offset = 0
                    previous_skipped_oversized = 0
                    previous_skipped_reviewed = 0
                if offset < 0 or offset > signature["size"]:
                    offset = 0
                    previous_skipped_oversized = 0
                    previous_skipped_reviewed = 0
                if not force_replay and first_manifest_run and os.path.abspath(path) == current_path:
                    offset, _ = input_start_offset(path, {}, signature)
                if os.path.abspath(path) == current_path:
                    offset = max(offset, min(skip_current, signature["size"]))
                final_offset = offset
                f.seek(offset)
                while True:
                    line_offset = f.tell()
                    raw_line = f.readline(max_line_bytes + 1)
                    if not raw_line:
                        break
                    final_offset = f.tell()
                    if len(raw_line) > max_line_bytes:
                        complete, record_length, record_digest = _drain_oversized_jsonl_identity(f, raw_line)
                        if not complete:
                            final_offset = line_offset
                            raise RuntimeError(f"torn oversized committed keycheck input at {path}:{line_offset}")
                        final_offset = f.tell()
                        if skip_reviewed_corruption(
                            path, line_offset, record_length, record_digest, file_reviews,
                        ):
                            skipped_oversized += 1
                            skipped_reviewed += 1
                            total_skipped_reviewed += 1
                            continue
                        skipped_oversized += 1
                        _warn_skipped_oversized(path, line_offset, max_line_bytes)
                        continue
                    if not raw_line.endswith(b"\n"):
                        if skip_reviewed_corruption(
                            path, line_offset, len(raw_line), hashlib.sha256(raw_line).hexdigest(), file_reviews,
                        ):
                            skipped_reviewed += 1
                            total_skipped_reviewed += 1
                            continue
                        final_offset = line_offset
                        raise RuntimeError(f"torn committed keycheck input at {path}:{line_offset}")
                    try:
                        line = raw_line.decode("utf-8")
                        data = json.loads(line)
                    except (UnicodeDecodeError, ValueError) as exc:
                        if skip_reviewed_corruption(
                            path, line_offset, len(raw_line), hashlib.sha256(raw_line).hexdigest(), file_reviews,
                        ):
                            skipped_reviewed += 1
                            total_skipped_reviewed += 1
                            continue
                        final_offset = line_offset
                        kind = "torn" if not raw_line.endswith(b"\n") else "invalid"
                        raise RuntimeError(f"{kind} committed keycheck input at {path}:{line_offset}") from exc
                    processed += 1
                    total_processed += 1
                    yield {
                        "source": f"{path}:byte:{line_offset}",
                        "line_offset": line_offset,
                        "data": data,
                    }
                checkpoint_signature = input_file_signature(path, f)
            files_state[os.path.abspath(path)] = {
                **checkpoint_signature,
                "offset": int(final_offset),
                "done": int(final_offset) >= int(checkpoint_signature["size"]),
                "processed": int(processed),
                "skipped_oversized": previous_skipped_oversized + int(skipped_oversized),
                "skipped_reviewed_corrupt": previous_skipped_reviewed + skipped_reviewed,
                "updated_at": now_iso(),
            }
            total_skipped_oversized += skipped_oversized
    finally:
        new_state = {
            "manifest_path": os.path.abspath(manifest_path),
            "current_file": os.path.abspath(input_file),
            "files": files_state,
            "processed": int(total_processed),
            "skipped_oversized": sum(
                int(item.get("skipped_oversized", 0) or 0)
                for item in files_state.values() if isinstance(item, dict)
            ),
            "skipped_reviewed_corrupt": sum(
                int(item.get('skipped_reviewed_corrupt', 0) or 0)
                for item in files_state.values() if isinstance(item, dict)
            ),
            "mode": "manifest",
            "updated_at": now_iso(),
        }
        write_json_file(state_file, new_state)
        print(
            f"Input reader done: service={service or 'unknown'} mode=manifest "
            f"processed={total_processed} skipped_oversized={total_skipped_oversized}",
            flush=True,
        )


def mask_secret(value):
    if not value or len(value) < 12:
        return value or ""
    return f"{value[:8]}...{value[-4:]}"


def keycheck_status_group(status):
    status = str(status or "UNKNOWN").strip().upper()
    alive = {"ALIVE", "VALID", "VALID_2FA", "VALID_RATE_LIMITED", "BEDROCK", "ADMIN", "VERTEX", "CANARY", "FOUNDRY"}
    dead = {"DEAD", "INVALID", "EXPIRED", "LEAKED_REVOKED", "INVALID_OR_REVOKED"}
    restricted = {"RESTRICTED", "API_DISABLED", "ACCESS_DENIED", "QUARANTINED", "DISABLED"}
    no_balance = {"NO_BALANCE", "NO_QUOTA", "LIMITED_OR_NO_BALANCE", "LIMITED_OR_QUOTA"}
    no_context = {"NO_CONTEXT", "NO_USERNAME", "NO_TARGET", "NO_TARGET_MODELS", "NO_GENERATION_MODEL"}
    limited = {"LIMITED", "RATE_LIMITED"}
    network = {"NETWORK", "NETWORK_ERROR"}
    if status in alive:
        return "alive"
    if status in dead:
        return "dead"
    if status in restricted:
        return "restricted"
    if status in no_balance:
        return "no_balance"
    if status in no_context:
        return "no_context"
    if status in limited:
        return "limited"
    if status in network:
        return "network"
    return "unknown"


def redact_value(value, raw_key):
    if isinstance(value, dict):
        return {k: redact_value(v, raw_key) for k, v in value.items() if k not in ("finding", "raw_secret", "raw", "raw_v2")}
    if isinstance(value, list):
        return [redact_value(item, raw_key) for item in value]
    if isinstance(value, str):
        text = value.replace(raw_key, "***REDACTED***") if raw_key else value
        return text[:4000]
    return value


def result_message(result):
    if not isinstance(result, dict):
        return ""
    if result.get("message"):
        return result.get("message")
    error = result.get("error")
    if isinstance(error, dict):
        return error.get("message") or json.dumps(error, ensure_ascii=False)[:1000]
    return error or ""


def finding_secret_hash(finding):
    if not isinstance(finding, dict):
        return ""
    raw = extract_raw_secret(finding)
    return sha256_text(raw) if raw else ""


def clear_provider_routing_evidence_cache():
    global PROVIDER_ROUTING_EVIDENCE_CACHE_AUTHORITY, PROVIDER_ROUTING_EVIDENCE_DB_FAILED
    PROVIDER_ROUTING_EVIDENCE_CACHE.clear()
    PROVIDER_ROUTING_EVIDENCE_CACHE_AUTHORITY = None
    PROVIDER_ROUTING_EVIDENCE_DB_FAILED = False


def _provider_routing_explicit_evidence(finding):
    if not isinstance(finding, dict):
        return set()
    detector = str(finding.get("DetectorName") or "").strip().lower()
    detector_name = detector
    if detector == "customregex":
        extra = finding.get("ExtraData") if isinstance(finding.get("ExtraData"), dict) else {}
        detector_name = str(extra.get("name") or "").strip().lower()

    evidence = set()
    if detector_name in QWEN_EXPLICIT_ROUTING_DETECTORS:
        evidence.add("qwen")
    if detector_name in DEEPSEEK_EXPLICIT_ROUTING_DETECTORS:
        evidence.add("deepseek")
    if detector_name in KIMI_EXPLICIT_ROUTING_DETECTORS:
        evidence.add("kimi")
    if detector_name in ZAI_EXPLICIT_ROUTING_DETECTORS:
        evidence.add("zai")

    context = finding.get("ScannerContext") if isinstance(finding.get("ScannerContext"), dict) else {}
    if context.get("provider_hint_source") != EXPLICIT_ASSIGNMENT_PROVIDER_HINT_SOURCE:
        return evidence
    hint = context.get("provider_hint")
    if hint == AMBIGUOUS_QWEN_DEEPSEEK_HINT:
        evidence.update(QWEN_DEEPSEEK_PROVIDERS)
        return evidence
    if hint == AMBIGUOUS_GENERIC_SK_HINT:
        evidence.update(GENERIC_SK_PROVIDERS)
        return evidence
    if hint in GENERIC_SK_PROVIDERS:
        evidence.add(hint)
        return evidence
    return set(GENERIC_SK_PROVIDERS)


def _provider_routing_cache_limits():
    return {
        "items": min(100000, max(1, env_int("KEYCHECK_PROVIDER_EVIDENCE_CACHE_ITEMS", 4096))),
        "rows": min(1024, max(1, env_int("KEYCHECK_PROVIDER_EVIDENCE_MAX_ROWS", 64))),
        "json_chars": min(
            1024 * 1024,
            max(1024, env_int("KEYCHECK_PROVIDER_EVIDENCE_MAX_JSON_CHARS", 256 * 1024)),
        ),
    }


def _provider_routing_database_authority():
    db_url = str(os.getenv("KEYCHECK_DB_URL") or "").strip()
    db_path = "" if db_url else str(os.getenv("KEYCHECK_DB_PATH") or "").strip()
    identity = db_url or (os.path.abspath(db_path) if db_path else "")
    return db_url, db_path, sha256_text(identity) if identity else ""


def _load_provider_routing_evidence(secret_hash, limits):
    db_url, db_path, _ = _provider_routing_database_authority()
    if not db_url and not db_path:
        raise RuntimeError("canonical keycheck database is unavailable")
    db = ScannerDB(db_path=db_path or None, db_url=db_url or "", initialize=False)
    try:
        if not db.enabled:
            raise RuntimeError("canonical keycheck database connection is unavailable")
        db.require_runtime_safety_schema()
        if getattr(db.conn, "is_postgres", False):
            statement_timeout_ms = min(
                30000,
                max(250, env_int("KEYCHECK_PROVIDER_EVIDENCE_STATEMENT_TIMEOUT_MS", 5000)),
            )
            lock_timeout_ms = min(
                statement_timeout_ms,
                max(100, env_int("KEYCHECK_PROVIDER_EVIDENCE_LOCK_TIMEOUT_MS", 1000)),
            )
            db.conn.execute("SELECT set_config('statement_timeout', ?, false)", (f"{statement_timeout_ms}ms",))
            db.conn.execute("SELECT set_config('lock_timeout', ?, false)", (f"{lock_timeout_ms}ms",))
        rows = db.provider_routing_finding_rows(
            secret_hash,
            limits["rows"] + 1,
            limits["json_chars"],
        )
    finally:
        db.close()

    if len(rows) > limits["rows"]:
        return frozenset(GENERIC_SK_PROVIDERS)
    evidence = set()
    for row in rows:
        if bool(row["truncated"]):
            return frozenset(GENERIC_SK_PROVIDERS)
        raw_finding_json = row["raw_finding_json"] or ""
        if not raw_finding_json:
            detector_name = str(row["detector_name"] or "").strip().lower()
            detector_evidence = _provider_routing_explicit_evidence({
                "DetectorName": detector_name,
            })
            if detector_evidence:
                evidence.update(detector_evidence)
            elif detector_name not in GENERIC_SK_NON_EXPLICIT_ROUTING_DETECTORS:
                return frozenset(GENERIC_SK_PROVIDERS)
            if GENERIC_SK_PROVIDERS.issubset(evidence):
                return frozenset(GENERIC_SK_PROVIDERS)
            continue
        try:
            finding = json.loads(raw_finding_json)
        except (TypeError, ValueError, json.JSONDecodeError):
            return frozenset(GENERIC_SK_PROVIDERS)
        if isinstance(finding, dict) and isinstance(finding.get("finding"), dict):
            finding = finding["finding"]
        if not isinstance(finding, dict):
            return frozenset(GENERIC_SK_PROVIDERS)
        raw = extract_raw_secret(finding)
        if not raw or sha256_text(raw) != secret_hash:
            return frozenset(GENERIC_SK_PROVIDERS)
        evidence.update(_provider_routing_explicit_evidence(finding))
        if GENERIC_SK_PROVIDERS.issubset(evidence):
            return frozenset(GENERIC_SK_PROVIDERS)
    return frozenset(evidence)


def global_provider_routing_evidence(raw_key):
    """Return persisted generic sk-key provider evidence without retaining the raw key."""
    global PROVIDER_ROUTING_EVIDENCE_CACHE_AUTHORITY, PROVIDER_ROUTING_EVIDENCE_DB_FAILED
    PROVIDER_ROUTING_EVIDENCE_DB_FAILED = False
    secret_hash = sha256_text(raw_key)
    if not secret_hash:
        return set(GENERIC_SK_PROVIDERS)
    _, _, authority = _provider_routing_database_authority()
    if authority != PROVIDER_ROUTING_EVIDENCE_CACHE_AUTHORITY:
        PROVIDER_ROUTING_EVIDENCE_CACHE.clear()
        PROVIDER_ROUTING_EVIDENCE_CACHE_AUTHORITY = authority
        PROVIDER_ROUTING_EVIDENCE_DB_FAILED = False
    cached = PROVIDER_ROUTING_EVIDENCE_CACHE.get(secret_hash)
    if cached is not None:
        evidence, retry_at = cached
        if retry_at is None or retry_at > time.monotonic():
            PROVIDER_ROUTING_EVIDENCE_CACHE.move_to_end(secret_hash)
            PROVIDER_ROUTING_EVIDENCE_DB_FAILED = retry_at is not None
            return set(evidence)
        del PROVIDER_ROUTING_EVIDENCE_CACHE[secret_hash]

    limits = _provider_routing_cache_limits()
    try:
        evidence = _load_provider_routing_evidence(secret_hash, limits)
    except Exception:
        PROVIDER_ROUTING_EVIDENCE_DB_FAILED = True
        evidence = frozenset(GENERIC_SK_PROVIDERS)
        retry_at = time.monotonic() + min(
            5,
            max(1, env_int("KEYCHECK_PROVIDER_EVIDENCE_ERROR_CACHE_SEC", 1)),
        )
    else:
        retry_at = None

    if evidence == GENERIC_SK_PROVIDERS:
        PROVIDER_ROUTING_EVIDENCE_CACHE[secret_hash] = (evidence, retry_at)
        PROVIDER_ROUTING_EVIDENCE_CACHE.move_to_end(secret_hash)
        while len(PROVIDER_ROUTING_EVIDENCE_CACHE) > limits["items"]:
            PROVIDER_ROUTING_EVIDENCE_CACHE.popitem(last=False)
    return set(evidence)


def provider_routing_database_failed():
    return bool(PROVIDER_ROUTING_EVIDENCE_DB_FAILED)


def combined_provider_routing_hint(raw_key, local_hint=""):
    global PROVIDER_ROUTING_EVIDENCE_DB_FAILED
    PROVIDER_ROUTING_EVIDENCE_DB_FAILED = False
    evidence = set()
    if local_hint == AMBIGUOUS_QWEN_DEEPSEEK_HINT:
        evidence.update(QWEN_DEEPSEEK_PROVIDERS)
    elif local_hint == AMBIGUOUS_GENERIC_SK_HINT:
        evidence.update(GENERIC_SK_PROVIDERS)
    elif local_hint in GENERIC_SK_PROVIDERS:
        evidence.add(local_hint)
    elif keycheck_input_mode() == 'postgres':
        service = keycheck_service_name()
        if service in GENERIC_SK_PROVIDERS:
            evidence.add(service)
    if len(evidence) > 1:
        return (
            AMBIGUOUS_QWEN_DEEPSEEK_HINT
            if evidence == set(QWEN_DEEPSEEK_PROVIDERS)
            else AMBIGUOUS_GENERIC_SK_HINT
        )
    evidence.update(global_provider_routing_evidence(raw_key))
    if len(evidence) > 1:
        return (
            AMBIGUOUS_QWEN_DEEPSEEK_HINT
            if evidence == set(QWEN_DEEPSEEK_PROVIDERS)
            else AMBIGUOUS_GENERIC_SK_HINT
        )
    if not evidence:
        return ""
    provider = next(iter(evidence))
    if provider not in GENERIC_SK_PROVIDERS:
        return AMBIGUOUS_GENERIC_SK_HINT
    return provider


def strip_finding_nearby_context(finding):
    if not isinstance(finding, dict):
        return finding
    output = dict(finding)
    context = output.get("ScannerContext")
    if isinstance(context, dict) and "nearby" in context:
        output["ScannerContext"] = {key: value for key, value in context.items() if key != "nearby"}
    return output


def finding_detector_secret_hash(finding):
    if not isinstance(finding, dict):
        return ""
    raw = extract_raw_secret(finding)
    secret_hash = sha256_text(raw) if raw else ""
    fallback_hash = sha256_text(json_dumps(finding)) if not secret_hash else ""
    detector = str(finding.get("DetectorName") or finding.get("DetectorType") or "")
    return sha256_text("|".join([detector, secret_hash or fallback_hash])) if detector or secret_hash or fallback_hash else ""


def finding_uid(finding):
    if not isinstance(finding, dict):
        return ""
    return str(finding.get("finding_uid") or finding.get("FindingUID") or "")


def keycheck_event_id(service, key_hash, status, source, finding_uid_value, checked_at, result_source, payload):
    explicit = payload.get("event_id") if isinstance(payload, dict) else ""
    if explicit:
        return str(explicit)
    cached_id = payload.get("cached_occurrence_id") if isinstance(payload, dict) else ""
    if cached_id:
        return sha256_text("|".join([str(service or ""), "cached", str(cached_id)]))
    return sha256_text("|".join([
        str(service or ""),
        str(key_hash or ""),
        str(status or ""),
        str(source or ""),
        str(finding_uid_value or ""),
        str(checked_at or ""),
        str(result_source or "api_check"),
    ]))


def write_keycheck_event(service, results_file, key, result, source="", finding=None, detector=None, result_source="api_check"):
    result = result if isinstance(result, dict) else {"status": str(result or "UNKNOWN")}
    finding = strip_finding_nearby_context(finding)
    status = str(result.get("status") or "UNKNOWN").upper()
    checked_at = result.get("checked_at") or now_iso()
    detector_name = detector or get_detector_name(finding or {}) or result.get("detector") or ""
    key_hash = result.get("key_hash") or sha256_text(key)
    secret_hash = result.get("secret_hash") or finding_secret_hash(finding) or key_hash
    detector_secret_hash = result.get("detector_secret_hash") or finding_detector_secret_hash(finding)
    finding_uid_value = result.get("finding_uid") or finding_uid(finding)
    payload = {
        "key_masked": result.get("key_masked") or mask_secret(key),
        "key_hash": key_hash,
        "secret_hash": secret_hash,
        "detector_secret_hash": detector_secret_hash,
        "finding_uid": finding_uid_value,
        "detector": detector_name,
        "source": source,
        "finding": finding or {},
        "checked_at": checked_at,
        "result_source": result.get("result_source") or result_source or "api_check",
        **result,
    }
    payload["status"] = status
    payload["checked_at"] = checked_at
    payload["result_source"] = result.get("result_source") or result_source or "api_check"
    payload["event_id"] = keycheck_event_id(
        service,
        payload.get("key_hash"),
        payload.get("status"),
        source,
        payload.get("finding_uid"),
        payload.get("checked_at"),
        payload.get("result_source"),
        payload,
    )
    if keycheck_input_mode() == 'postgres':
        global _ACTIVE_DB_CANDIDATE
        candidate = _ACTIVE_DB_CANDIDATE
        if not candidate or str(candidate.get('service') or '') != str(service or ''):
            raise RuntimeError('keycheck result has no exact active PostgreSQL candidate lease')
        payload['event_id'] = sha256_text('|'.join((
            'truf-keycheck-event-v1', str(candidate['id']), str(candidate['attempts']),
        )))
        metadata = {
            name: value for name, value in payload.items()
            if name not in ('finding', 'key_masked', 'key_hash', 'secret_hash')
        }
        resolved_service = str(payload.get('resolved_provider') or '').strip().lower()
        if resolved_service:
            metadata['resolution_origin_service'] = str(service or '').strip().lower()
        outcome = _postgres_candidate_db().complete_keycheck_candidate(
            candidate['id'], candidate['lease_token'], payload['event_id'],
            payload['status'], keycheck_status_group(payload['status']),
            checked_at=payload['checked_at'], message=result_message(payload),
            metadata=redact_value(metadata, key), result_source=payload['result_source'],
            resolved_service=resolved_service,
        )
        if not outcome or not outcome.get('completed'):
            raise RuntimeError('PostgreSQL keycheck candidate completion lost its fence')
        candidate['_completed'] = True
    else:
        append_jsonl(results_file, payload)
    return payload


def record_validation_result(service, key, result, source="", finding=None, detector=None, db_path=None):
    if keycheck_input_mode() == 'postgres':
        candidate = _ACTIVE_DB_CANDIDATE
        return bool(candidate and candidate.get('_completed'))
    db_path = db_path or os.getenv("KEYCHECK_DB_PATH") or os.getenv("SCANNER_DB_PATH") or os.getenv("SCAN_DB_PATH")
    db_url = os.getenv("KEYCHECK_DB_URL") or os.getenv("SCANNER_DB_URL") or os.getenv("DATABASE_URL")
    if not db_path and not db_url:
        return
    if not truthy_env("KEYCHECK_DB_INLINE", False):
        return False
    result = result if isinstance(result, dict) else {"status": str(result or "UNKNOWN")}
    finding = strip_finding_nearby_context(finding)
    status = str(result.get("status") or "UNKNOWN").upper()
    secret_hash = finding_secret_hash(finding) or sha256_text(key)
    detector_secret_hash = finding_detector_secret_hash(finding)
    metadata = redact_value({**result, "source_line": source, "detector_secret_hash": detector_secret_hash}, key)
    result_source = metadata.get("result_source") or "api_check"
    try:
        db = ScannerDB(db_path=db_path, db_url=db_url, initialize=False)
        try:
            db.require_runtime_safety_schema()
            if getattr(db.conn, "is_postgres", False):
                statement_timeout_ms = max(1000, env_int("KEYCHECK_DB_STATEMENT_TIMEOUT_MS", 8000))
                lock_timeout_ms = max(250, env_int("KEYCHECK_DB_LOCK_TIMEOUT_MS", 2000))
                db.conn.execute("SELECT set_config('statement_timeout', ?, false)", (f"{statement_timeout_ms}ms",))
                db.conn.execute("SELECT set_config('lock_timeout', ?, false)", (f"{lock_timeout_ms}ms",))
            ok = db.record_keycheck_result(
                service=service,
                status=status,
                status_group=keycheck_status_group(status),
                checked_at=result.get("checked_at") or now_iso(),
                key_hash=sha256_text(key),
                secret_hash=secret_hash,
                key_masked=result.get("key_masked") or mask_secret(key),
                detector_name=detector or get_detector_name(finding or {}) or result.get("detector") or "",
                message=result_message(result),
                metadata=metadata,
                link_findings=truthy_env("KEYCHECK_DB_LINK_FINDINGS", False),
                source_line=source,
                detector_secret_hash=detector_secret_hash,
            )
            if not ok:
                raise RuntimeError("keycheck DB write returned false")
        finally:
            db.close()
        record_db_write_metric(service, True, result_source, status)
        return True
    except Exception as exc:
        print(f"Warning: unable to record keycheck result in DB: {exc}")
        record_db_write_metric(service, False, result_source, status, str(exc))
        return False


def record_db_write_metric(service, ok, result_source, status, error=""):
    output_dir = os.getenv("KEYCHECK_OUTPUT_DIR") or service_output_dir(service or keycheck_service_name() or "default")
    require_private_directory(output_dir, create=True)
    payload = {
        "service": service or keycheck_service_name(),
        "ok": bool(ok),
        "result_source": result_source or "api_check",
        "status": status,
        "error": str(error or "")[:500],
        "created_at": now_iso(),
    }
    try:
        append_jsonl(os.path.join(output_dir, "db_write_metrics.jsonl"), payload)
        if not ok:
            with private_append_writer(os.path.join(output_dir, "db_write_failures.log")) as f:
                f.write(f"{payload['created_at']}\t{payload['status']}\t{payload['result_source']}\t{payload['error']}\n")
    except OSError:
        pass


def normalize_status_key(line):
    line = line.strip()
    if not line:
        return None
    if "\t" in line:
        return line.split("\t", 1)[0].strip()
    return line.strip()


def iter_bounded_text_lines(path):
    if keycheck_input_mode() == 'postgres':
        return
    max_bytes = max(1, env_int('KEYCHECK_INPUT_LIST_MAX_BYTES', 32 * 1024 * 1024))
    max_items = max(1, env_int('KEYCHECK_INPUT_LIST_MAX_ITEMS', 100000))
    max_line_bytes = status_projection_line_max_bytes()
    if not os.path.exists(path):
        return
    if os.path.getsize(path) > max_bytes:
        raise RuntimeError(f'keycheck input list exceeds its aggregate byte bound: {path}')
    with open(path, 'rb') as handle:
        for index, raw_line in enumerate(handle, 1):
            if index > max_items:
                raise RuntimeError(f'keycheck input list exceeds its item bound: {path}')
            if len(raw_line) > max_line_bytes:
                raise RuntimeError(f'keycheck input list line exceeds its byte bound: {path}:{index}')
            yield raw_line.decode('utf-8', errors='replace')


def load_keys_from_file(path):
    if keycheck_input_mode() == 'postgres':
        return set()
    if not os.path.exists(path):
        return set()
    return {key for key in (normalize_status_key(line) for line in iter_bounded_text_lines(path)) if key}


def load_checked_statuses(path):
    statuses = {}
    if keycheck_input_mode() == 'postgres':
        return statuses
    if not os.path.exists(path):
        return statuses
    for line in iter_bounded_text_lines(path):
        parts = line.rstrip("\n").split("\t")
        if not parts or not parts[0]:
            continue
        statuses[parts[0]] = parts[1] if len(parts) > 1 else "UNKNOWN"
    return statuses


def ensure_output_files(paths):
    if keycheck_input_mode() == 'postgres':
        return
    for path in paths:
        parent = os.path.dirname(path)
        if parent:
            require_private_directory(parent, create=True)
        if os.path.lexists(path):
            require_private_file(path)
            continue
        with private_atomic_writer(path, binary=True, suffix=".empty.tmp"):
            pass


def remove_key_from_files(key, paths):
    if keycheck_input_mode() == 'postgres':
        return
    for path in paths:
        if not os.path.exists(path):
            continue
        lock_path = f"{path}.lock"
        lock = acquire_file_lock(lock_path, timeout_sec=30)
        try:
            require_private_file(path)
            with open(path, "r", encoding="utf-8") as f:
                lines = f.readlines()
            with private_atomic_writer(path) as f:
                for line in lines:
                    if normalize_status_key(line) != key:
                        f.write(line)
        finally:
            release_file_lock(lock, lock_path)


def append_status(path, key, status, message="", extra=""):
    if keycheck_input_mode() == 'postgres':
        return 'postgres-authoritative'
    message = str(message or "").replace("\n", " ")[:1000]
    extra = str(extra or "").replace("\n", " ")[:1000]
    require_private_directory(os.path.dirname(os.path.abspath(path)), create=True)
    if os.path.lexists(path):
        reject_reparse_components(path)
    lock_path = f"{path}.lock"
    lock = acquire_file_lock(lock_path, timeout_sec=30)
    try:
        with private_append_writer(path) as f:
            f.write(f"{key}\t{status}\t{message}\t{extra}\n")
    finally:
        release_file_lock(lock, lock_path)


def append_checked(path, key, status):
    if keycheck_input_mode() == 'postgres':
        return 'postgres-authoritative'
    require_private_directory(os.path.dirname(os.path.abspath(path)), create=True)
    if os.path.lexists(path):
        reject_reparse_components(path)
    lock_path = f"{path}.lock"
    lock = acquire_file_lock(lock_path, timeout_sec=30)
    try:
        with private_append_writer(path) as f:
            f.write(f"{key}\t{status}\t{now_iso()}\n")
    finally:
        release_file_lock(lock, lock_path)


def append_jsonl(path, payload):
    if keycheck_input_mode() == 'postgres':
        return 'postgres-authoritative'
    parent = os.path.dirname(path)
    if parent:
        require_private_directory(parent, create=True)
    if os.path.lexists(path):
        reject_reparse_components(path)
    fd = None
    lock_path = f"{path}.lock"
    try:
        fd = acquire_file_lock(lock_path, env_int("KEYCHECK_JSONL_LOCK_STALE_SEC", 300))
        repair_keycheck_jsonl_tail(path)
        reconcile_keycheck_jsonl_segments(path)
        if should_rotate_jsonl(path):
            max_mb = env_int("KEYCHECK_RESULTS_MAX_MB", 32)
            rotate_jsonl_if_needed(path, max_mb * 1024 * 1024)
        with private_append_writer(path) as f:
            f.write(json.dumps(payload, ensure_ascii=False, default=str) + "\n")
    finally:
        if fd is not None:
            release_file_lock(fd, lock_path)


def load_known_keys(checked_file, status_files):
    return set(load_known_statuses(checked_file, status_files).keys())


def normalize_status_files(status_files):
    output = {}
    if not status_files:
        return output
    if isinstance(status_files, dict):
        for key, value in status_files.items():
            key_text = str(key)
            value_text = str(value)
            if key_text.lower().endswith(".txt") or os.sep in key_text or "/" in key_text:
                output[value_text.upper()] = key_text
            else:
                output[key_text.upper()] = value_text
    else:
        for path in status_files:
            output["UNKNOWN"] = str(path)
    return output


def status_transaction_journal_path(checked_file):
    return os.path.join(
        os.path.dirname(os.path.abspath(checked_file)),
        STATUS_TRANSACTION_JOURNAL_FILENAME,
    )


def status_transaction_lock_path(checked_file):
    return os.path.join(
        os.path.dirname(os.path.abspath(checked_file)),
        STATUS_TRANSACTION_LOCK_FILENAME,
    )


def _status_transaction_layout(checked_file, status_files):
    checked_file = os.path.abspath(checked_file)
    output_dir = os.path.dirname(checked_file)
    require_private_directory(output_dir, create=True)
    normalized = {
        str(status or "").strip().upper(): os.path.abspath(path)
        for status, path in normalize_status_files(status_files).items()
    }
    if not normalized or any(
        not re.fullmatch(r"[A-Z][A-Z0-9_]{0,127}", status)
        for status in normalized
    ):
        raise ValueError("status transaction requires named status files")
    status_paths = list(dict.fromkeys(
        os.path.abspath(path) for path in normalized.values()
    ))
    paths = [checked_file, *status_paths]
    if any(os.path.normcase(os.path.dirname(path)) != os.path.normcase(output_dir) for path in paths):
        raise ValueError("status transaction files must share one service directory")
    if os.path.normcase(checked_file) in {os.path.normcase(path) for path in status_paths}:
        raise ValueError("status transaction checked and status paths must be distinct")
    ensure_output_files(paths)
    return checked_file, normalized


def _status_transaction_key(key):
    key = str(key or "")
    if not key or any(character in key for character in ("\x00", "\t", "\r", "\n")):
        raise ValueError("status transaction key is empty or contains a control separator")
    if len(key.encode("utf-8", errors="strict")) > status_projection_line_max_bytes():
        raise ValueError("status transaction key exceeds its byte bound")
    return key


def _status_transaction_field(value):
    return str(value or "").replace("\r", " ").replace("\n", " ")[:1000]


def _status_transaction_line(value, name):
    if not isinstance(value, str):
        raise ValueError(f"status transaction {name} must be text")
    if not value.endswith("\n") or "\n" in value[:-1] or "\r" in value or "\x00" in value:
        raise ValueError(f"status transaction {name} must be exactly one LF-terminated line")
    if len(value.encode("utf-8", errors="strict")) > status_projection_line_max_bytes():
        raise ValueError(f"status transaction {name} exceeds its byte bound")
    return value


def _status_projection_line_matches_key(line, key):
    value = str(line or "").rstrip("\r\n")
    if value == key:
        return True
    if not value.startswith(key):
        return False
    suffix = value[len(key):]
    return suffix.startswith("\t") or suffix.startswith(":")


def _validate_status_projection_line(line, key):
    line = _status_transaction_line(line, "status line")
    if not _status_projection_line_matches_key(line, key):
        raise ValueError("status transaction status line does not contain its exact key")
    return line


def _validate_checked_projection_line(line, key, status):
    line = _status_transaction_line(line, "checked line")
    parts = line[:-1].split("\t")
    if len(parts) < 2 or parts[0] != key or parts[1] != status:
        raise ValueError("status transaction checked line does not match its key and status")
    return line


def _new_status_transaction(
    key,
    status,
    status_path,
    message,
    extra,
    status_line=None,
    checked_line=None,
):
    key = _status_transaction_key(key)
    status = str(status or "UNKNOWN").strip().upper()
    message = _status_transaction_field(message)
    extra = _status_transaction_field(extra)
    checked_at = now_iso()
    status_line = _validate_status_projection_line(
        status_line if status_line is not None else f"{key}\t{status}\t{message}\t{extra}\n",
        key,
    )
    checked_line = _validate_checked_projection_line(
        checked_line if checked_line is not None else f"{key}\t{status}\t{checked_at}\n",
        key,
        status,
    )
    transaction = {
        "version": 2,
        "transaction_id": uuid.uuid4().hex,
        "key": key,
        "status": status,
        "status_file": os.path.basename(status_path),
        "message": message,
        "extra": extra,
        "checked_at": checked_at,
        "status_line": status_line,
        "checked_line": checked_line,
    }
    encoded = json.dumps(transaction, ensure_ascii=True, sort_keys=True, separators=(",", ":")).encode("utf-8")
    if len(encoded) + 1 > status_transaction_journal_max_bytes():
        raise ValueError("status transaction journal exceeds its byte bound")
    return transaction


def _validate_status_transaction(transaction, status_files):
    if not isinstance(transaction, dict) or type(transaction.get("version")) is not int or transaction["version"] not in (1, 2):
        raise RuntimeError("invalid status transaction journal version")
    version = transaction["version"]
    transaction_id = transaction.get("transaction_id")
    if not isinstance(transaction_id, str) or not re.fullmatch(r"[a-f0-9]{32}", transaction_id):
        raise RuntimeError("invalid status transaction identity")
    raw_key = transaction.get("key")
    if not isinstance(raw_key, str):
        raise RuntimeError("invalid status transaction key")
    try:
        key = _status_transaction_key(raw_key)
    except (UnicodeError, ValueError) as exc:
        raise RuntimeError(f"invalid status transaction key: {exc}") from exc
    raw_status = transaction.get("status")
    if not isinstance(raw_status, str) or raw_status != raw_status.strip().upper():
        raise RuntimeError("status transaction status is not canonical")
    status = raw_status
    if status not in status_files:
        raise RuntimeError("status transaction targets an unknown projection")
    status_file = transaction.get("status_file")
    if not isinstance(status_file, str) or os.path.normcase(status_file) != os.path.normcase(
        os.path.basename(status_files[status])
    ):
        raise RuntimeError("status transaction target does not match its projection")
    message = _status_transaction_field(transaction.get("message"))
    extra = _status_transaction_field(transaction.get("extra"))
    if message != transaction.get("message") or extra != transaction.get("extra"):
        raise RuntimeError("status transaction fields are not canonical")
    checked_at = transaction.get("checked_at")
    if not isinstance(checked_at, str):
        raise RuntimeError("invalid status transaction timestamp")
    if not checked_at or len(checked_at) > 64 or any(character in checked_at for character in ("\t", "\r", "\n")):
        raise RuntimeError("invalid status transaction timestamp")
    expected_status_line = f"{key}\t{status}\t{message}\t{extra}\n"
    expected_checked_line = f"{key}\t{status}\t{checked_at}\n"
    try:
        if version == 1:
            if transaction.get("status_line") != expected_status_line or transaction.get("checked_line") != expected_checked_line:
                raise RuntimeError("status transaction projection data is inconsistent")
            _validate_status_projection_line(transaction["status_line"], key)
            _validate_checked_projection_line(transaction["checked_line"], key, status)
        else:
            _validate_status_projection_line(transaction.get("status_line"), key)
            _validate_checked_projection_line(transaction.get("checked_line"), key, status)
    except (TypeError, UnicodeError, ValueError) as exc:
        raise RuntimeError(f"invalid status transaction projection data: {exc}") from exc
    encoded = json.dumps(transaction, ensure_ascii=True, sort_keys=True, separators=(",", ":")).encode("utf-8")
    if len(encoded) + 1 > status_transaction_journal_max_bytes():
        raise RuntimeError("status transaction journal exceeds its byte bound")
    return transaction


def _rewrite_status_projection(path, key, replacement_line=None):
    require_private_file(path)
    with private_atomic_writer(path, suffix=".status.tmp") as output:
            for line in iter_bounded_text_lines(path):
                if _status_projection_line_matches_key(line, key):
                    continue
                output.write(line if line.endswith("\n") else line + "\n")
            if replacement_line is not None:
                output.write(replacement_line)


def publish_status_transaction_target(transaction, status_files):
    status_files = normalize_status_files(status_files)
    _rewrite_status_projection(
        status_files[transaction["status"]],
        transaction["key"],
        transaction["status_line"],
    )


def publish_status_transaction_checked(transaction, checked_file):
    _rewrite_status_projection(
        checked_file,
        transaction["key"],
        transaction["checked_line"],
    )


def remove_status_transaction_old_copies(transaction, status_files):
    status_files = normalize_status_files(status_files)
    target_path = os.path.normcase(os.path.abspath(status_files[transaction["status"]]))
    visited = {target_path}
    for _, path in sorted(status_files.items()):
        normalized_path = os.path.normcase(os.path.abspath(path))
        if normalized_path in visited:
            continue
        visited.add(normalized_path)
        _rewrite_status_projection(path, transaction["key"])


def delete_status_transaction_journal(journal_path):
    durable_unlink(journal_path)


def _apply_status_transaction(transaction, checked_file, status_files, journal_path):
    transaction = _validate_status_transaction(transaction, status_files)
    publish_status_transaction_target(transaction, status_files)
    publish_status_transaction_checked(transaction, checked_file)
    remove_status_transaction_old_copies(transaction, status_files)
    delete_status_transaction_journal(journal_path)


def _recover_status_transaction_locked(checked_file, status_files, journal_path):
    if not os.path.lexists(journal_path):
        return False
    transaction = read_private_json(journal_path, max_bytes=status_transaction_journal_max_bytes())
    _apply_status_transaction(transaction, checked_file, status_files, journal_path)
    return True


def recover_status_transaction(checked_file, status_files):
    if keycheck_input_mode() == 'postgres':
        return False
    journal_path = status_transaction_journal_path(checked_file)
    if not os.path.lexists(journal_path):
        return False
    checked_file, status_files = _status_transaction_layout(checked_file, status_files)
    journal_path = status_transaction_journal_path(checked_file)
    lock_path = status_transaction_lock_path(checked_file)
    timeout = min(120, max(1, env_int("KEYCHECK_STATUS_TRANSACTION_LOCK_TIMEOUT_SEC", 30)))
    lock = acquire_file_lock(lock_path, timeout_sec=timeout)
    try:
        return _recover_status_transaction_locked(checked_file, status_files, journal_path)
    finally:
        release_file_lock(lock, lock_path)


def commit_status_transaction(
    checked_file,
    status_files,
    key,
    status,
    message="",
    extra="",
    status_line=None,
    checked_line=None,
):
    if keycheck_input_mode() == 'postgres':
        return 'postgres-authoritative'
    checked_file, status_files = _status_transaction_layout(checked_file, status_files)
    status = str(status or "UNKNOWN").strip().upper()
    if status not in status_files:
        status = "UNKNOWN"
    if status not in status_files:
        raise ValueError("status transaction has no UNKNOWN projection")
    journal_path = status_transaction_journal_path(checked_file)
    lock_path = status_transaction_lock_path(checked_file)
    timeout = min(120, max(1, env_int("KEYCHECK_STATUS_TRANSACTION_LOCK_TIMEOUT_SEC", 30)))
    lock = acquire_file_lock(lock_path, timeout_sec=timeout)
    try:
        _recover_status_transaction_locked(checked_file, status_files, journal_path)
        transaction = _new_status_transaction(
            key,
            status,
            status_files[status],
            message,
            extra,
            status_line,
            checked_line,
        )
        encoded = json.dumps(
            transaction, ensure_ascii=True, sort_keys=True, separators=(",", ":"),
        ).encode("utf-8") + b"\n"
        if len(encoded) > status_transaction_journal_max_bytes():
            raise ValueError("status transaction journal exceeds its byte bound")
        with private_atomic_writer(journal_path, binary=True, suffix=".journal.tmp") as handle:
            handle.write(encoded)
        _apply_status_transaction(transaction, checked_file, status_files, journal_path)
        return transaction["transaction_id"]
    finally:
        release_file_lock(lock, lock_path)


def load_known_statuses(checked_file, status_files):
    if keycheck_input_mode() == 'postgres':
        return {}
    statuses = load_checked_statuses(checked_file)
    for status, path in normalize_status_files(status_files).items():
        for key in load_keys_from_file(path):
            statuses.setdefault(key, status)
    return statuses


def cached_occurrence_identity(service, key, source, finding):
    finding_hash = finding_secret_hash(finding)
    if not finding_hash and isinstance(finding, dict):
        finding_hash = sha256_text(json.dumps(finding, ensure_ascii=False, sort_keys=True, default=str))
    return sha256_text("|".join([str(service or ""), sha256_text(key), str(source or ""), finding_hash]))


def cached_occurrence_log_path(service=None):
    service = service or keycheck_service_name() or "default"
    output_dir = os.getenv("KEYCHECK_OUTPUT_DIR") or service_output_dir(service)
    require_private_directory(output_dir, create=True)
    return os.path.join(output_dir, "cached_occurrences.tsv")


CACHED_OCCURRENCE_CACHE = {}
CACHED_OCCURRENCE_ROWS = {}
CACHED_OCCURRENCE_SIGNATURES = {}


def cached_occurrence_limits():
    return {
        "items": max(1, env_int("KEYCHECK_CACHED_OCCURRENCE_MAX_ITEMS", 100000)),
        "bytes": max(1, env_int("KEYCHECK_CACHED_OCCURRENCE_MAX_BYTES", 16 * 1024 * 1024)),
        "ttl": max(60, env_int("KEYCHECK_CACHED_OCCURRENCE_TTL_SEC", 30 * 86400)),
    }


def _cached_occurrence_timestamp(value):
    try:
        parsed = datetime.fromisoformat(str(value or "").replace("Z", "+00:00"))
        if parsed.tzinfo is None:
            parsed = parsed.replace(tzinfo=timezone.utc)
        return parsed.timestamp()
    except (TypeError, ValueError):
        return 0.0


def _cached_occurrence_signature(path):
    try:
        details = os.stat(path)
        return (int(details.st_size), int(getattr(details, "st_mtime_ns", 0)))
    except OSError:
        return (0, 0)


def _load_bounded_cached_occurrences(path, limits):
    if not os.path.exists(path):
        return [], False
    size = os.path.getsize(path)
    start = max(0, size - limits["bytes"])
    rows = []
    truncated = start > 0
    cutoff = time.time() - limits["ttl"]
    expired = False
    with open(path, "rb") as handle:
        handle.seek(start)
        if start:
            handle.readline(limits["bytes"] + 1)
        payload = handle.read(limits["bytes"] + 1)
    if len(payload) > limits["bytes"]:
        payload = payload[-limits["bytes"]:]
        first_newline = payload.find(b"\n")
        payload = payload[first_newline + 1:] if first_newline >= 0 else b""
        truncated = True
    for raw_line in payload.splitlines():
        try:
            text = raw_line.decode("utf-8")
        except UnicodeDecodeError:
            truncated = True
            continue
        identity, separator, timestamp = text.partition("\t")
        identity = identity.strip()
        if not identity or not separator:
            truncated = True
            continue
        epoch = _cached_occurrence_timestamp(timestamp.strip())
        if epoch < cutoff:
            expired = True
            continue
        rows.append((identity, timestamp.strip(), epoch))
    deduped = {}
    for identity, timestamp, epoch in rows:
        deduped[identity] = (identity, timestamp, epoch)
    retained = sorted(deduped.values(), key=lambda item: item[2])[-limits["items"]:]
    return retained, truncated or expired or len(retained) != len(rows)


def _save_cached_occurrence_rows(path, rows, byte_limit):
    encoded = [f"{identity}\t{timestamp}\n".encode("utf-8") for identity, timestamp, _ in rows]
    total = sum(len(item) for item in encoded)
    remove = 0
    while remove < len(encoded) and total > byte_limit:
        total -= len(encoded[remove])
        remove += 1
    if remove:
        encoded = encoded[remove:]
        rows = rows[remove:]
    with private_atomic_writer(path, binary=True) as handle:
        for row in encoded:
            handle.write(row)
    return rows


def cached_occurrence_set(service=None):
    service = service or keycheck_service_name() or "default"
    path = cached_occurrence_log_path(service)
    signature = _cached_occurrence_signature(path)
    cached_rows = CACHED_OCCURRENCE_ROWS.get(service) or []
    cache_unexpired = not cached_rows or cached_rows[0][2] >= time.time() - cached_occurrence_limits()["ttl"]
    if service in CACHED_OCCURRENCE_CACHE and CACHED_OCCURRENCE_SIGNATURES.get(service) == signature and cache_unexpired:
        return CACHED_OCCURRENCE_CACHE[service]
    limits = cached_occurrence_limits()
    rows, needs_compaction = _load_bounded_cached_occurrences(path, limits)
    if needs_compaction:
        lock_path = f"{path}.lock"
        lock = acquire_file_lock(lock_path, timeout_sec=30)
        try:
            rows, _ = _load_bounded_cached_occurrences(path, limits)
            rows = _save_cached_occurrence_rows(path, rows, limits["bytes"])
        finally:
            release_file_lock(lock, lock_path)
        signature = _cached_occurrence_signature(path)
    seen = {identity for identity, _, _ in rows}
    CACHED_OCCURRENCE_CACHE[service] = seen
    CACHED_OCCURRENCE_ROWS[service] = rows
    CACHED_OCCURRENCE_SIGNATURES[service] = signature
    return seen


def cached_occurrence_seen(identity, service=None):
    return identity in cached_occurrence_set(service)


def mark_cached_occurrence_seen(identity, service=None):
    service = service or keycheck_service_name() or "default"
    path = cached_occurrence_log_path(service)
    lock_path = f"{path}.lock"
    lock = acquire_file_lock(lock_path, timeout_sec=30)
    try:
        limits = cached_occurrence_limits()
        signature = _cached_occurrence_signature(path)
        if CACHED_OCCURRENCE_SIGNATURES.get(service) == signature:
            rows = list(CACHED_OCCURRENCE_ROWS.get(service) or [])
            cutoff = time.time() - limits["ttl"]
            retained = [row for row in rows if row[2] >= cutoff]
            needs_compaction = len(retained) != len(rows)
            rows = retained
        else:
            rows, needs_compaction = _load_bounded_cached_occurrences(path, limits)
        seen = {value for value, _, _ in rows}
        if identity in seen:
            CACHED_OCCURRENCE_CACHE[service] = seen
            CACHED_OCCURRENCE_ROWS[service] = rows
            CACHED_OCCURRENCE_SIGNATURES[service] = signature
            return False
        timestamp = now_iso()
        rows.append((str(identity), timestamp, time.time()))
        line = f"{identity}\t{timestamp}\n".encode("utf-8")
        current_size = signature[0]
        if (
            needs_compaction
            or len(rows) > limits["items"]
            or current_size + len(line) > limits["bytes"]
        ):
            rows = rows[-limits["items"]:]
            rows = _save_cached_occurrence_rows(path, rows, limits["bytes"])
        else:
            with private_append_writer(path, binary=True) as handle:
                handle.write(line)
        seen = {value for value, _, _ in rows}
        CACHED_OCCURRENCE_CACHE[service] = seen
        CACHED_OCCURRENCE_ROWS[service] = rows
        CACHED_OCCURRENCE_SIGNATURES[service] = _cached_occurrence_signature(path)
        return True
    finally:
        release_file_lock(lock, lock_path)


def record_cached_keycheck_occurrence(service, key, status, source="", finding=None, detector=None):
    if keycheck_input_mode() == 'postgres':
        raise RuntimeError('PostgreSQL candidates cannot be completed from cached file state')
    if not key or not service or not status:
        return False
    identity = cached_occurrence_identity(service, key, source, finding)
    if cached_occurrence_seen(identity, service):
        return False
    finding = strip_finding_nearby_context(finding)
    secret_hash = finding_secret_hash(finding) or sha256_text(key)
    detector_secret_hash = finding_detector_secret_hash(finding)
    result = {
        "status": str(status).upper(),
        "result_source": "cached_status",
        "cached_status": True,
        "cached_occurrence_id": identity,
        "checked_at": now_iso(),
        "message": "cached status occurrence; provider API not called",
        "key_hash": sha256_text(key),
        "secret_hash": secret_hash,
        "detector_secret_hash": detector_secret_hash,
        "finding_uid": finding_uid(finding),
    }
    output_dir = os.getenv("KEYCHECK_OUTPUT_DIR") or service_output_dir(service)
    require_private_directory(output_dir, create=True)
    write_keycheck_event(
        service,
        os.path.join(output_dir, f"{service}Results.jsonl"),
        key,
        result,
        source,
        finding,
        detector or get_detector_name(finding or {}) or "",
        "cached_status",
    )
    mark_cached_occurrence_seen(identity, service)
    if not truthy_env("KEYCHECK_DB_INLINE", False):
        return True
    ok = record_validation_result(service, key, result, source, finding, detector)
    return ok


def should_skip_key(key, checked_statuses, known_keys, args, retry_statuses=None, service=None, source="", finding=None, detector=None, known_statuses=None):
    if keycheck_input_mode() == 'postgres':
        if not _ACTIVE_DB_CANDIDATE:
            raise RuntimeError('PostgreSQL provider probe has no active fenced candidate')
        return False
    attempted = getattr(args, "_keycheck_attempted_keys", None)
    if not isinstance(attempted, set):
        attempted = set()
        setattr(args, "_keycheck_attempted_keys", attempted)
    attempt_identity = (str(service or keycheck_service_name() or "default").lower(), key)
    if attempt_identity in attempted:
        if service and source:
            cached_status = checked_statuses.get(key) or (known_statuses or {}).get(key) or "UNKNOWN"
            record_cached_keycheck_occurrence(service, key, cached_status, source, finding, detector)
        return True

    if getattr(args, "recheck_all", False):
        attempted.add(attempt_identity)
        return False
    status = checked_statuses.get(key) or (known_statuses or {}).get(key)
    retry_statuses = retry_statuses or set()
    if status in retry_statuses:
        attempted.add(attempt_identity)
        return False
    if status is None and retry_statuses and key in known_keys:
        attempted.add(attempt_identity)
        return False
    skip = key in known_keys or key in checked_statuses
    if skip and service and source:
        cached_status = checked_statuses.get(key) or (known_statuses or {}).get(key) or "UNKNOWN"
        record_cached_keycheck_occurrence(service, key, cached_status, source, finding, detector)
    if not skip:
        attempted.add(attempt_identity)
    return skip


def load_proxies(proxy_file):
    if not os.path.exists(proxy_file):
        print(f"Info: {proxy_file} not found. Requests will go directly.")
        return None
    proxies = []
    with open(proxy_file, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                ip, port, login, password = line.split(":")
                proxy_url = f"http://{login}:{password}@{ip}:{port}"
                proxies.append({"http": proxy_url, "https": proxy_url})
            except ValueError:
                print(f"Warning: bad proxy format: {line}. Skipping.")
    if not proxies:
        print(f"Warning: {proxy_file} is empty. Requests will go directly.")
        return None
    print(f"Loaded proxies: {len(proxies)}")
    return cycle(proxies)


def get_detector_name(data):
    if data.get("DetectorName"):
        return data.get("DetectorName")
    if data.get("detector"):
        return data.get("detector")
    finding = data.get("finding")
    if isinstance(finding, dict):
        return finding.get("DetectorName")
    return None


def get_raw_values(data):
    if data.get("DetectorName"):
        return data.get("Raw"), data.get("RawV2"), data
    if data.get("detector"):
        finding = data.get("finding") if isinstance(data.get("finding"), dict) else data
        return data.get("raw"), data.get("raw_v2"), finding
    finding = data.get("finding")
    if isinstance(finding, dict):
        return finding.get("Raw"), finding.get("RawV2"), finding
    return None, None, data


def iter_findings(input_file, detector_names):
    if keycheck_input_mode() == 'postgres':
        global _ACTIVE_DB_CANDIDATE
        service = keycheck_service_name()
        if not service:
            raise RuntimeError('PostgreSQL keycheck input mode requires KEYCHECK_SERVICE')
        owner = f'{service}:{os.getpid()}:{uuid.uuid4().hex}'
        detector_names = set(detector_names) if detector_names is not None else None
        slice_limit = max(0, env_int('KEYCHECK_PROVIDER_SLICE_KEYS', 0))
        yielded = 0
        try:
            while True:
                if slice_limit and yielded >= slice_limit:
                    return
                candidate = _postgres_candidate_db().claim_keycheck_candidate(
                    service, owner,
                    lease_seconds=max(30, env_int('KEYCHECK_CANDIDATE_LEASE_SEC', 300)),
                    result_projection_reserve_bytes=max(
                        3 * 1024 * 1024,
                        env_int('KEYCHECK_RESULT_PROJECTION_RESERVE_BYTES', 3 * 1024 * 1024),
                    ),
                    projection_max_items=max(
                        1, env_int('KEYCHECK_PROJECTION_MAX_ITEMS', 10000),
                    ),
                    projection_max_bytes=max(
                        3 * 1024 * 1024,
                        env_int('KEYCHECK_PROJECTION_MAX_BYTES', 2 * 1024 * 1024 * 1024),
                    ),
                )
                if not candidate:
                    return
                if candidate.get('_capacity_blocked'):
                    print(
                        'PostgreSQL keycheck blocked: projection capacity is saturated',
                        flush=True,
                    )
                    raise SystemExit(KEYCHECK_CAPACITY_BLOCKED_EXIT)
                _ACTIVE_DB_CANDIDATE = candidate
                yielded += 1
                finding = dict(candidate.get('finding') or {})
                candidate_metadata = safe_json_loads(candidate.get('metadata_json')) or {}
                if not finding:
                    finding = {
                        'DetectorName': candidate.get('detector_name') or candidate_metadata.get('detector_name'),
                        'Raw': candidate.get('secret_text') or '',
                        'RawV2': candidate_metadata.get('raw_v2') or candidate.get('secret_json') or '',
                    }
                detector = finding.get('DetectorName') or candidate.get('detector_name')
                candidate_kind = candidate.get('candidate_kind') or ''
                structured_route = candidate_kind not in ('', 'provider_key')
                provider_hint = str(candidate_metadata.get('provider_hint') or '').lower()
                persisted_route = (
                    provider_hint == service
                    or (
                        service == 'provider_resolver'
                        and provider_hint in (
                            AMBIGUOUS_QWEN_DEEPSEEK_HINT, AMBIGUOUS_GENERIC_SK_HINT,
                        )
                    )
                )
                if (
                    detector_names is not None and detector not in detector_names
                    and not structured_route and not persisted_route
                ):
                    _postgres_candidate_db().quarantine_keycheck_candidate(
                        candidate['id'], candidate['lease_token'],
                        'candidate_provider_route_mismatch',
                        'candidate detector does not match provider route',
                    )
                    _ACTIVE_DB_CANDIDATE = None
                    continue
                candidate_db = _postgres_candidate_db()
                cached_completed = False
                cached_state_retry_exhausted = False
                for cached_attempt in range(2):
                    cached = candidate_db.keycheck_candidate_cached_status(
                        candidate['id'], candidate['lease_token'],
                    )
                    if cached is None:
                        raise RuntimeError('PostgreSQL cached-status lookup lost its candidate fence')
                    if cached.get('probe_required'):
                        break
                    event_id = sha256_text('|'.join((
                        'truf-keycheck-cached-event-v1', str(candidate['id']),
                    )))
                    cached_metadata = {
                        'cached_status': True,
                        'cached_state_version': cached['state_version'],
                        'cached_result_id': cached['last_result_id'],
                        'cached_checked_at': cached['checked_at'],
                        'cached_result_source': cached['result_source'],
                    }
                    outcome = candidate_db.complete_keycheck_candidate(
                        candidate['id'], candidate['lease_token'], event_id,
                        cached['status'], cached['status_group'], checked_at=now_iso(),
                        message='cached current status occurrence; provider API not called',
                        metadata=cached_metadata, result_source='cached_status',
                        cached_state_version=cached['state_version'],
                        cached_last_result_id=cached['last_result_id'],
                    )
                    if outcome and outcome.get('completed'):
                        candidate['_completed'] = True
                        _ACTIVE_DB_CANDIDATE = None
                        cached_completed = True
                        break
                    if outcome and outcome.get('cached_state_changed'):
                        cached_state_retry_exhausted = cached_attempt == 1
                        continue
                    if outcome and outcome.get('probe_required'):
                        break
                    raise RuntimeError('PostgreSQL cached-status completion lost its candidate fence')
                if cached_completed:
                    continue
                if cached_state_retry_exhausted:
                    candidate_db.defer_keycheck_candidate(
                        candidate['id'], candidate['lease_token'],
                        'current status changed during cached occurrence completion', 1,
                    )
                    _ACTIVE_DB_CANDIDATE = None
                    continue
                raw = finding.get('Raw')
                raw_v2 = finding.get('RawV2')
                if not raw and candidate.get('secret_text'):
                    raw = candidate['secret_text']
                if not raw_v2 and candidate.get('secret_json'):
                    raw_v2 = candidate['secret_json']
                yield {
                    'line': candidate['id'],
                    'detector': detector,
                    'raw': raw or '',
                    'raw_v2': raw_v2 or '',
                    'finding': finding,
                    'source': f'postgres:keycheck_candidates:{candidate["id"]}',
                    'candidate_id': candidate['id'],
                    'candidate_kind': candidate_kind,
                    'credential_secret_text': candidate.get('secret_text') or '',
                    'credential_secret_json': candidate.get('secret_json') or '',
                    'credential_endpoint': candidate.get('endpoint') or '',
                    'credential_principal': candidate.get('principal') or '',
                    'candidate_metadata': candidate_metadata,
                }
                if not candidate.get('_completed'):
                    reason = 'provider iterator resumed without a committed result'
                    max_unconsumed = max(
                        1, env_int('KEYCHECK_CANDIDATE_MAX_UNCONSUMED_ATTEMPTS', 3),
                    )
                    if int(candidate.get('attempts') or 0) >= max_unconsumed:
                        _postgres_candidate_db().quarantine_keycheck_candidate(
                            candidate['id'], candidate['lease_token'],
                            'provider_candidate_unconsumed', reason,
                        )
                    else:
                        _postgres_candidate_db().defer_keycheck_candidate(
                            candidate['id'], candidate['lease_token'], reason, 60,
                        )
                _ACTIVE_DB_CANDIDATE = None
        finally:
            candidate = _ACTIVE_DB_CANDIDATE
            if candidate and not candidate.get('_completed'):
                try:
                    _postgres_candidate_db().defer_keycheck_candidate(
                        candidate['id'], candidate['lease_token'],
                        'provider iterator closed without a committed result', 60,
                    )
                finally:
                    _ACTIVE_DB_CANDIDATE = None
        return
    detector_names = set(detector_names) if detector_names is not None else None
    for item in iter_jsonl_input(input_file):
        data = item.get("data") or {}
        detector = get_detector_name(data)
        if detector_names is not None and detector not in detector_names:
            continue
        raw, raw_v2, finding = get_raw_values(data)
        yield {
            "line": item.get("line_offset"),
            "detector": detector,
            "raw": raw or "",
            "raw_v2": raw_v2 or "",
            "finding": finding,
            "source": item.get("source") or input_file,
        }


def read_plain_keys(paths, regex):
    if keycheck_input_mode() == 'postgres':
        return
    seen = set()
    for path in paths or []:
        if not os.path.exists(path):
            continue
        with open(path, "r", encoding="utf-8") as f:
            content = f.read()
        for match in regex.findall(content):
            key = match if isinstance(match, str) else ":".join(match)
            if key not in seen:
                seen.add(key)
                yield {"source": path, "key": key}


def request_error_message(response):
    try:
        payload = response.json()
    except ValueError:
        return response.text[:1000]
    return json.dumps(payload, ensure_ascii=False)[:1000]


def classify_common_http_status(status_code):
    if status_code in (401, 403):
        return "DEAD"
    if status_code == 429:
        return "LIMITED"
    if 500 <= status_code <= 599:
        return "NETWORK"
    return "UNKNOWN"


def requests_network_error(exc):
    return isinstance(exc, requests.RequestException)