extract-knowledge.py

#!/usr/bin/env python3
"""
extract-knowledge.py — Extract structured knowledge from session checkpoints

Parses checkpoint sections to identify and categorize:
  - Patterns: Reusable coding/architecture best practices
  - Mistakes: Errors made and lessons learned
  - Decisions: Technical choices and their rationale
  - Tools: Tool configurations and usage notes

Usage:
    python extract-knowledge.py                # Extract from all checkpoints
    python extract-knowledge.py --stats        # Show extraction statistics
    python extract-knowledge.py --list         # List all extracted entries
    python extract-knowledge.py --category mistakes  # Show specific category

Cross-platform: Windows, macOS, Linux. Pure Python stdlib.
"""

import hashlib
import json
import os
import re
import sqlite3
import subprocess
import sys
import time
from datetime import datetime
from pathlib import Path

# Fix Windows console encoding for Unicode output
if os.name == "nt":
    try:
        sys.stdout.reconfigure(encoding="utf-8", errors="replace")
        sys.stderr.reconfigure(encoding="utf-8", errors="replace")
    except Exception:
        pass

SESSION_STATE = Path.home() / ".copilot" / "session-state"
DB_PATH = Path(os.environ.get("SK_DB_PATH", str(SESSION_STATE / "knowledge.db"))).expanduser()


def _emit_knowledge_event_fail_open(event_type: str, data: dict) -> None:
    try:
        events_script = Path(__file__).with_name("events.py")
        if not events_script.is_file():
            return
        subprocess.call(
            [
                sys.executable,
                str(events_script),
                "append",
                event_type,
                "--data",
                json.dumps(data, ensure_ascii=False),
            ],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            timeout=5,
        )
    except Exception:
        return


def _stable_sha256(*parts) -> str:
    payload = "\0".join("" if p is None else str(p) for p in parts)
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()


def _knowledge_stable_id(session_id: str, category: str, title: str, topic_key: str) -> str:
    return _stable_sha256("knowledge", session_id, category, title or "", topic_key or "")


def _knowledge_relation_stable_id(source_stable_id: str, target_stable_id: str, relation_type: str) -> str:
    return _stable_sha256(
        "knowledge_relation",
        source_stable_id or "",
        target_stable_id or "",
        relation_type or "",
    )


def _utc_now() -> str:
    return datetime.utcnow().replace(microsecond=0).isoformat() + "Z"


def _default_local_replica_id() -> str:
    host = os.environ.get("HOSTNAME") or os.environ.get("COMPUTERNAME") or ""
    user = os.environ.get("USER") or os.environ.get("USERNAME") or ""
    return f"replica-{_stable_sha256('local-replica', host, user, str(Path.home()))[:16]}"


def _get_local_replica_id(db: sqlite3.Connection) -> str:
    try:
        row = db.execute("SELECT value FROM sync_state WHERE key='local_replica_id'").fetchone()
        current = str(row[0]) if row and row[0] else ""
        if current and current != "local":
            return current
        replica_id = _default_local_replica_id()
        db.execute(
            """
            INSERT INTO sync_state (key, value)
            VALUES ('local_replica_id', ?)
            ON CONFLICT(key) DO UPDATE SET
                value = excluded.value,
                updated_at = datetime('now')
        """,
            (replica_id,),
        )
        db.execute(
            """
            INSERT INTO sync_metadata (key, value)
            VALUES ('local_replica_id', ?)
            ON CONFLICT(key) DO UPDATE SET
                value = excluded.value,
                updated_at = datetime('now')
        """,
            (replica_id,),
        )
        return replica_id
    except Exception:
        return ""


def _enqueue_sync_op_fail_open(
    db: sqlite3.Connection,
    table_name: str,
    row_stable_id: str,
    row_payload: dict,
    op_type: str = "upsert",
):
    try:
        tools_dir = str(Path(__file__).resolve().parent)
        if tools_dir not in sys.path:
            sys.path.insert(0, tools_dir)
        from sync_enqueue import enqueue_sync_op_fail_open

        enqueue_sync_op_fail_open(db, table_name, row_stable_id, row_payload, op_type)
    except Exception:
        return


def _seed_sync_table_policies(db: sqlite3.Connection):
    rows = [
        ("sessions", "canonical", "id"),
        ("documents", "canonical", "stable_id"),
        ("sections", "canonical", "stable_id"),
        ("knowledge_entries", "canonical", "stable_id"),
        ("knowledge_relations", "canonical", "stable_id"),
        ("entity_relations", "canonical", "stable_id"),
        ("search_feedback", "canonical", "stable_id"),
        ("recall_events", "upload_only", ""),
        ("entry_recall_stats", "upload_only", ""),
        ("entry_recall_day_log", "upload_only", ""),
        ("entry_recall_query_log", "upload_only", ""),
        ("knowledge_fts", "local_only", ""),
        ("ke_fts", "local_only", ""),
        ("sessions_fts", "local_only", ""),
        ("event_offsets", "local_only", ""),
        ("embeddings", "local_only", ""),
        ("embedding_meta", "local_only", ""),
        ("tfidf_model", "local_only", ""),
        ("entry_concept_tags", "local_only", ""),
    ]
    policy_sql = db.execute(
        "SELECT sql FROM sqlite_master WHERE type='table' AND name='sync_table_policies'"
    ).fetchone()
    needs_rebuild = policy_sql and "upload_only" not in (policy_sql[0] or "")
    if needs_rebuild:
        db.executescript("""
            CREATE TABLE sync_table_policies_new (
                table_name TEXT PRIMARY KEY,
                sync_scope TEXT NOT NULL CHECK(sync_scope IN ('canonical', 'local_only', 'upload_only')),
                stable_id_column TEXT DEFAULT ''
            );
            INSERT INTO sync_table_policies_new (table_name, sync_scope, stable_id_column)
            SELECT table_name, sync_scope, COALESCE(stable_id_column, '')
            FROM sync_table_policies;
            DROP TABLE sync_table_policies;
            ALTER TABLE sync_table_policies_new RENAME TO sync_table_policies;
        """)
    db.executescript("""
        CREATE TABLE IF NOT EXISTS sync_metadata (
            key TEXT PRIMARY KEY,
            value TEXT NOT NULL,
            updated_at TEXT DEFAULT (datetime('now'))
        );
        CREATE TABLE IF NOT EXISTS sync_state (
            key TEXT PRIMARY KEY,
            value TEXT NOT NULL,
            updated_at TEXT DEFAULT (datetime('now'))
        );
        CREATE TABLE IF NOT EXISTS sync_txns (
            txn_id TEXT PRIMARY KEY,
            replica_id TEXT NOT NULL,
            status TEXT NOT NULL CHECK(status IN ('pending', 'committed', 'failed')),
            created_at TEXT NOT NULL,
            committed_at TEXT DEFAULT ''
        );
        CREATE TABLE IF NOT EXISTS sync_ops (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            txn_id TEXT NOT NULL,
            table_name TEXT NOT NULL,
            op_type TEXT NOT NULL CHECK(op_type IN ('insert', 'update', 'delete', 'upsert')),
            row_stable_id TEXT NOT NULL,
            row_payload TEXT NOT NULL,
            op_index INTEGER NOT NULL,
            created_at TEXT NOT NULL,
            UNIQUE(txn_id, op_index)
        );
        CREATE INDEX IF NOT EXISTS idx_sync_ops_txn ON sync_ops(txn_id);
        CREATE INDEX IF NOT EXISTS idx_sync_ops_table_row ON sync_ops(table_name, row_stable_id);
        CREATE TABLE IF NOT EXISTS sync_cursors (
            replica_id TEXT PRIMARY KEY,
            last_txn_id TEXT DEFAULT '',
            updated_at TEXT DEFAULT (datetime('now'))
        );
        CREATE TABLE IF NOT EXISTS sync_failures (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            txn_id TEXT DEFAULT '',
            table_name TEXT DEFAULT '',
            row_stable_id TEXT DEFAULT '',
            error_code TEXT DEFAULT '',
            error_message TEXT DEFAULT '',
            failed_at TEXT NOT NULL,
            retry_count INTEGER DEFAULT 0
        );
        CREATE INDEX IF NOT EXISTS idx_sync_failures_txn ON sync_failures(txn_id);
        CREATE TABLE IF NOT EXISTS sync_table_policies (
            table_name TEXT PRIMARY KEY,
            sync_scope TEXT NOT NULL CHECK(sync_scope IN ('canonical', 'local_only', 'upload_only')),
            stable_id_column TEXT DEFAULT ''
        );
    """)
    db.executemany(
        """
        INSERT INTO sync_table_policies (table_name, sync_scope, stable_id_column)
        VALUES (?, ?, ?)
        ON CONFLICT(table_name) DO UPDATE SET
            sync_scope = excluded.sync_scope,
            stable_id_column = excluded.stable_id_column
    """,
        rows,
    )
    db.execute("""
        INSERT OR IGNORE INTO sync_metadata (key, value)
        VALUES ('local_replica_id', 'local')
    """)
    db.execute("""
        INSERT OR IGNORE INTO sync_state (key, value)
        VALUES ('local_replica_id', 'local')
    """)


def _backfill_stable_ids(db: sqlite3.Connection):
    for row in db.execute("""
        SELECT id, session_id, category, title, COALESCE(topic_key, ''), COALESCE(stable_id, '')
        FROM knowledge_entries
    """).fetchall():
        ke_id, session_id, category, title, topic_key, existing = row
        stable = _knowledge_stable_id(session_id, category, title, topic_key)
        if existing != stable:
            db.execute("UPDATE knowledge_entries SET stable_id = ? WHERE id = ?", (stable, ke_id))
            _enqueue_sync_op_fail_open(
                db,
                "knowledge_entries",
                stable,
                {
                    "session_id": session_id,
                    "category": category,
                    "title": title,
                    "topic_key": topic_key,
                    "stable_id": stable,
                },
            )

    for row in db.execute("""
        SELECT id, subject, predicate, object, COALESCE(stable_id, '')
        FROM entity_relations
    """).fetchall():
        er_id, subject, predicate, obj, existing = row
        stable = _stable_sha256("entity_relation", subject or "", predicate or "", obj or "")
        if existing != stable:
            db.execute("UPDATE entity_relations SET stable_id = ? WHERE id = ?", (stable, er_id))
            _enqueue_sync_op_fail_open(
                db,
                "entity_relations",
                stable,
                {
                    "subject": subject,
                    "predicate": predicate,
                    "object": obj,
                    "stable_id": stable,
                },
            )

    for row in db.execute("""
        SELECT kr.id,
               kr.source_id,
               kr.target_id,
               kr.relation_type,
               COALESCE(kr.source_stable_id, ''),
               COALESCE(kr.target_stable_id, ''),
               COALESCE(kr.stable_id, ''),
               COALESCE(src.stable_id, ''),
               COALESCE(tgt.stable_id, '')
        FROM knowledge_relations kr
        LEFT JOIN knowledge_entries src ON kr.source_id = src.id
        LEFT JOIN knowledge_entries tgt ON kr.target_id = tgt.id
    """).fetchall():
        rel_id, _, _, relation_type, src_existing, tgt_existing, existing, src_stable, tgt_stable = row
        if not src_stable or not tgt_stable:
            continue
        stable = _knowledge_relation_stable_id(src_stable, tgt_stable, relation_type)
        if src_existing != src_stable or tgt_existing != tgt_stable or existing != stable:
            db.execute(
                """
                UPDATE knowledge_relations
                SET source_stable_id = ?, target_stable_id = ?, stable_id = ?
                WHERE id = ?
            """,
                (src_stable, tgt_stable, stable, rel_id),
            )


def _dedupe_stable_rows(db: sqlite3.Connection, table: str):
    if table not in {"knowledge_entries", "knowledge_relations", "entity_relations"}:
        return
    db.execute(
        """
        DELETE FROM {table}
        WHERE id IN (
            SELECT dupe.id
            FROM {table} dupe
            JOIN (
                SELECT stable_id, MIN(id) AS keep_id
                FROM {table}
                WHERE COALESCE(stable_id, '') != ''
                GROUP BY stable_id
                HAVING COUNT(*) > 1
            ) grouped ON grouped.stable_id = dupe.stable_id
            WHERE dupe.id != grouped.keep_id
        )
    """.replace("{table}", table)
    )


def _enforce_stable_id_uniqueness(db: sqlite3.Connection):
    has_table = lambda t: (
        db.execute(
            "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?",
            (t,),
        ).fetchone()
        is not None
    )
    for table, index_name in [
        ("knowledge_entries", "uq_knowledge_entries_stable_id"),
        ("knowledge_relations", "uq_knowledge_relations_stable_id"),
        ("entity_relations", "uq_entity_relations_stable_id"),
    ]:
        if has_table(table):
            _dedupe_stable_rows(db, table)
            db.execute(f"CREATE UNIQUE INDEX IF NOT EXISTS {index_name} ON {table}(stable_id)")


# ── WBS-013: Secret / injection scanner ──────────────────────────────────────
# Mirror of learn.py._INJECTION_PATTERNS (kept in sync with WBS-019 expansions).
# Applied before every DB insert in extract_from_sections() — fail-open (skip entry).


class _ContextAwareHexMatcher:
    """Matches long lowercase-hex strings only when NOT in a git/checksum context.

    Prevents false-positive blocking of 40-char git commit SHAs and 64-char
    SHA-256 checksums while still catching bare secret hex tokens that appear
    without any identifying reference keyword nearby.

    Duck-types the compiled regex interface: implements .search(text) returning
    a match object (or None) so it can be used transparently in pattern lists.
    """

    _HEX_RE = re.compile(r"(?<![A-Za-z0-9])([0-9a-f]{40,})(?![A-Za-z0-9])")
    _SAFE_CTX_RE = re.compile(r"(?i)\b(?:commit|sha\d*|hash|checksum|digest|fingerprint)\b")

    def search(self, text: str):
        """Return the first match for a secret-context hex run; None if all safe."""
        for m in self._HEX_RE.finditer(text):
            pre = text[max(0, m.start() - 100) : m.start()]
            if self._SAFE_CTX_RE.search(pre):
                continue
            return m
        return None


_EXTRACT_INJECTION_PATTERNS = [
    (
        re.compile(r"(?i)\bignore\s+(all\s+)?previous\s+instructions?\b"),
        "prompt injection: 'ignore previous instructions'",
    ),
    (re.compile(r"(?i)\byou\s+are\s+now\b"), "role hijacking: 'you are now'"),
    (re.compile(r"(?i)\bsystem\s*:\s*"), "role injection: 'system:' prefix"),
    (re.compile(r"(?i)\b(assistant|user|human)\s*:\s*"), "role injection: fake role prefix"),
    (re.compile(r"(?i)\bforget\s+(everything|all|your)\b"), "memory manipulation: 'forget everything'"),
    (re.compile(r"(?i)\bdo\s+not\s+follow\b"), "instruction override: 'do not follow'"),
    (
        re.compile(r"(?i)\b(api[_-]?key|secret[_-]?key|password|token)\s*[:=]\s*\S+"),
        "credential leak: API key/password/token",
    ),
    (re.compile(r"(?i)ssh-rsa\s+AAAA"), "credential leak: SSH public key"),
    (re.compile(r"(?i)-----BEGIN\s+(RSA\s+)?PRIVATE\s+KEY-----"), "credential leak: private key"),
    (re.compile(r"(?i)\beval\s*\("), "code injection: eval()"),
    (re.compile(r"(?i)\bexec\s*\("), "code injection: exec()"),
    (re.compile(r"[\u200b\u200c\u200d\u2060\ufeff]"), "invisible Unicode characters (zero-width)"),
    (re.compile(r"(?i)\bACT\s+AS\b"), "role hijacking: 'act as'"),
    (re.compile(r"(?i)\bpretend\s+(you\s+are|to\s+be)\b"), "role hijacking: 'pretend to be'"),
    (re.compile(r"(?i)\b(curl|wget|nc|ncat)\s+.*\|\s*(ba)?sh\b"), "remote code execution pattern"),
    (
        re.compile(r"\beyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\b"),
        "credential leak: JWT token",
    ),
    (
        re.compile(r"(?i)\bAuthorization\s*:\s*Bearer\s+\S{16,}"),
        "credential leak: Authorization Bearer token",
    ),
    (
        re.compile(r"\bAKIA[0-9A-Z]{16}\b"),
        "credential leak: AWS access key ID",
    ),
    (
        re.compile(r"(?i)\b(aws[_-]?secret[_-]?access[_-]?key|aws[_-]?secret)\s*[:=]\s*[A-Za-z0-9/+]{30,}"),
        "credential leak: AWS secret access key",
    ),
    # WBS-019: generic long hex token (context-aware — skips git commit SHAs and checksums)
    (
        _ContextAwareHexMatcher(),
        "credential leak: long hex secret/token",
    ),
]


def _scan_extract_chunk(title: str, content: str) -> str:
    """Scan title + content for injection/credential patterns.

    Returns the first matching description, or empty string if clean.
    Used by extract_from_sections() to skip unsafe chunks before DB insert.
    Fail-open: any exception returns empty string (scan miss is preferable to
    blocking extraction entirely).
    """
    try:
        text = f"{title}\n{content}"
        for pattern, description in _EXTRACT_INJECTION_PATTERNS:
            if pattern.search(text):
                return description
    except Exception:
        pass
    return ""


# ── Extraction patterns — regex + heuristics for each category ────────────────
MISTAKE_INDICATORS = [
    r"(?:mistake|error|bug|wrong|incorrect|broken|fail|crash|fix(?:ed)?)\b",
    r"(?:should\s+(?:have|not)|shouldn't|don't|avoid|never|careful)",
    r"(?:root\s+cause|caused\s+by|problem\s+was|issue\s+was)",
    r"(?:lỗi|sai|sửa|tránh|không\s+nên|nguyên\s+nhân)",
]

PATTERN_INDICATORS = [
    r"(?:always|must|should|convention|pattern|best\s+practice|rule)\b",
    r"(?:use\s+\w+\s+instead\s+of|prefer|recommend)",
    r"(?:standard|template|reusable|common\s+(?:pattern|style|approach))",
    r"(?:luôn|nên|quy\s+tắc|mẫu|chuẩn)",
    r"(?:good\s+practice|consistent(?:ly)?|enforce|ensure\s+(?:that|you))\b",
    r"\b(?:tip|guideline|approach|technique|strategy)\b",
    r"\b(?:make\s+sure|remember\s+to|keep\s+in\s+mind|note\s+that)\b",
]

DECISION_INDICATORS = [
    r"(?:chose|decided|selected|picked|went\s+with|opted)\b",
    r"(?:because|reason|rationale|trade-off|tradeoff)",
    r"(?:option\s+[A-C]|alternative|compared|versus|vs\.?)\b",
    r"(?:chọn|quyết\s+định|lý\s+do|so\s+sánh)",
]

TOOL_INDICATORS = [
    r"(?:install|configure|setup|version|upgrade|dependency)\b",
    r"(?:gradle|maven|docker|redis|postgres|spring\s+boot)\b",
    r"(?:JDK|SDK|IDE|VSCode|extension)\b",
    r"(?:cài|cấu\s+hình|phiên\s+bản|nâng\s+cấp)",
]

FEATURE_INDICATORS = [
    r"\b(?:implement(?:ed|ing)?|add(?:ed|ing)?|create(?:d|ing)?|build|built|develop(?:ed|ing)?)\b",
    r"(?:new\s+(?:feature|endpoint|handler|screen|component|API))\b",
    r"\b(?:feature|functionality|capability|user\s+story)\b",
    r"(?:thêm|tạo|xây\s+dựng|tính\s+năng|chức\s+năng)\b",
]

REFACTOR_INDICATORS = [
    r"\b(?:refactor|restructur|simplif|clean\s*up|extract|reorganiz)",
    r"\b(?:rename[ds]?|move[ds]?|split|merge[ds]?|consolidat|dedup)",
    r"\b(?:improv(?:e[ds]?|ing)|optimiz|reduc)",
    r"(?:tái\s+cấu\s+trúc|đơn\s+giản\s+hóa|tối\s+ưu)",
]

DISCOVERY_INDICATORS = [
    r"\b(?:discover|found|learn|realiz|notic|observ)",
    r"\b(?:turns\s+out|apparently|actually|interesting)\b",
    r"\b(?:TIL|insight|understanding|revelation)\b",
    r"(?:phát\s+hiện|nhận\s+ra|hiểu|thấy\s+rằng)",
]

# Error type classification patterns for auto-detection
ERROR_TYPE_PATTERNS = {
    "syntax": [
        r"(?:syntax\s*error|parse\s*error|unexpected\s*token|unterminated)",
        r"(?:indentation|missing\s*(?:bracket|paren|semicolon|colon|comma))",
        r"(?:SyntaxError|ParseError|invalid\s*syntax)",
    ],
    "runtime": [
        r"(?:runtime\s*error|exception|traceback|stack\s*trace)",
        r"(?:TypeError|ValueError|AttributeError|KeyError|IndexError|NameError)",
        r"(?:NullPointerException|segfault|segmentation\s*fault|SIGSEGV)",
        r"(?:crash(?:ed|es|ing)?|abort|panic|unhandled)",
    ],
    "logic": [
        r"(?:logic\s*error|wrong\s*(?:result|output|behavior|value))",
        r"(?:incorrect|unexpected\s*(?:result|output|behavior))",
        r"(?:off-by-one|race\s*condition|deadlock|infinite\s*loop)",
    ],
    "timeout": [
        r"(?:timeout|timed?\s*out|hang(?:s|ing|ed)?|stuck|frozen)",
        r"(?:too\s*(?:slow|long)|deadline\s*exceeded|connection\s*timeout)",
    ],
    "permission": [
        r"(?:permission\s*denied|access\s*denied|unauthorized|forbidden)",
        r"(?:EACCES|EPERM|403|401|authentication\s*fail)",
    ],
    "build": [
        r"(?:build\s*(?:fail|error)|compilation\s*(?:fail|error))",
        r"(?:linker\s*error|import\s*error|module\s*not\s*found)",
        r"(?:cannot\s*find\s*module|unresolved\s*(?:import|reference))",
    ],
    "deploy": [
        r"(?:deploy(?:ment)?\s*(?:fail|error)|rollback|service\s*(?:down|unavailable))",
        r"(?:health\s*check\s*fail|container\s*(?:crash|restart))",
    ],
    "config": [
        r"(?:config(?:uration)?\s*(?:error|missing|invalid))",
        r"(?:env(?:ironment)?\s*(?:variable|missing)|\.env|settings?\s*(?:wrong|missing))",
    ],
}

# Root cause extraction patterns
ROOT_CAUSE_EXTRACTORS = [
    r"(?:root\s*cause|caused\s*by|because|reason(?:\s*was)?|due\s*to|problem\s*was|issue\s*was)[:\s]+(.{10,200})",
    r"(?:nguyên\s*nhân|do|vì|bởi\s*vì)[:\s]+(.{10,200})",
    r"(?:the\s*(?:real|actual|underlying)\s*(?:issue|problem|cause)\s*(?:is|was))[:\s]+(.{10,200})",
    r"(?:fixed\s*by|resolved\s*by|solution\s*was)[:\s]+(.{10,200})",
]

# Severity keywords for auto-detection
SEVERITY_KEYWORDS = {
    "critical": [r"critical", r"fatal", r"data\s*loss", r"security\s*(?:vuln|breach)", r"production\s*(?:down|crash)"],
    "high": [r"crash", r"hang", r"block(?:ed|ing)", r"regression", r"broken\s*(?:build|deploy|test)"],
    "low": [r"cosmetic", r"typo", r"formatting", r"style", r"minor", r"warning\b"],
}


def classify_error_type(text: str) -> str:
    """Auto-classify error type from content. Returns empty string if no match."""
    text_lower = text.lower()
    best_type = ""
    best_score = 0
    for etype, patterns in ERROR_TYPE_PATTERNS.items():
        score = 0
        for p in patterns:
            score += len(re.findall(p, text_lower, re.IGNORECASE))
        if score > best_score:
            best_score = score
            best_type = etype
    return best_type if best_score >= 1 else ""


def extract_root_cause(text: str) -> str:
    """Extract root cause description from content. Returns empty string if none found."""
    for p in ROOT_CAUSE_EXTRACTORS:
        m = re.search(p, text, re.IGNORECASE)
        if m:
            cause = m.group(1).strip().rstrip(".")
            if len(cause) > 10:
                return cause[:200]
    return ""


def detect_severity(text: str) -> str:
    """Auto-detect severity from content keywords. Returns 'medium' as default."""
    text_lower = text.lower()
    for sev in ("critical", "high", "low"):
        for p in SEVERITY_KEYWORDS[sev]:
            if re.search(p, text_lower, re.IGNORECASE):
                return sev
    return "medium"


# Stopwords for concept tag extraction (pure stdlib, no ML imports)
_CONCEPT_STOPWORDS = frozenset(
    {
        "the",
        "a",
        "an",
        "and",
        "or",
        "but",
        "in",
        "on",
        "at",
        "to",
        "for",
        "of",
        "with",
        "by",
        "from",
        "is",
        "are",
        "was",
        "were",
        "be",
        "been",
        "have",
        "has",
        "had",
        "do",
        "does",
        "did",
        "will",
        "would",
        "could",
        "should",
        "may",
        "might",
        "can",
        "it",
        "this",
        "that",
        "these",
        "those",
        "i",
        "we",
        "you",
        "he",
        "she",
        "they",
        "not",
        "no",
        "so",
        "if",
        "then",
        "when",
        "where",
        "what",
        "which",
        "who",
        "how",
        "all",
        "any",
        "each",
        "more",
        "most",
        "also",
        "just",
        "up",
        "out",
        "as",
        "into",
        "than",
        "their",
        "its",
        "our",
        "my",
        "your",
        "his",
        "her",
        "them",
        "us",
        "me",
        "after",
        "before",
        "during",
        "while",
        "since",
        "until",
        "too",
        "very",
        "about",
        "above",
        "below",
        "between",
        "through",
        "use",
        "used",
        "using",
        "run",
        "running",
        "make",
        "new",
        "only",
        "now",
        "time",
        "way",
        "need",
        "needs",
        "see",
        "get",
        "set",
        "add",
        "put",
        "let",
        "say",
        "one",
        "two",
        "per",
        "via",
        "etc",
        "yet",
        "got",
    }
)


def extract_concept_tags(text: str, top_k: int = 5) -> list:
    """Extract top_k concept tags from text using pure-stdlib term frequency.

    Distinct from extract_tags() (which parses explicit user-supplied tags from
    checkpoint text). This function performs automatic keyword extraction from
    free-form entry title + content using stopword-filtered term frequency.

    Args:
        text: Combined title and content text to analyze.
        top_k: Maximum number of concept tags to return.

    Returns:
        List of up to top_k lowercase concept tag strings, sorted by frequency desc.
    """
    if not text:
        return []
    # Tokenize: sequences of letters/digits/hyphens/underscores, min 3 chars
    tokens = re.findall(r"[a-zA-Z][a-zA-Z0-9_-]{2,}", text.lower())
    freq: dict = {}
    for tok in tokens:
        if tok not in _CONCEPT_STOPWORDS:
            freq[tok] = freq.get(tok, 0) + 1
    # Sort by frequency descending, then alphabetically for stability
    ranked = sorted(freq.items(), key=lambda x: (-x[1], x[0]))
    return [tag for tag, _ in ranked[:top_k]]


def ensure_tables(db: sqlite3.Connection):
    """Create knowledge_entries table if not exists."""
    db.executescript("""
        CREATE TABLE IF NOT EXISTS knowledge_entries (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            session_id TEXT NOT NULL,
            document_id INTEGER,
            category TEXT NOT NULL,
            title TEXT NOT NULL,
            stable_id TEXT,
            content TEXT NOT NULL,
            tags TEXT DEFAULT '',
            confidence REAL DEFAULT 1.0,
            occurrence_count INTEGER DEFAULT 1,
            first_seen TEXT,
            last_seen TEXT,
            source TEXT DEFAULT 'copilot',
            topic_key TEXT,
            revision_count INTEGER DEFAULT 1,
            content_hash TEXT,
            wing TEXT DEFAULT '',
            room TEXT DEFAULT '',
            facts TEXT DEFAULT '[]',
            est_tokens INTEGER DEFAULT 0,
            task_id TEXT DEFAULT '',
            affected_files TEXT DEFAULT '[]',
            source_section TEXT DEFAULT '',
            source_file TEXT DEFAULT '',
            start_line INTEGER DEFAULT 0,
            end_line INTEGER DEFAULT 0,
            code_language TEXT DEFAULT '',
            code_snippet TEXT DEFAULT '',
            UNIQUE(category, title, session_id)
        );

        CREATE INDEX IF NOT EXISTS idx_ke_category ON knowledge_entries(category);
        CREATE INDEX IF NOT EXISTS idx_ke_session ON knowledge_entries(session_id);
        CREATE INDEX IF NOT EXISTS idx_ke_source ON knowledge_entries(source);
        CREATE INDEX IF NOT EXISTS idx_ke_topic ON knowledge_entries(topic_key);
        CREATE INDEX IF NOT EXISTS idx_ke_hash ON knowledge_entries(content_hash);
        CREATE INDEX IF NOT EXISTS idx_ke_task ON knowledge_entries(task_id);
        CREATE INDEX IF NOT EXISTS idx_ke_stable_id ON knowledge_entries(stable_id);

        CREATE TABLE IF NOT EXISTS knowledge_relations (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            source_id INTEGER REFERENCES knowledge_entries(id),
            target_id INTEGER REFERENCES knowledge_entries(id),
            source_stable_id TEXT DEFAULT '',
            target_stable_id TEXT DEFAULT '',
            relation_type TEXT NOT NULL,
            stable_id TEXT,
            confidence REAL DEFAULT 0.8,
            created_at TEXT,
            UNIQUE(source_id, target_id, relation_type)
        );

        CREATE INDEX IF NOT EXISTS idx_kr_source ON knowledge_relations(source_id);
        CREATE INDEX IF NOT EXISTS idx_kr_target ON knowledge_relations(target_id);
        CREATE INDEX IF NOT EXISTS idx_kr_source_stable ON knowledge_relations(source_stable_id);
        CREATE INDEX IF NOT EXISTS idx_kr_target_stable ON knowledge_relations(target_stable_id);
        CREATE INDEX IF NOT EXISTS idx_kr_stable_id ON knowledge_relations(stable_id);

        CREATE TABLE IF NOT EXISTS entity_relations (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            subject TEXT NOT NULL,
            predicate TEXT NOT NULL,
            object TEXT NOT NULL,
            stable_id TEXT,
            noted_at TEXT DEFAULT (datetime('now')),
            session_id TEXT DEFAULT '',
            UNIQUE(subject, predicate, object)
        );

        CREATE INDEX IF NOT EXISTS idx_er_subject ON entity_relations(subject);
        CREATE INDEX IF NOT EXISTS idx_er_object ON entity_relations(object);
        CREATE INDEX IF NOT EXISTS idx_er_stable_id ON entity_relations(stable_id);

        CREATE TABLE IF NOT EXISTS wakeup_config (
            key TEXT PRIMARY KEY,
            value TEXT NOT NULL,
            updated_at TEXT DEFAULT (datetime('now'))
        );

        CREATE TABLE IF NOT EXISTS schema_version (
            version INTEGER PRIMARY KEY,
            name TEXT NOT NULL,
            applied_at TEXT DEFAULT (datetime('now'))
        );
    """)

    # Migrate existing databases: add new columns if missing
    _ALLOWED_COLUMNS = {
        "stable_id",
        "source",
        "topic_key",
        "revision_count",
        "content_hash",
        "wing",
        "room",
        "facts",
        "est_tokens",
        "task_id",
        "affected_files",
        "source_section",
        "source_file",
        "start_line",
        "end_line",
        "code_language",
        "code_snippet",
    }
    for col, col_def in [
        ("stable_id", "TEXT"),
        ("source", "TEXT DEFAULT 'copilot'"),
        ("topic_key", "TEXT"),
        ("revision_count", "INTEGER DEFAULT 1"),
        ("content_hash", "TEXT"),
        ("wing", "TEXT DEFAULT ''"),
        ("room", "TEXT DEFAULT ''"),
        ("facts", "TEXT DEFAULT '[]'"),
        ("est_tokens", "INTEGER DEFAULT 0"),
        ("task_id", "TEXT DEFAULT ''"),
        ("affected_files", "TEXT DEFAULT '[]'"),
        ("source_section", "TEXT DEFAULT ''"),
        ("source_file", "TEXT DEFAULT ''"),
        ("start_line", "INTEGER DEFAULT 0"),
        ("end_line", "INTEGER DEFAULT 0"),
        ("code_language", "TEXT DEFAULT ''"),
        ("code_snippet", "TEXT DEFAULT ''"),
    ]:
        assert col in _ALLOWED_COLUMNS, f"Unexpected column: {col}"
        try:
            db.execute(f"ALTER TABLE knowledge_entries ADD COLUMN {col} {col_def}")
        except sqlite3.OperationalError:
            pass  # Column already exists

    db.execute("CREATE INDEX IF NOT EXISTS idx_ke_task ON knowledge_entries(task_id)")
    db.execute("CREATE INDEX IF NOT EXISTS idx_ke_stable_id ON knowledge_entries(stable_id)")

    for col, col_def in [
        ("source_stable_id", "TEXT DEFAULT ''"),
        ("target_stable_id", "TEXT DEFAULT ''"),
        ("stable_id", "TEXT"),
    ]:
        try:
            db.execute(f"ALTER TABLE knowledge_relations ADD COLUMN {col} {col_def}")
        except sqlite3.OperationalError:
            pass
    db.execute("CREATE INDEX IF NOT EXISTS idx_kr_source_stable ON knowledge_relations(source_stable_id)")
    db.execute("CREATE INDEX IF NOT EXISTS idx_kr_target_stable ON knowledge_relations(target_stable_id)")
    db.execute("CREATE INDEX IF NOT EXISTS idx_kr_stable_id ON knowledge_relations(stable_id)")

    try:
        db.execute("ALTER TABLE entity_relations ADD COLUMN stable_id TEXT")
    except sqlite3.OperationalError:
        pass
    db.execute("CREATE INDEX IF NOT EXISTS idx_er_stable_id ON entity_relations(stable_id)")

    # Create FTS table if needed (standalone, no content= sync issues)
    db.execute("""
        CREATE VIRTUAL TABLE IF NOT EXISTS ke_fts USING fts5(
            title, content, tags, category, wing, room, facts,
            tokenize='unicode61 remove_diacritics 2'
        )
    """)
    _seed_sync_table_policies(db)
    _backfill_stable_ids(db)
    _enforce_stable_id_uniqueness(db)


def classify_paragraph(text: str) -> list[tuple[str, float]]:
    """Classify a paragraph into knowledge categories with confidence.

    Category-aware thresholds and confidence floors:
      - pattern: threshold=1 (easier to extract), floor=0.5
      - decision: threshold=2, floor=0.5
      - mistake/tool/feature/refactor/discovery: threshold=2, floor=0.4
    """
    # Skip noise: interview Q&A, pure tables, pure code
    if _is_noise(text):
        return []

    text_lower = text.lower()
    results = []

    # (threshold, confidence_floor) per category
    _CATEGORY_CONFIG = {
        "pattern": (1, 0.5),
        "decision": (2, 0.5),
        "mistake": (2, 0.4),
        "tool": (2, 0.4),
        "feature": (2, 0.4),
        "refactor": (2, 0.4),
        "discovery": (2, 0.4),
    }

    for category, indicators in [
        ("mistake", MISTAKE_INDICATORS),
        ("pattern", PATTERN_INDICATORS),
        ("decision", DECISION_INDICATORS),
        ("tool", TOOL_INDICATORS),
        ("feature", FEATURE_INDICATORS),
        ("refactor", REFACTOR_INDICATORS),
        ("discovery", DISCOVERY_INDICATORS),
    ]:
        threshold, conf_floor = _CATEGORY_CONFIG.get(category, (2, 0.4))
        score = 0
        for pattern in indicators:
            matches = len(re.findall(pattern, text_lower, re.IGNORECASE))
            score += matches
        if score >= threshold:
            confidence = max(conf_floor, min(1.0, score / 5.0))
            results.append((category, confidence))

    return results


# Noise detection patterns
_NOISE_PATTERNS = [
    r"(?:phỏng\s*vấn|interview|câu\s*hỏi|bộ\s*câu)",
    r"(?:đáp\s*án|mong\s*đợi|tiêu\s*chí|ghi\s*điểm)",
    r"(?:bảng\s*đánh\s*giá|evaluation\s*rubric)",
    r"(?:trọng\s*số|scoring|rubric|interviewer)",
]

# Strong noise — single match is enough to discard
_STRONG_NOISE_PATTERNS = [
    r"đáp\s*án\s*(mong\s*đợi|chi\s*tiết)",
    r"bảng\s*(đánh\s*giá|ghi\s*điểm)",
    r"câu\s*hỏi\s*phỏng\s*vấn",
    r"interview\s*question",
    r"nội\s*dung\s*cần\s*đề\s*cập",
]

# User-quote patterns — checkpoint summaries quoting the user, not real mistakes
# KEEP: "User pointed out", "User noticed", "User called out", "User criticized",
#        "User demanded" — these are real bug reports / legitimate feedback
_USER_QUOTE_PATTERNS = [
    r"^(?:\d+\.\s*)?user\s+(?:said|asked|reported|requested|mentioned|noted)\b",
    r"^(?:\d+\.\s*)?user\s+(?:wants?|confirmed|approved|rejected)\b",
    r"^(?:\d+\.\s*)?user\s+(?:clarified|provided|applied|selected|chose)\b",
    r'^(?:\d+\.\s*)?user\s+said\s*[:"]',
    r'^(?:\d+\.\s*)?user\s+reported\s*[:"]',
]

# Action-summary patterns — past-tense descriptions of completed work
_ACTION_SUMMARY_PATTERNS = [
    r"^(?:\d+\.\s*)?(?:fixed|implemented|launched|created|updated|added|deployed|"
    r"committed|pushed|merged|resolved|completed|refactored|migrated|upgraded|"
    r"configured|installed|removed|deleted|replaced|renamed|moved)\s",
]


def _is_noise(text: str) -> bool:
    """Check if text is interview Q&A, scoring rubric, or other noise."""
    text_lower = text.lower()

    # Strong noise — single match enough
    for pattern in _STRONG_NOISE_PATTERNS:
        if re.search(pattern, text_lower):
            return True

    # User quotes — checkpoint summaries that quote what user said
    first_line = text_lower.strip().split("\n")[0].strip()
    for pattern in _USER_QUOTE_PATTERNS:
        if re.search(pattern, first_line):
            return True

    # Action summaries — past-tense descriptions of completed work
    # Only filter short entries (< 200 chars) that are just action log lines
    if len(text.strip()) < 200:
        for pattern in _ACTION_SUMMARY_PATTERNS:
            if re.search(pattern, first_line):
                return True

    # Weak noise — need 2+ matches
    noise_score = 0
    for pattern in _NOISE_PATTERNS:
        if re.search(pattern, text_lower):
            noise_score += 1
    if noise_score >= 2:
        return True

    # Pure markdown table (>70% of lines are table rows)
    lines = [l.strip() for l in text.split("\n") if l.strip()]
    if lines:
        table_lines = sum(1 for l in lines if l.startswith("|") and l.endswith("|"))
        if table_lines / len(lines) > 0.7 and len(lines) > 3:
            return True

    # Pure code block (>70% of content inside ```)
    # Exception: preserve blocks that contain stack traces or error info
    code_chars = sum(len(m.group(0)) for m in re.finditer(r"```[\s\S]*?```", text))
    if len(text) > 100 and code_chars / len(text) > 0.7:
        # Check if the code block contains error/traceback info worth keeping
        has_error_info = bool(re.search(r"(?:Traceback|Error|Exception|FAILED|panic|stack trace)", text, re.IGNORECASE))
        if not has_error_info:
            return True

    return False


def extract_title(text: str, max_len: int = 100) -> str:
    """Extract a meaningful title from paragraph text."""
    lines = text.strip().split("\n")

    for line in lines[:5]:  # Try first 5 lines
        line = line.strip()
        if not line:
            continue
        # Skip table rows, code markers, empty headings
        if line.startswith("|") or line.startswith("```") or line.startswith("---"):
            continue
        # Skip lines that are just separators
        if re.match(r"^[-=_]{3,}$", line):
            continue

        # Remove markdown formatting
        title = re.sub(r"[#*_`\[\]]", "", line).strip()
        # Remove leading bullets/numbers/emoji
        title = re.sub(r"^[\d.)\-•]+\s*", "", title).strip()
        title = re.sub(r"^[^\w\s]{1,3}\s*", "", title).strip()  # emoji prefix

        if len(title) >= 10:
            if len(title) > max_len:
                title = title[: max_len - 3] + "..."
            return title

    # Fallback: first 100 chars of text
    fallback = re.sub(r"\s+", " ", text[:max_len]).strip()
    return fallback or "Untitled"


def extract_tags(text: str) -> str:
    """Extract relevant tags from text."""
    tag_patterns = [
        (r"\b(?:Spring\s+Boot|SpringBoot)\b", "spring-boot"),
        (r"\b(?:Thymeleaf)\b", "thymeleaf"),
        (r"\b(?:JPQL|JPA|Hibernate)\b", "jpa"),
        (r"\b(?:PostgreSQL|Postgres|PG)\b", "postgresql"),
        (r"\b(?:Docker|docker-compose)\b", "docker"),
        (r"\b(?:Redis)\b", "redis"),
        (r"\b(?:Gradle)\b", "gradle"),
        (r"\b(?:CSRF)\b", "csrf"),
        (r"\b(?:Liquibase)\b", "liquibase"),
        (r"\b(?:JavaScript|jQuery|JS)\b", "javascript"),
        (r"\b(?:CSS|styles?\.css)\b", "css"),
        (r"\b(?:i18n|internationalization|messages\.properties)\b", "i18n"),
        (r"\b(?:JDK|Java\s+\d+)\b", "java"),
        (r"\b(?:Git|git\s+hook)\b", "git"),
        (r"\b(?:VSCode|VS\s+Code)\b", "vscode"),
        (r"\b(?:Excel|xlsx)\b", "excel"),
        (r"\b(?:CRUD)\b", "crud"),
        (r"\b(?:pagination)\b", "pagination"),
        (r"\b(?:modal|dialog)\b", "ui"),
        (r"\b(?:SQL|native\s+SQL)\b", "sql"),
        # Extended coverage
        (r"\b(?:Python|python3?)\b", "python"),
        (r"\b(?:TypeScript)\b", "typescript"),
        (r"\b(?:React(?:JS)?|ReactDOM|react-native)\b", "react"),
        (r"\b(?:Node(?:\.js)?|nodejs)\b", "nodejs"),
        (r"\b(?:Kotlin)\b", "kotlin"),
        (r"\b(?:Swift)\b", "swift"),
        (r"\b(?:AWS|Lambda|ECS|CloudFront|S3|DynamoDB|SQS|SNS)\b", "aws"),
        (r"\b(?:pytest|unittest|vitest|jest|mocha)\b", "testing"),
        (r"\b(?:async|await|asyncio|coroutine)\b", "async"),
        (r"\b(?:SQLite|sqlite3)\b", "sqlite"),
        (r"\b(?:migration|schema|database)\b", "database"),
        (r"\b(?:authentication|authorization|auth|JWT|OAuth|token)\b", "auth"),
        (r"\b(?:REST|GraphQL|API|endpoint|HTTP)\b", "api"),
        (r"\b(?:CI|CD|GitHub\s+Actions|pipeline|workflow)\b", "ci-cd"),
        (r"\b(?:Compose|Jetpack\s+Compose|Composable)\b", "compose"),
        (r"\b(?:iOS|UIKit|SwiftUI|Xcode)\b", "ios"),
        (r"\b(?:Android|Gradle|AndroidX)\b", "android"),
    ]

    tags = set()
    for pattern, tag in tag_patterns:
        if re.search(pattern, text, re.IGNORECASE):
            tags.add(tag)
    return ",".join(sorted(tags))


def split_into_knowledge_chunks(content: str) -> list[str]:
    """Split section content into meaningful chunks for classification."""
    chunks = []

    # Split by numbered items, bullet points, or double newlines
    # Prefer structured items (numbered lists, bullets)
    items = re.split(r"\n(?=\d+\.\s|\-\s|\*\s|#{1,3}\s)", content)

    for item in items:
        item = item.strip()
        if len(item) < 30:  # Skip very short fragments
            continue
        if len(item) > 2000:  # Split long chunks by paragraphs
            paragraphs = item.split("\n\n")
            for p in paragraphs:
                if len(p.strip()) >= 30:
                    chunks.append(p.strip())
        else:
            chunks.append(item)

    return chunks


def _slugify(text: str) -> str:
    """Convert text to URL-friendly slug for topic keys."""
    text = text.lower().strip()
    text = re.sub(r"[^\w\s-]", "", text)
    text = re.sub(r"[\s_]+", "-", text)
    text = re.sub(r"-+", "-", text).strip("-")
    return text[:60]


def _compute_content_hash(category: str, title: str, content: str) -> str:
    """Compute dedup hash from normalized category + title + key content."""
    normalized = (
        category.lower().strip()
        + "|"
        + re.sub(r"\s+", " ", title.lower().strip())
        + "|"
        + re.sub(r"\s+", " ", content[:200].lower().strip())
    )
    return hashlib.sha256(normalized.encode("utf-8")).hexdigest()[:16]


def _generate_topic_key(category: str, title: str) -> str:
    """Generate a topic key like 'decision/auth-jwt-approach'."""
    return f"{category}/{_slugify(title)}"


def _parse_file_list(content: str) -> list:
    """Parse file paths from an important_files section content."""
    files = []
    for line in content.splitlines():
        line = line.strip()
        # Strip leading bullet markers
        if line.startswith(("- ", "* ", "+ ")):
            path = line[2:].strip()
        elif line.startswith(("  - ", "  * ")):
            path = line[4:].strip()
        else:
            path = line
        # Strip backtick wrapping
        path = path.strip("`")
        # Strip inline comments and table separators
        path = path.split("#")[0].strip()
        path = path.split("|")[0].strip()
        # Skip empty, too long, or URL-like lines
        if not path or len(path) > 256:
            continue
        if path.startswith(("http://", "https://", "[")):
            continue
        # Must resemble a file path: contains a dot (extension) or a slash
        if "." in path or "/" in path:
            files.append(path)
    return files


# Matches explicit task/tentacle slug markers in text, e.g. "task: fix-auth" or "tentacle=wave3-signal"
_TASK_ID_MARKER_RE = re.compile(
    r"(?:task|tentacle)\s*[=:]\s*([a-z][a-z0-9]*(?:-[a-z0-9]+)+)\b",
    re.IGNORECASE,
)
# Valid kebab-case slug: lowercase, at least one hyphen, min 5 chars total
_KEBAB_SLUG_RE = re.compile(r"^[a-z][a-z0-9]*(?:-[a-z0-9]+)+$")


def _backfill_affected_files_from_session_evidence(db: sqlite3.Connection, session_ids: list | None = None) -> int:
    """Backfill affected_files from the same session's important_files section.

    Only updates entries where affected_files is empty.
    Does NOT overwrite manually populated affected_files.
    Safe evidence only: same-session important_files checkpoint section.
    Returns count of updated rows.
    """
    query = """
        SELECT id, session_id FROM knowledge_entries
        WHERE (affected_files IS NULL OR affected_files = '[]' OR affected_files = '')
        AND session_id IS NOT NULL AND session_id != ''
    """
    params = []
    if session_ids:
        placeholders = ",".join("?" for _ in session_ids)
        query += f" AND session_id IN ({placeholders})"
        params.extend(session_ids)
    empty_entries = db.execute(query, params).fetchall()

    if not empty_entries:
        return 0

    # Group entry IDs by session
    by_session: dict = {}
    for row in empty_entries:
        by_session.setdefault(row[1], []).append(row[0])

    updated = 0
    for session_id, entry_ids in by_session.items():
        # Fetch most recent important_files section for this session
        try:
            row = db.execute(
                """
                SELECT s.content FROM sections s
                JOIN documents d ON s.document_id = d.id
                WHERE d.session_id = ? AND s.section_name = 'important_files'
                ORDER BY d.seq DESC LIMIT 1
            """,
                (session_id,),
            ).fetchone()
        except sqlite3.OperationalError:
            continue

        if not row or not row[0]:
            continue

        files = _parse_file_list(row[0])
        if not files:
            continue

        files_json = json.dumps(files[:20])  # cap per-entry at 20 files
        for entry_id in entry_ids:
            db.execute(
                "UPDATE knowledge_entries SET affected_files = ? "
                "WHERE id = ? AND (affected_files IS NULL OR affected_files = '[]' OR affected_files = '')",
                (files_json, entry_id),
            )
            updated += 1

    return updated


def _infer_task_ids_from_content(db: sqlite3.Connection, session_ids: list | None = None) -> int:
    """Infer task_id from explicit task/tentacle markers in content.

    Only assigns when exactly ONE kebab-case slug candidate is found.
    Does NOT overwrite existing task_id values.
    Returns count of updated rows.
    """
    query = """
        SELECT id, title, content FROM knowledge_entries
        WHERE (task_id IS NULL OR task_id = '')
        AND (title IS NOT NULL OR content IS NOT NULL)
    """
    params = []
    if session_ids:
        placeholders = ",".join("?" for _ in session_ids)
        query += f" AND session_id IN ({placeholders})"
        params.extend(session_ids)
    rows = db.execute(query, params).fetchall()

    updated = 0
    for entry_id, title, content in rows:
        text = f"{title or ''} {content or ''}"
        candidates = set()
        for m in _TASK_ID_MARKER_RE.finditer(text):
            slug = m.group(1).lower().strip()
            if _KEBAB_SLUG_RE.match(slug) and 5 <= len(slug) <= 50:
                candidates.add(slug)

        if len(candidates) == 1:
            slug = next(iter(candidates))
            db.execute(
                "UPDATE knowledge_entries SET task_id = ? WHERE id = ? AND (task_id IS NULL OR task_id = '')",
                (slug, entry_id),
            )
            updated += 1

    return updated


def extract_from_sections(db: sqlite3.Connection, session_ids: list = None):
    """Extract knowledge entries from indexed sections.

    Args:
        session_ids: If provided, only extract from these sessions (selective mode).
    """
    now = datetime.now().isoformat()
    extracted = 0
    skipped = 0
    deduped = 0

    # Focus on the most knowledge-rich sections
    target_sections = ["technical_details", "history", "work_done", "next_steps", "full", "conversation"]

    section_placeholders = ",".join("?" for _ in target_sections)
    base_query = f"""
        SELECT s.id, s.document_id, s.section_name, s.content, d.session_id,
               COALESCE(d.source, 'copilot') as source,
               COALESCE(d.stable_id, '') as document_stable_id
        FROM sections s
        JOIN documents d ON s.document_id = d.id
        WHERE s.section_name IN ({section_placeholders})
    """
    query_params = list(target_sections)

    if session_ids:
        session_placeholders = ",".join("?" for _ in session_ids)
        base_query += f" AND d.session_id IN ({session_placeholders})"
        query_params.extend(session_ids)

    base_query += " ORDER BY d.session_id, d.seq"
    try:
        rows = db.execute(base_query, query_params).fetchall()
    except sqlite3.OperationalError:
        rows = []

    # Pre-load existing content hashes for fast dedup
    existing_hashes = set()
    try:
        for row in db.execute("SELECT content_hash FROM knowledge_entries WHERE content_hash IS NOT NULL"):
            existing_hashes.add(row[0])
    except sqlite3.OperationalError:
        pass

    for section_id, doc_id, section_name, content, session_id, source, document_stable_id in rows:
        chunks = split_into_knowledge_chunks(content)

        for chunk in chunks:
            classifications = classify_paragraph(chunk)

            for category, confidence in classifications:
                title = extract_title(chunk)
                tags = extract_tags(chunk)
                content_hash = _compute_content_hash(category, title, chunk)
                topic_key = _generate_topic_key(category, title)

                # WBS-013: security scan — skip entries with credentials/injection
                unsafe_reason = _scan_extract_chunk(title, chunk)
                if unsafe_reason:
                    print(
                        f"  [extract] SKIPPED unsafe chunk ({unsafe_reason}): {title[:60]!r}",
                        file=sys.stderr,
                    )
                    skipped += 1
                    continue

                # Hash-based dedup: skip if exact content already exists
                if content_hash in existing_hashes:
                    deduped += 1
                    continue

                # Topic key upsert: if same topic exists, update instead of insert
                existing = db.execute(
                    "SELECT id, revision_count FROM knowledge_entries WHERE topic_key = ? AND session_id != ?",
                    (topic_key, session_id),
                ).fetchone()

                if existing:
                    stable_id = _knowledge_stable_id(session_id, category, title, topic_key)
                    # Upsert: update existing entry with newer content + recurrence reward
                    db.execute(
                        """
                        UPDATE knowledge_entries
                        SET content = ?, confidence = MIN(1.0, MAX(confidence, ?) + 0.03),
                            revision_count = revision_count + 1,
                            occurrence_count = occurrence_count + 1,
                            last_seen = ?, content_hash = ?, tags = ?,
                            source_section = ?, stable_id = CASE
                                WHEN COALESCE(stable_id, '') = '' THEN ?
                                ELSE stable_id
                            END
                        WHERE id = ?
                    """,
                        (chunk[:3000], confidence, now, content_hash, tags, section_name, stable_id, existing[0]),
                    )
                    _enqueue_sync_op_fail_open(
                        db,
                        "knowledge_entries",
                        stable_id,
                        {
                            "session_id": session_id,
                            "document_stable_id": document_stable_id,
                            "category": category,
                            "title": title,
                            "stable_id": stable_id,
                            "content": chunk[:3000],
                            "tags": tags,
                            "confidence": confidence,
                            "last_seen": now,
                            "content_hash": content_hash,
                            "source": source,
                            "topic_key": topic_key,
                            "source_section": section_name,
                        },
                    )
                    existing_hashes.add(content_hash)
                    extracted += 1
                    continue

                try:
                    est_tokens = len(f"{title} {chunk[:3000]}") // 4
                    stable_id = _knowledge_stable_id(session_id, category, title, topic_key)
                    db.execute(
                        """
                        INSERT INTO knowledge_entries
                        (session_id, document_id, category, title, stable_id, content, tags,
                         confidence, first_seen, last_seen, source, topic_key,
                         revision_count, content_hash, est_tokens, source_section)
                        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?, ?)
                        ON CONFLICT(category, title, session_id) DO UPDATE SET
                            confidence = MAX(knowledge_entries.confidence, excluded.confidence),
                            last_seen = excluded.last_seen,
                            content_hash = excluded.content_hash,
                            topic_key = excluded.topic_key,
                            est_tokens = excluded.est_tokens,
                            source_section = excluded.source_section,
                            stable_id = excluded.stable_id
                    """,
                        (
                            session_id,
                            doc_id,
                            category,
                            title,
                            stable_id,
                            chunk[:3000],
                            tags,
                            confidence,
                            now,
                            now,
                            source,
                            topic_key,
                            content_hash,
                            est_tokens,
                            section_name,
                        ),
                    )
                    _enqueue_sync_op_fail_open(
                        db,
                        "knowledge_entries",
                        stable_id,
                        {
                            "session_id": session_id,
                            "document_stable_id": document_stable_id,
                            "category": category,
                            "title": title,
                            "stable_id": stable_id,
                            "content": chunk[:3000],
                            "tags": tags,
                            "confidence": confidence,
                            "first_seen": now,
                            "last_seen": now,
                            "source": source,
                            "topic_key": topic_key,
                            "revision_count": 1,
                            "content_hash": content_hash,
                            "est_tokens": est_tokens,
                            "source_section": section_name,
                        },
                    )
                    existing_hashes.add(content_hash)
                    extracted += 1

                    # Auto-classify error lifecycle fields for mistake entries
                    if category == "mistake":
                        entry_id_row = db.execute("SELECT last_insert_rowid()").fetchone()
                        if entry_id_row:
                            eid = entry_id_row[0]
                            et = classify_error_type(chunk)
                            rc = extract_root_cause(chunk)
                            sv = detect_severity(chunk)
                            if et or rc or sv != "medium":
                                try:
                                    db.execute(
                                        "UPDATE knowledge_entries SET error_type = ?, root_cause = ?, severity = ? WHERE id = ? AND COALESCE(error_type, '') = ''",
                                        (et, rc, sv, eid),
                                    )
                                except sqlite3.OperationalError:
                                    pass  # error lifecycle columns may not exist yet

                except sqlite3.IntegrityError as e:
                    print(f"⚠ Duplicate entry skipped: {e}", file=sys.stderr)
                    skipped += 1

    # Signal backfill: populate affected_files and task_id from safe session-local evidence
    _backfill_affected_files_from_session_evidence(db, session_ids=session_ids)
    _infer_task_ids_from_content(db, session_ids=session_ids)

    # Confidence decay: entries not seen recently get slightly lower confidence
    # Apply at most once per day to prevent compounding from repeated runs
    try:
        today = now[:10]
        last_decay = None
        try:
            row = db.execute("SELECT value FROM embedding_meta WHERE key = 'last_decay_date'").fetchone()
            if row:
                last_decay = row[0]
        except sqlite3.OperationalError:
            pass  # embedding_meta table may not exist

        if last_decay != today:
            db.execute(
                """
                UPDATE knowledge_entries
                SET confidence = MAX(0.3,
                    CASE WHEN confidence >= 0.8 THEN confidence * 0.98
                         ELSE confidence * 0.95
                    END)
                WHERE last_seen < ? AND confidence > 0.3
            """,
                (today,),
            )
            try:
                db.execute("""
                    CREATE TABLE IF NOT EXISTS embedding_meta (
                        key TEXT PRIMARY KEY, value TEXT
                    )
                """)
                db.execute(
                    """
                    INSERT OR REPLACE INTO embedding_meta (key, value)
                    VALUES ('last_decay_date', ?)
                """,
                    (today,),
                )
            except sqlite3.OperationalError:
                pass
    except sqlite3.OperationalError:
        pass

    # Sync ke_fts — incremental when session_ids is provided, global otherwise.
    # Incremental path: delete then re-insert only rows belonging to the affected
    # sessions (rowid in ke_fts maps 1-to-1 to id in knowledge_entries).
    # Global path: full DELETE + INSERT — preserved for non-scoped runs.
    if session_ids:
        _ph = ",".join("?" for _ in session_ids)
        db.execute(
            f"DELETE FROM ke_fts WHERE rowid IN (SELECT id FROM knowledge_entries WHERE session_id IN ({_ph}))",
            session_ids,
        )
        db.execute(
            f"""
            INSERT INTO ke_fts (rowid, title, content, tags, category, wing, room, facts)
            SELECT id, title, content, tags, category,
                   COALESCE(wing,''), COALESCE(room,''), COALESCE(facts,'[]')
            FROM knowledge_entries
            WHERE session_id IN ({_ph})
            """,
            session_ids,
        )
    else:
        db.execute("DELETE FROM ke_fts")
        db.execute("""
            INSERT INTO ke_fts (rowid, title, content, tags, category, wing, room, facts)
            SELECT id, title, content, tags, category,
                   COALESCE(wing,''), COALESCE(room,''), COALESCE(facts,'[]')
            FROM knowledge_entries
        """)

    # Extract relations between knowledge entries
    relations_count = extract_relations(db)

    db.commit()
    return extracted, skipped, deduped, relations_count


def extract_relations(db: sqlite3.Connection) -> int:
    """Detect and insert relationships between knowledge entries.

    Relation types:
      SAME_SESSION       — entries from same session but different categories (0.7)
      SAME_TOPIC         — entries with same topic_key from different sessions (0.9)
      TAG_OVERLAP        — entries sharing 2+ tags (0.5 + 0.1 * shared_count)
      RESOLVED_BY        — mistake paired with pattern/tool in same session (0.8)
      SEMANTIC_PROXIMITY — TF-IDF cosine similarity >= 0.75 when scikit-learn is
                           available; pairs already covered by stronger relation
                           types are skipped and missing sklearn is a no-op

    Ordering: entries are processed newest-first (ORDER BY id DESC) so that
    recent context is never starved when per-type budgets fill.  A per-group
    pair cap spreads each budget evenly across sessions/topics.
    """
    now = datetime.now().isoformat()
    MAX_PER_TYPE = 1500  # budget per relation type
    MAX_RELATIONS = 5000

    try:
        db.execute("DELETE FROM knowledge_relations")
    except sqlite3.OperationalError:
        pass

    # Ensure unique constraint index exists
    db.execute("""
        CREATE UNIQUE INDEX IF NOT EXISTS idx_relations_unique
        ON knowledge_relations(source_id, target_id, relation_type)
    """)

    # Load entries newest-first so recent context gets first access to relation
    # budgets.  Dict insertion order (Python 3.7+) then naturally makes
    # by_session / by_topic iterate newest sessions/topics first.
    entries = db.execute("""
        SELECT id, session_id, category, title, tags, topic_key, COALESCE(stable_id, '')
        FROM knowledge_entries
        ORDER BY id DESC
    """).fetchall()

    if not entries:
        return 0

    relations: list[
        tuple
    ] = []  # (source_id, target_id, src_stable, tgt_stable, relation_type, stable_id, confidence, created_at)

    # Build indexes for efficient lookups
    by_session: dict[str, list[tuple]] = {}
    by_topic: dict[str, list[tuple]] = {}
    by_id: dict[int, tuple] = {}
    for e in entries:
        eid, sid, cat, title, tags, topic, stable_id = e
        by_session.setdefault(sid, []).append(e)
        if topic:
            by_topic.setdefault(topic, []).append(e)
        by_id[eid] = e

    seen = set()  # (source_id, target_id, relation_type)
    type_counts = {}  # track count per relation type

    def _add(src: int, tgt: int, rtype: str, conf: float) -> bool:
        if src == tgt:
            return False
        key = (src, tgt, rtype)
        if key not in seen:
            cnt = type_counts.get(rtype, 0)
            if cnt >= MAX_PER_TYPE or len(relations) >= MAX_RELATIONS:
                return True  # signal budget exhausted
            seen.add(key)
            src = by_id.get(key[0])
            tgt = by_id.get(key[1])
            if not src or not tgt:
                return False
            src_sid = src[6] or _knowledge_stable_id(src[1], src[2], src[3], src[5] or "")
            tgt_sid = tgt[6] or _knowledge_stable_id(tgt[1], tgt[2], tgt[3], tgt[5] or "")
            stable_id = _knowledge_relation_stable_id(src_sid, tgt_sid, rtype)
            relations.append((key[0], key[1], src_sid, tgt_sid, rtype, stable_id, round(conf, 2), now))
            type_counts[rtype] = cnt + 1
        return False

    # 1. SAME_SESSION — different categories within same session
    # Per-session cap: each session contributes at most a fair share of the budget
    # so a single large session cannot crowd out all others.
    _active_sessions = [g for g in by_session.values() if len(g) >= 2]
    _ss_cap = max(3, MAX_PER_TYPE // max(1, len(_active_sessions)))
    for group in _active_sessions:
        session_added = 0
        budget_done = False
        for i, a in enumerate(group):
            if session_added >= _ss_cap:
                break
            for b in group[i + 1 :]:
                if session_added >= _ss_cap:
                    break
                if a[2] != b[2]:  # different category
                    pre_count = type_counts.get("SAME_SESSION", 0)
                    if _add(a[0], b[0], "SAME_SESSION", 0.7):
                        budget_done = True
                        break
                    if type_counts.get("SAME_SESSION", 0) > pre_count:
                        session_added += 1
            if budget_done:
                break
        if budget_done:
            break

    # 2. SAME_TOPIC — same topic_key, different sessions
    # Per-topic cap mirrors the SAME_SESSION coverage protection.
    _active_topics = [g for g in by_topic.values() if len(g) >= 2]
    _st_cap = max(3, MAX_PER_TYPE // max(1, len(_active_topics)))
    for group in _active_topics:
        topic_added = 0
        budget_done = False
        for i, a in enumerate(group):
            if topic_added >= _st_cap:
                break
            for b in group[i + 1 :]:
                if topic_added >= _st_cap:
                    break
                if a[1] != b[1]:  # different session_id
                    pre_count = type_counts.get("SAME_TOPIC", 0)
                    if _add(a[0], b[0], "SAME_TOPIC", 0.9):
                        budget_done = True
                        break
                    if type_counts.get("SAME_TOPIC", 0) > pre_count:
                        topic_added += 1
            if budget_done:
                break
        if budget_done:
            break

    # 3. TAG_OVERLAP — entries sharing 2+ tags
    entry_tags = []
    for e in entries:
        tags_str = e[4] or ""
        tag_set = frozenset(t.strip() for t in tags_str.split(",") if t.strip())
        if len(tag_set) >= 2:
            entry_tags.append((e[0], tag_set))

    for i, (aid, atags) in enumerate(entry_tags):
        done = False
        for bid, btags in entry_tags[i + 1 :]:
            shared = len(atags & btags)
            if shared >= 2:
                conf = min(1.0, 0.5 + 0.1 * min(shared, 5))
                if _add(aid, bid, "TAG_OVERLAP", conf):
                    done = True
                    break
        if done:
            break

    # 4. RESOLVED_BY — mistake + pattern/tool in same session
    # Iterate session groups in newest-first order and stop only when the
    # global relation budget is exhausted.
    for sid, group in by_session.items():
        mistakes = [e for e in group if e[2] == "mistake"]
        resolvers = [e for e in group if e[2] in ("pattern", "tool")]
        budget_done = False
        for m in mistakes:
            if budget_done:
                break
            for r in resolvers:
                if _add(m[0], r[0], "RESOLVED_BY", 0.8):
                    budget_done = True
                    break
        if budget_done:
            break

    # 5. SEMANTIC_PROXIMITY — TF-IDF cosine similarity >= 0.75 for pairs not
    # already covered by stronger relation types. Keep this offline-safe:
    # missing scikit-learn becomes a silent no-op.
    try:
        from sklearn.feature_extraction.text import TfidfVectorizer
        from sklearn.metrics.pairwise import cosine_similarity as sklearn_cosine_similarity
    except ImportError:
        TfidfVectorizer = None  # type: ignore[assignment]
        sklearn_cosine_similarity = None  # type: ignore[assignment]

    if TfidfVectorizer is not None and sklearn_cosine_similarity is not None:
        semantic_threshold = 0.75
        semantic_max_entries = 500
        stronger_pairs: set[tuple[int, int]] = set()
        for src_id, tgt_id, *_rest in relations:
            stronger_pairs.add((src_id, tgt_id))
            stronger_pairs.add((tgt_id, src_id))

        try:
            semantic_rows = db.execute(
                """
                SELECT id, COALESCE(title, ''), COALESCE(tags, ''), COALESCE(content, '')
                FROM knowledge_entries
                ORDER BY id DESC
                LIMIT ?
                """,
                (semantic_max_entries,),
            ).fetchall()
        except sqlite3.OperationalError:
            semantic_rows = [(e[0], e[3] or "", e[4] or "", "") for e in entries[:semantic_max_entries]]

        semantic_entries = []
        for entry_id, title, tags, content in semantic_rows:
            text = " ".join(part.strip() for part in (title, tags, content) if part and part.strip())
            if text:
                semantic_entries.append((int(entry_id), text))

        if len(semantic_entries) >= 2:
            try:
                semantic_ids = [item[0] for item in semantic_entries]
                semantic_texts = [item[1] for item in semantic_entries]
                vectorizer = TfidfVectorizer(
                    max_features=8000,
                    ngram_range=(1, 2),
                    sublinear_tf=True,
                    strip_accents="unicode",
                    min_df=1,
                    max_df=0.95,
                )
                matrix = vectorizer.fit_transform(semantic_texts)
                similarity_matrix = sklearn_cosine_similarity(matrix)
                for i, src_id in enumerate(semantic_ids):
                    for j in range(i + 1, len(semantic_ids)):
                        score = float(similarity_matrix[i, j])
                        if score < semantic_threshold:
                            continue
                        tgt_id = semantic_ids[j]
                        if (src_id, tgt_id) in stronger_pairs or (tgt_id, src_id) in stronger_pairs:
                            continue
                        _add(src_id, tgt_id, "SEMANTIC_PROXIMITY", round(score, 2))
            except ValueError:
                pass

    # Batch insert all relations
    if relations:
        db.executemany(
            """
            INSERT OR IGNORE INTO knowledge_relations
            (source_id, target_id, source_stable_id, target_stable_id, relation_type, stable_id, confidence, created_at)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
        """,
            relations,
        )
        for rel in relations:
            _enqueue_sync_op_fail_open(
                db,
                "knowledge_relations",
                rel[5],
                {
                    "source_stable_id": rel[2],
                    "target_stable_id": rel[3],
                    "relation_type": rel[4],
                    "stable_id": rel[5],
                    "confidence": rel[6],
                    "created_at": rel[7],
                },
            )

    return len(relations)


def show_stats(db: sqlite3.Connection):
    """Show extraction statistics."""
    print(f"\n{'=' * 50}")
    print("  Knowledge Extraction Statistics")
    print(f"{'=' * 50}")

    total = db.execute("SELECT COUNT(*) FROM knowledge_entries").fetchone()[0]
    print(f"  Total entries: {total}")

    print("\n  By category:")
    for row in db.execute("""
        SELECT category, COUNT(*), ROUND(AVG(confidence), 2)
        FROM knowledge_entries GROUP BY category ORDER BY COUNT(*) DESC
    """):
        print(f"    {row[0]:12s}: {row[1]:3d} entries (avg confidence: {row[2]})")

    print("\n  Top tags:")
    # Manually count tags since they're comma-separated
    tag_counts = {}
    for row in db.execute("SELECT tags FROM knowledge_entries WHERE tags != ''"):
        for tag in row[0].split(","):
            tag = tag.strip()
            if tag:
                tag_counts[tag] = tag_counts.get(tag, 0) + 1
    for tag, count in sorted(tag_counts.items(), key=lambda x: -x[1])[:15]:
        print(f"    {tag:20s}: {count}")

    print("\n  Cross-session patterns (appearing in 2+ sessions):")
    for row in db.execute("""
        SELECT title, category, COUNT(DISTINCT session_id) as sessions
        FROM knowledge_entries
        GROUP BY title, category
        HAVING sessions >= 2
        ORDER BY sessions DESC
        LIMIT 10
    """):
        print(f"    [{row[1]}] {row[0][:60]} ({row[2]} sessions)")


def list_entries(db: sqlite3.Connection, category: str = None, limit: int = 20):
    """List knowledge entries."""
    sql = "SELECT id, category, title, tags, confidence, session_id FROM knowledge_entries"
    params = []
    if category:
        sql += " WHERE category = ?"
        params.append(category)
    sql += " ORDER BY confidence DESC, category LIMIT ?"
    params.append(limit)

    print(f"\n{'ID':>4s} {'Category':12s} {'Conf':>5s} {'Session':10s} Title")
    print(f"{'─' * 4} {'─' * 12} {'─' * 5} {'─' * 10} {'─' * 50}")

    for row in db.execute(sql, params):
        sid = row[5][:8] + ".."
        print(f"{row[0]:4d} {row[1]:12s} {row[4]:5.2f} {sid:10s} {row[2][:50]}")


def _run_semantic_proximity(db: sqlite3.Connection) -> int:
    """Run SEMANTIC_PROXIMITY relation extraction using TF-IDF cosine similarity.

    Wave 17 extraction point: called by both --residual-only (backward compat via
    _run_residual_work) and --semantic-only (new mode for sk watch when sklearn IS
    available, after native Rust helpers ran backfill/task_id/decay).

    Missing scikit-learn is a silent no-op (fail-open).
    Does NOT commit — caller is responsible.
    Returns the number of SEMANTIC_PROXIMITY relations written (0 if sklearn absent).
    """
    now = datetime.now().isoformat()
    try:
        from sklearn.feature_extraction.text import TfidfVectorizer
        from sklearn.metrics.pairwise import cosine_similarity as _sklearn_cos
    except ImportError:
        TfidfVectorizer = None  # type: ignore[assignment]
        _sklearn_cos = None  # type: ignore[assignment]

    if TfidfVectorizer is None or _sklearn_cos is None:
        return 0

    semantic_threshold = 0.75
    semantic_max_entries = 500

    # Build stronger_pairs from Rust-written relations already in the DB.
    stronger_pairs: set[tuple[int, int]] = set()
    try:
        for row in db.execute("SELECT source_id, target_id FROM knowledge_relations"):
            stronger_pairs.add((row[0], row[1]))
            stronger_pairs.add((row[1], row[0]))
    except sqlite3.OperationalError:
        pass

    try:
        sem_rows = db.execute(
            """
            SELECT id, COALESCE(title,''), COALESCE(tags,''), COALESCE(content,'')
            FROM knowledge_entries ORDER BY id DESC LIMIT ?
            """,
            (semantic_max_entries,),
        ).fetchall()
    except sqlite3.OperationalError:
        sem_rows = []

    semantic_entries = []
    for eid, title, tags, content in sem_rows:
        text = " ".join(p.strip() for p in (title, tags, content) if p and p.strip())
        if text:
            semantic_entries.append((int(eid), text))

    if len(semantic_entries) < 2:
        return 0

    try:
        sem_ids = [item[0] for item in semantic_entries]
        sem_texts = [item[1] for item in semantic_entries]
        vectorizer = TfidfVectorizer(
            max_features=8000,
            ngram_range=(1, 2),
            sublinear_tf=True,
            strip_accents="unicode",
            min_df=1,
            max_df=0.95,
        )
        matrix = vectorizer.fit_transform(sem_texts)
        sim = _sklearn_cos(matrix)
        # Collect (src_id, tgt_id, stable_id, conf) for batch insert.
        all_entries_map = {
            row[0]: row
            for row in db.execute(
                "SELECT id, session_id, category, title, COALESCE(topic_key,''), COALESCE(stable_id,'')"
                " FROM knowledge_entries ORDER BY id DESC LIMIT ?",
                (semantic_max_entries,),
            ).fetchall()
        }
        sem_relations = []
        for i, src_id in enumerate(sem_ids):
            for j in range(i + 1, len(sem_ids)):
                score = float(sim[i, j])
                if score < semantic_threshold:
                    continue
                tgt_id = sem_ids[j]
                if (src_id, tgt_id) in stronger_pairs or (tgt_id, src_id) in stronger_pairs:
                    continue
                src_row = all_entries_map.get(src_id)
                tgt_row = all_entries_map.get(tgt_id)
                if not src_row or not tgt_row:
                    continue
                src_stable = src_row[5] or _knowledge_stable_id(src_row[1], src_row[2], src_row[3], src_row[4])
                tgt_stable = tgt_row[5] or _knowledge_stable_id(tgt_row[1], tgt_row[2], tgt_row[3], tgt_row[4])
                stable_id = _knowledge_relation_stable_id(src_stable, tgt_stable, "SEMANTIC_PROXIMITY")
                sem_relations.append(
                    (src_id, tgt_id, src_stable, tgt_stable, "SEMANTIC_PROXIMITY", stable_id, round(score, 2), now)
                )
        if sem_relations:
            # Ensure unique index exists before batch insert.
            db.execute("""
                CREATE UNIQUE INDEX IF NOT EXISTS idx_relations_unique
                ON knowledge_relations(source_id, target_id, relation_type)
            """)
            db.executemany(
                """
                INSERT OR IGNORE INTO knowledge_relations
                (source_id, target_id, source_stable_id, target_stable_id,
                 relation_type, stable_id, confidence, created_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                """,
                sem_relations,
            )
            for rel in sem_relations:
                _enqueue_sync_op_fail_open(
                    db,
                    "knowledge_relations",
                    rel[5],
                    {
                        "source_stable_id": rel[2],
                        "target_stable_id": rel[3],
                        "relation_type": rel[4],
                        "stable_id": rel[5],
                        "confidence": rel[6],
                        "created_at": rel[7],
                    },
                )
        return len(sem_relations)
    except (ValueError, Exception):
        return 0  # TF-IDF failure is a no-op


def _run_residual_work(db: sqlite3.Connection) -> None:
    """Residual Python work when native Rust already wrote entries and relations.

    Wave 16 boundary: Rust's native extract handles the classification/write
    loop and SAME_SESSION/SAME_TOPIC/TAG_OVERLAP/RESOLVED_BY relations.
    This function covers the remaining Python-owned work:

      1. Backfill: affected_files and task_id from session-local evidence.
      2. Confidence decay: daily sliding-window decay via embedding_meta.
      3. SEMANTIC_PROXIMITY relations: TF-IDF cosine ≥ 0.75 (scikit-learn).
         Appended on top of Rust-written rows via INSERT OR IGNORE.
         Missing scikit-learn is a silent no-op (fail-open).

    NOT run here:
      - extract_from_sections()      — done natively by Rust
      - SAME_SESSION / SAME_TOPIC / TAG_OVERLAP / RESOLVED_BY — done natively

    Called by main() when --residual-only is passed (from sk watch after a
    successful native extract pass).
    """
    now = datetime.now().isoformat()

    # 1. Backfill
    _backfill_affected_files_from_session_evidence(db)
    _infer_task_ids_from_content(db)

    # 2. Confidence decay (at most once per day)
    try:
        today = now[:10]
        last_decay = None
        try:
            db.execute("""
                CREATE TABLE IF NOT EXISTS embedding_meta (
                    key TEXT PRIMARY KEY, value TEXT
                )
            """)
            row = db.execute("SELECT value FROM embedding_meta WHERE key = 'last_decay_date'").fetchone()
            if row:
                last_decay = row[0]
        except sqlite3.OperationalError:
            pass
        if last_decay != today:
            db.execute(
                """
                UPDATE knowledge_entries
                SET confidence = MAX(0.3,
                    CASE WHEN confidence >= 0.8 THEN confidence * 0.98
                         ELSE confidence * 0.95
                    END)
                WHERE last_seen < ? AND confidence > 0.3
                """,
                (today,),
            )
            try:
                db.execute(
                    "INSERT OR REPLACE INTO embedding_meta (key, value) VALUES ('last_decay_date', ?)",
                    (today,),
                )
            except sqlite3.OperationalError:
                pass
    except sqlite3.OperationalError:
        pass

    # 3. SEMANTIC_PROXIMITY — delegated to _run_semantic_proximity() (wave 17 refactor).
    # INSERT OR IGNORE preserves the Rust-written deterministic relation rows.
    # Missing scikit-learn is a silent no-op (fail-open).
    _run_semantic_proximity(db)

    db.commit()
    print("[residual] backfill + decay + SEMANTIC_PROXIMITY done")


def main():
    args = sys.argv[1:]

    if "--help" in args or "-h" in args:
        print(__doc__)
        return

    if not DB_PATH.exists():
        print(f"Error: Knowledge database not found at {DB_PATH}")
        print("Run 'python build-session-index.py' first.")
        sys.exit(1)

    db = sqlite3.connect(str(DB_PATH))
    ensure_tables(db)

    if "--stats" in args:
        show_stats(db)
        db.close()
        return

    if "--list" in args:
        category = None
        if "--category" in args:
            idx = args.index("--category")
            category = args[idx + 1] if idx + 1 < len(args) else None
        limit = 20
        if "--limit" in args:
            idx = args.index("--limit")
            limit = int(args[idx + 1]) if idx + 1 < len(args) else 20
        list_entries(db, category, limit)
        db.close()
        return

    if "--relations" in args:
        try:
            total = db.execute("SELECT COUNT(*) FROM knowledge_relations").fetchone()[0]
            print(f"\n{'=' * 50}")
            print("  Knowledge Relations Statistics")
            print(f"{'=' * 50}")
            print(f"  Total relations: {total}")
            print("\n  By type:")
            for row in db.execute("""
                SELECT relation_type, COUNT(*), ROUND(AVG(confidence), 2)
                FROM knowledge_relations GROUP BY relation_type ORDER BY COUNT(*) DESC
            """):
                print(f"    {row[0]:15s}: {row[1]:4d} relations (avg confidence: {row[2]})")
        except sqlite3.OperationalError:
            print("  No relations found. Run extraction first.")
        db.close()
        return

    # --semantic-only: Wave 17 hot-path mode for sk watch when sklearn IS available.
    # Native Rust handles backfill, task_id, and decay.  Only SEMANTIC_PROXIMITY runs here.
    if "--semantic-only" in args:
        n = _run_semantic_proximity(db)
        db.commit()
        db.close()
        print(f"[semantic] SEMANTIC_PROXIMITY: {n} relations")
        return

    # --residual-only: Wave 16 hot-path mode.
    # Called by sk watch after native Rust extract wrote entries and deterministic
    # relations.  Python covers only the remaining residual work.
    if "--residual-only" in args:
        _run_residual_work(db)
        db.close()
        return

    # Default: run extraction
    session_ids = None
    if "--sessions" in args:
        idx = args.index("--sessions")
        session_ids = [s.strip() for s in args[idx + 1].split(",") if s.strip()] if idx + 1 < len(args) else None
        if session_ids:
            print(f"Selective extraction for {len(session_ids)} session(s)...")
    else:
        print("Extracting knowledge from indexed sessions...")
    extracted, skipped, deduped, relations_count = extract_from_sections(db, session_ids=session_ids)
    print(f"Extracted {extracted} entries ({skipped} duplicates skipped, {deduped} deduped by hash)")
    print(f"Extracted {relations_count} relations")
    avg_conf_row = db.execute("SELECT AVG(confidence) FROM knowledge_entries WHERE confidence IS NOT NULL").fetchone()
    avg_confidence = round(float(avg_conf_row[0]), 4) if avg_conf_row and avg_conf_row[0] is not None else None
    if extracted > 0:
        _emit_knowledge_event_fail_open(
            "skill_extracted",
            {
                "extracted": extracted,
                "skipped": skipped,
                "deduped": deduped,
                "relations": relations_count,
                "session_count": len(session_ids) if session_ids else 0,
                "confidence": avg_confidence,
            },
        )

    # Clean up stale embeddings and orphan relations
    try:
        stale_embeds = db.execute("""
            DELETE FROM embeddings WHERE
            (source_type = 'knowledge' AND source_id NOT IN (SELECT id FROM knowledge_entries))
            OR (source_type = 'section' AND source_id NOT IN (SELECT id FROM sections))
        """).rowcount
        orphan_rels = db.execute("""
            DELETE FROM knowledge_relations WHERE
            source_id NOT IN (SELECT id FROM knowledge_entries)
            OR target_id NOT IN (SELECT id FROM knowledge_entries)
        """).rowcount
        if stale_embeds or orphan_rels:
            db.commit()
            print(f"Cleaned up {stale_embeds} stale embeddings, {orphan_rels} orphan relations")
    except Exception:
        pass

    show_stats(db)
    db.close()


if __name__ == "__main__":
    main()