"""Distillation trigger hook — auto-distills undistilled DailyActivity files.

Checks the count of undistilled DailyActivity files after each session
close.  When any undistilled files exist, runs a lightweight
rule-based distillation directly in the hook (no agent session needed),
writing curated entries to MEMORY.md via ``_modify_content()`` under flock.

Also archives old DailyActivity files (>90 days) and enforces section
caps on MEMORY.md to prevent unbounded growth.

Falls back to the flag-file approach if direct distillation fails,
so the next agent session can pick it up.

Key public symbols:

- ``DistillationTriggerHook``  — Implements ``SessionLifecycleHook``.
- ``UNDISTILLED_THRESHOLD``    — Minimum undistilled files to trigger (0 = every session).
- ``ARCHIVE_DAYS``             — Age threshold for DailyActivity archival (90).
- ``SECTION_CAPS``             — Max entries per MEMORY.md section after distillation.
"""

from __future__ import annotations

import asyncio
import json
import logging
import re
import subprocess
from datetime import date, datetime, timedelta
from pathlib import Path

from core.session_hooks import HookContext
from core.initialization_manager import initialization_manager
from core.daily_activity_writer import parse_frontmatter, write_frontmatter, read_jsonl_sidecar
from scripts.locked_write import LockedWriteError
from hooks.evolution_maintenance_hook import _append_changelog

logger = logging.getLogger(__name__)

# Keywords that signal an implementation claim (as opposed to a decision or design choice).
# Claims containing these keywords are verified against git before promotion to MEMORY.md.
_IMPLEMENTATION_KEYWORDS = re.compile(
    r"\b(?:implement(?:ed|s|ing)?|ship(?:ped|s)?|built|creat(?:ed|ing)|"
    r"complet(?:ed|ing)|fix(?:ed)?|deploy(?:ed)?|launch(?:ed)?|"
    r"refactor(?:ed)?|migrat(?:ed)?|rewrit(?:ten|e)|"
    r"delet(?:ed|ing)|remov(?:ed|ing)|replac(?:ed|ing))\b",
    re.IGNORECASE,
)

# Regex to detect session sections in DailyActivity (## HH:MM format)
_DA_SECTION_RE = re.compile(r"^## \d{2}:\d{2}", re.MULTILINE)

UNDISTILLED_THRESHOLD = 0  # Run every session close (was 2 — caused 1-2 day staleness)
FLAG_FILENAME = ".needs_distillation"
SCAN_DAYS = 30  # Only check files from last 30 days
ARCHIVE_DAYS = 90  # Move files older than this to Archives/
SECTION_CAPS = {  # Max entries per MEMORY.md section after distillation
    "Key Decisions": 40,
    "Lessons Learned": 35,
    "Recent Context": 20,
    "COE Registry": 15,
    "Open Threads": 10,
}

# Centralized patterns — shared with summarization.py via extraction_patterns.py.
# Distillation uses the STRICT variant (runs on already-extracted DailyActivity).
from core.extraction_patterns import (
    DECISION_PATTERNS_STRICT as _DECISION_PATTERNS,
    LESSON_PATTERNS as _LESSON_PATTERNS,
    is_noise_entry as _is_noise_entry,
)

# Competence patterns: "now I know how to X" — positive capability acquisition.
# Distinct from lessons ("next time avoid X") which are corrective/negative.
# These stay here (not in extraction_patterns) because they're only used by distillation.
_COMPETENCE_PATTERNS = re.compile(
    r"(?:"
    # First-person learning
    r"(?:now )?(?:know|knows|understand|understands) (?:how to|that|about)|"
    r"can now|is able to|figured out how to|discovered that|"
    r"the (?:way|trick|method|pattern|technique) (?:is|to)|"
    # Technical facts: "X works by Y", "X points to Y"
    r"(?:works by|achieved by|accomplished via|done with|built using)|"
    r"(?:verified|confirmed) (?:that|working|live|in production)|"
    r"(?:root cause|fix) (?:was|is|:)|"
    r"(?:pipeline|system|hook|engine) (?:works|fires|runs|produces)|"
    r"uses? [\w./]+ (?:instead of|rather than|over)|"
    r"(?:requires?|needs?) (?:both |\w+ (?:to|for|before|and ))|"
    # Declarative facts: "X does not source Y", "X rejects Y"
    r"(?:does not|doesn't|cannot|can't|do not|don't) (?:source|inherit|load|"
    r"see|find|support|work|include|have|resolve|recognize|persist)|"
    r"(?:don't|do not|never) (?:\w[\w-]* ){1,3}(?:without|before|until) |"
    r"(?:blocks?|rejects?|prevents?) [\w/]+ \w+|"
    # Imperative guidance: "Use X to Y", "prefer X over Y"
    r"(?:use |prefer |always )(?:\w+ ){1,3}(?:to |for |when |instead |before |during )|"
    # Requirement patterns: "must run/use/be"
    r"must (?:run|use|be|have|call|spawn|import)|"
    # Structural observations: "X hierarchy: Y", "X order: Y"
    r"\w+ (?:hierarchy|order|precedence|priority): |"
    # Factual pointers: "X points to Y", "X refers to Y"
    r"(?:points? to|refers? to|maps? to) |"
    # Architecture: "content must be preserved in ref"
    r"(?:content|state|data) must be (?:preserved|stored|saved|maintained)"
    r")",
    re.IGNORECASE,
)


def _extract_repo_paths(tech_md_content: str) -> list[Path]:
    """Extract git repository paths from a TECH.md file.

    Looks for common patterns:
    - ``Clone: `git clone <url>` ``
    - ``Local: /path/to/repo``
    - ``Codebase location: /path/to/repo``
    - Any absolute path ending in a git-repo-like directory

    Returns a list of Path objects (may be empty, may not exist on disk).
    """
    candidates: list[Path] = []
    for line in tech_md_content.splitlines():
        # Match lines that reference repo locations
        if any(kw in line for kw in ("Clone:", "local:", "Local:",
                                      "Codebase", "codebase", "repo",
                                      "source", "Source")):
            # Extract absolute paths
            paths = re.findall(r"(/[^\s`\"']+)", line)
            for p in paths:
                # Clean trailing punctuation
                p = p.rstrip("/),;.")
                if len(p) > 5:  # Skip trivially short paths
                    candidates.append(Path(p))
    return candidates


class DistillationTriggerHook:
    """Checks undistilled DailyActivity count and runs direct distillation.

    Distills directly using ``_modify_content()`` under flock (single lock
    acquisition per section).  If direct distillation fails, falls back to
    writing a ``.needs_distillation`` flag for the next agent session.
    """

    name = "distillation_trigger"

    async def execute(self, context: HookContext) -> None:
        """Scan DailyActivity files and distill if threshold exceeded."""
        ws_path = initialization_manager.get_cached_workspace_path()
        da_dir = Path(ws_path) / "Knowledge" / "DailyActivity"

        if not da_dir.exists():
            return

        undistilled_files = await asyncio.to_thread(
            self._get_undistilled_files, da_dir
        )

        if len(undistilled_files) <= UNDISTILLED_THRESHOLD:
            logger.debug(
                "Undistilled count %d <= %d, no distillation needed",
                len(undistilled_files),
                UNDISTILLED_THRESHOLD,
            )
            return

        logger.info(
            "Distillation threshold exceeded (%d > %d), running direct distillation",
            len(undistilled_files),
            UNDISTILLED_THRESHOLD,
        )

        # Auto-archive old DailyActivity files (>90 days)
        try:
            archived = await asyncio.to_thread(
                self._archive_old_files, da_dir, Path(ws_path)
            )
            if archived:
                logger.info("Archived %d old DailyActivity files", archived)
        except Exception as exc:
            logger.warning("DailyActivity archival failed (non-blocking): %s", exc)

        # L4: Update effectiveness scoring before distillation
        # Compare last briefing suggestions against actual deliverables
        try:
            self._update_effectiveness_scoring(undistilled_files, Path(ws_path))
        except Exception as exc:
            logger.debug("Effectiveness scoring update failed (non-blocking): %s", exc)

        # Attempt direct distillation
        try:
            distilled_count = await asyncio.to_thread(
                self._distill_files, undistilled_files, Path(ws_path)
            )
            logger.info(
                "Direct distillation complete: %d files processed, entries promoted to MEMORY.md",
                distilled_count,
            )
            # Clean up any stale flag file
            flag_path = da_dir / FLAG_FILENAME
            if flag_path.exists():
                flag_path.unlink(missing_ok=True)
        except Exception as exc:
            logger.warning(
                "Direct distillation failed (%s), falling back to flag file",
                exc,
            )
            self._write_flag(da_dir, len(undistilled_files))

        # Archive stale Recent Context entries from MEMORY.md (non-blocking)
        try:
            await asyncio.to_thread(
                self._archive_stale_rc_entries,
                Path(ws_path) / ".context" / "MEMORY.md",
                Path(ws_path),
            )
        except Exception as exc:
            logger.warning("Stale RC archival failed (non-blocking): %s", exc)

    @staticmethod
    def _update_effectiveness_scoring(
        da_files: list[Path], ws_path: Path,
    ) -> None:
        """Compare briefing suggestions against actual deliverables.

        Reads proactive_state.json for last_briefing_suggested, extracts
        deliverables from the DailyActivity files being distilled, and
        calls _update_effectiveness() to adjust scoring.

        Non-fatal — any failure is logged and skipped.
        """
        from core.proactive_intelligence import (
            _load_learning_state,
            _save_learning_state,
            _update_effectiveness,
        )

        state = _load_learning_state(ws_path)
        last_suggested = state.last_briefing_suggested
        if not last_suggested:
            return

        # Extract deliverables from DailyActivity files
        deliverables: list[str] = []
        for da_file in da_files:
            try:
                content = da_file.read_text(encoding="utf-8")
            except (OSError, UnicodeDecodeError):
                continue
            in_deliverables = False
            for line in content.splitlines():
                stripped = line.strip()
                if stripped == "**Deliverables:**" or stripped.startswith("### Deliverables"):
                    in_deliverables = True
                    continue
                if in_deliverables and (
                    stripped.startswith("**") and stripped.endswith(":**")
                    or stripped.startswith("### ")
                    or stripped.startswith("## ")
                ):
                    in_deliverables = False
                    continue
                if in_deliverables and stripped.startswith("- "):
                    deliverables.append(stripped.lstrip("- ").strip())

        if deliverables:
            _update_effectiveness(state, last_suggested, deliverables)
            _save_learning_state(ws_path, state)
            logger.info(
                "L4 effectiveness updated: %d suggestions vs %d deliverables, "
                "follow_rate=%.2f, trend=%s",
                len(last_suggested), len(deliverables),
                state.effectiveness.get("follow_rate", 0),
                state.effectiveness.get("trend", "gathering"),
            )

    @staticmethod
    def _get_undistilled_files(da_dir: Path) -> list[Path]:
        """Get DailyActivity files where distilled != true.

        Only checks files from the last 30 days to bound scan scope.
        Returns sorted list (oldest first) for chronological processing.
        """
        files = []
        cutoff = date.today() - timedelta(days=SCAN_DAYS)
        for f in da_dir.glob("*.md"):
            try:
                file_date = date.fromisoformat(f.stem)
                if file_date < cutoff:
                    continue
            except ValueError:
                continue  # Skip non-date filenames
            try:
                content = f.read_text(encoding="utf-8")
            except (OSError, UnicodeDecodeError):
                continue
            if not _is_distilled(content):
                files.append(f)
        return sorted(files, key=lambda f: f.stem)

    def _distill_files(self, files: list[Path], ws_path: Path) -> int:
        """Extract entries from DailyActivity files and write to MEMORY.md + EVOLUTION.md.

        Extracts four categories:
        - Decisions → MEMORY.md "Key Decisions"
        - Lessons → MEMORY.md "Lessons Learned"
        - COE signals → MEMORY.md "COE Registry" (cross-session problem tracking)
        - Corrections → EVOLUTION.md "Corrections Captured" (agent behavior fixes)

        All entries per section are batched into a single locked_write call
        to minimize lock acquisitions (was: one call per entry).

        Returns the number of files successfully distilled.
        """
        memory_path = ws_path / ".context" / "MEMORY.md"
        evolution_path = ws_path / ".context" / "EVOLUTION.md"

        distilled_count = 0
        coe_entries: list[tuple[str, str, str]] = []  # (date, signal, topic)
        all_corrections: list[tuple[str, str]] = []  # (date, correction)
        all_competence: list[tuple[str, str]] = []  # (date, competence)

        # Collect all entries across files, then write once per section.
        # Track which files were successfully extracted (marked distilled
        # AFTER writes succeed — see GAP 12).
        all_decisions: list[str] = []
        all_lessons: list[str] = []
        extracted_files: list[tuple[Path, dict, str]] = []  # (path, frontmatter, body)

        for da_file in files:
            try:
                content = da_file.read_text(encoding="utf-8")
                fm, body = parse_frontmatter(content)
                file_date = da_file.stem  # YYYY-MM-DD
                da_rel_path = f"DailyActivity/{da_file.name}"

                # --- Prefer JSONL sidecar (structured, lossless) over regex ---
                jsonl_path = da_file.with_suffix(".jsonl")
                jsonl_records = read_jsonl_sidecar(jsonl_path)

                if jsonl_records:
                    # JSONL path: consume LLM-extracted structured data directly
                    decisions, lessons, corrections, competence = [], [], [], []
                    day_commit = None

                    for rec in jsonl_records:
                        decisions.extend(rec.get("decisions", []))
                        lessons.extend(rec.get("lessons", []))
                        corrections.extend(rec.get("corrections", []))
                        # process_reflection → promote as lesson (high-value LLM meta-analysis)
                        pr = rec.get("process_reflection", "")
                        if pr and len(pr) > 20:
                            lessons.append(pr)
                        # signal_driven_actions → promote as decisions (causal link)
                        for sda in rec.get("signal_driven_actions", []):
                            if sda and len(sda) > 15:
                                decisions.append(sda)
                        # COE signals
                        sig = rec.get("coe_signal", "")
                        topic = rec.get("coe_topic", "")
                        if sig and topic:
                            coe_entries.append((file_date, sig, topic))
                        # Git commits for provenance
                        commits = rec.get("git_commits", [])
                        if commits and not day_commit:
                            # First 7 chars of first commit hash
                            first = commits[0].split()[0] if commits[0] else ""
                            day_commit = first[:7] if first else None

                    logger.debug(
                        "Distilled %s via JSONL: %d decisions, %d lessons, %d corrections",
                        da_file.name, len(decisions), len(lessons), len(corrections),
                    )
                else:
                    # Regex fallback path: parse markdown (legacy, pre-JSONL files)
                    decisions = self._extract_decisions(body)
                    lessons = self._extract_lessons(body)
                    corrections = self._extract_corrections(body)
                    competence = self._extract_competence(body)
                    day_commit = self._extract_day_commit(body)

                    logger.debug(
                        "Distilled %s via regex: %d decisions, %d lessons, %d corrections, %d competence",
                        da_file.name, len(decisions), len(lessons), len(corrections), len(competence),
                    )

                # Batch entries with enriched provenance (write happens after the loop)
                for decision in decisions:
                    all_decisions.append(self._format_enriched_entry(
                        decision, file_date, da_rel_path, day_commit,
                    ))
                for lesson in lessons:
                    all_lessons.append(self._format_enriched_entry(
                        lesson, file_date, da_rel_path, day_commit,
                    ))

                # Collect corrections and competence for EVOLUTION.md
                for correction in corrections:
                    all_corrections.append((file_date, correction))
                for comp in competence:
                    all_competence.append((file_date, comp))

                # Collect COE signals from body (in case JSONL didn't have them)
                if not jsonl_records:
                    coe_items = self._extract_coe_entries(body)
                    if coe_items:
                        for signal, topic in coe_items:
                            coe_entries.append((file_date, signal, topic))

                # Track for post-write marking (NOT marked here — see GAP 12)
                extracted_files.append((da_file, fm, body))

                distilled_count += 1
            except Exception as exc:
                logger.warning("Failed to distill %s: %s", da_file.name, exc)
                continue

        # --- Also scan job result JSONL sidecars for agent-task insights ---
        # Agent-task jobs (morning-inbox, team-checkin, etc.) produce analysis
        # worth distilling. Only JSONL sidecars (tokens_used>0) contain real insights.
        job_results_dir = ws_path / "Knowledge" / "JobResults"
        if job_results_dir.is_dir():
            cutoff_date = (date.today() - timedelta(days=SCAN_DAYS)).isoformat()
            for jsonl_file in sorted(job_results_dir.glob("*.jsonl")):
                # Only process recent files (same SCAN_DAYS window as DA)
                if jsonl_file.stem < cutoff_date:
                    continue
                try:
                    records = read_jsonl_sidecar(jsonl_file)
                    for rec in records:
                        # Only agent tasks produce real insights
                        if rec.get("tokens_used", 0) == 0:
                            continue
                        result_text = rec.get("result_text", "")
                        if not result_text or len(result_text) < 50:
                            continue
                        file_date = jsonl_file.stem[:10]  # YYYY-MM-DD from filename
                        job_name = rec.get("job_name", rec.get("job_id", "unknown"))
                        # Extract decisions and lessons from job output using existing regex
                        decisions = self._extract_decisions(result_text)
                        lessons = self._extract_lessons(result_text)
                        da_rel_path = f"JobResults/{jsonl_file.name}"
                        for d in decisions:
                            all_decisions.append(self._format_enriched_entry(d, file_date, da_rel_path, None))
                        for l in lessons:
                            all_lessons.append(self._format_enriched_entry(l, file_date, da_rel_path, None))
                        if decisions or lessons:
                            logger.debug(
                                "Distilled job %s (%s): %d decisions, %d lessons",
                                job_name, jsonl_file.name, len(decisions), len(lessons),
                            )
                except Exception as exc:
                    logger.debug("Job JSONL distill failed for %s: %s", jsonl_file.name, exc)

        # Git-verify implementation claims before promotion (COE C005 fix).
        # Claims with implementation keywords that don't match any git commit
        # get tagged [UNVERIFIED] to prevent confident false memories.
        source_repo = self._get_source_repo_path()
        if source_repo:
            all_decisions = self._tag_unverified_claims(all_decisions, source_repo)
            all_lessons = self._tag_unverified_claims(all_lessons, source_repo)

        # Supersede older entries about the same topic before writing.
        # Without this, "L0+L1 implemented" (session A) and "L0-L4
        # implemented" (session B) BOTH get promoted — the stale claim
        # persists and poisons future sessions.  See COE: memory pipeline
        # temporal lag gap (2026-03-19).
        all_decisions = self._supersede_by_topic(all_decisions)
        all_lessons = self._supersede_by_topic(all_lessons)

        # Frequency gate: only promote entries that appear in >=2 DA files.
        # One-off observations stay in DailyActivity; recurring themes graduate.
        # This creates the compound loop: repeated patterns → MEMORY.md → better agent.
        da_files_data = [
            {"path": str(f), "date": f.stem, "body": f.read_text(encoding="utf-8")}
            for f in sorted(
                (ws_path / "Knowledge" / "DailyActivity").glob("*.md"),
                reverse=True,
            )[:SCAN_DAYS]
        ] if (ws_path / "Knowledge" / "DailyActivity").is_dir() else []

        if da_files_data:
            all_decisions = [
                e for e in all_decisions
                if self._passes_frequency_gate(e, da_files_data)
            ]
            all_lessons = [
                e for e in all_lessons
                if self._passes_frequency_gate(e, da_files_data)
            ]

        # Batched writes to MEMORY.md — one lock acquisition per section
        if all_decisions:
            self._run_locked_write(
                memory_path, "Key Decisions", "\n".join(all_decisions)
            )
        if all_lessons:
            self._run_locked_write(
                memory_path, "Lessons Learned", "\n".join(all_lessons)
            )

        # Write COE registry entries
        if coe_entries:
            self._write_coe_registry(memory_path, coe_entries)

        # Auto-manage Open Threads from COE signals (code-enforced)
        if coe_entries:
            self._update_open_threads(memory_path, coe_entries)

        # Auto-resolve Open Threads whose topics appear in deliverables.
        # This closes the staleness gap: when work is done (captured in
        # DailyActivity "Deliverables" section), the corresponding Open
        # Thread is moved to "Resolved" automatically — no manual cleanup.
        # See COE: memory pipeline temporal lag gap (2026-03-19).
        self._reconcile_open_threads_from_deliverables(
            memory_path, [f for f, _, _ in extracted_files]
        )

        # Write corrections and competence to EVOLUTION.md
        if all_corrections:
            self._write_corrections(evolution_path, all_corrections)
            # Auto-trigger steeringify proposals when new corrections are written
            self._check_steeringify_proposals(evolution_path, ws_path)
            # Cross-loop signal: notify evolution pipeline that new corrections
            # are available. If ≥5 pending, evolution_maintenance_hook will run
            # the cycle on next session close without waiting for the 7-day timer.
            self._signal_evolution_corrections(ws_path, len(all_corrections))
        if all_competence:
            self._write_competence(evolution_path, all_competence)

        # Enforce section caps on MEMORY.md to prevent unbounded growth.
        # Always run — caps must be enforced even when no new entries were added
        # this cycle (prior runs may have accumulated entries past the cap).
        self._enforce_section_caps(memory_path, ws_path)

        # Mark files as distilled AFTER all writes succeed.
        # Previous ordering (mark inside loop, write after) could lose entries:
        # if batch writes failed, already-marked files wouldn't be re-extracted.
        for da_file, fm, body in extracted_files:
            try:
                fm["distilled"] = True
                fm["distilled_date"] = date.today().isoformat()
                new_content = write_frontmatter(fm, body)
                da_file.write_text(new_content, encoding="utf-8")
            except Exception as exc:
                logger.warning("Failed to mark %s as distilled: %s", da_file.name, exc)
                # Non-fatal: file will be re-processed next time, but dedup
                # in _run_locked_write prevents duplicate MEMORY.md entries.

        # Log distillation to EVOLUTION_CHANGELOG.jsonl
        if distilled_count > 0:
            changelog_path = ws_path / ".context" / "EVOLUTION_CHANGELOG.jsonl"
            _append_changelog(
                changelog_path,
                "distill",
                f"batch-{date.today().isoformat()}",
                f"Distilled {distilled_count} DailyActivity file(s) to MEMORY.md",
                source="distillation_hook",
            )

        return distilled_count

    @staticmethod
    def _format_enriched_entry(
        text: str,
        date_str: str,
        source_file: str,
        commit_hash: str | None = None,
    ) -> str:
        """Format a distillation entry with source provenance.

        Enriched format: summary + source file + optional commit hash.
        This connects Brain entries back to Library for episodic recall.

        Args:
            text: The distilled entry text (decision, lesson, etc.).
            date_str: Date string (YYYY-MM-DD).
            source_file: Relative path to the source DailyActivity file.
            commit_hash: Optional git commit hash associated with the entry.

        Returns:
            Formatted entry string like:
            ``- 2026-03-23: **Two credential chains** — summary text.
            Detail: DailyActivity/2026-03-23.md, commit aca865b.``
        """
        entry = f"- {date_str}: {text}"
        detail_parts = [f"Detail: {source_file}"]
        if commit_hash:
            detail_parts.append(f"commit {commit_hash}")
        entry += f"\n  {', '.join(detail_parts)}."
        # P2 Temporal validity: add valid_from metadata to every new entry
        try:
            from core.memory_index import add_temporal_metadata_to_entry
            entry = add_temporal_metadata_to_entry(entry, valid_from=date_str)
        except ImportError:
            pass  # Module not available yet (graceful)
        return entry

    @staticmethod
    def _extract_day_commit(body: str) -> str | None:
        """Extract the first git commit hash from a DailyActivity body.

        Looks for patterns like ``commit abc1234`` or backtick-wrapped hashes
        in **Git activity:** sections.
        """
        # Look for **Git activity:** section
        git_match = re.search(r"\*\*Git activity:\*\*\s*\n((?:- .+\n?)+)", body)
        if git_match:
            # Extract first commit hash (7+ hex chars after backtick)
            hash_match = re.search(r"`([0-9a-f]{7,40})", git_match.group(1))
            if hash_match:
                return hash_match.group(1)[:7]

        # Fallback: look for commit hashes anywhere
        hash_match = re.search(r"commit[s ]?\s*`?([0-9a-f]{7,40})", body, re.IGNORECASE)
        if hash_match:
            return hash_match.group(1)[:7]

        return None

    @staticmethod
    def _extract_decisions(body: str) -> list[str]:
        """Extract decision-worthy lines from DailyActivity body.

        Handles both old format (### Key Decisions) and new format
        (**Decisions:**) section headers.
        """
        decisions = []
        in_decisions_section = False
        for line in body.splitlines():
            stripped = line.strip()
            # Track Decisions sections — both old and new format
            if stripped.startswith("### Key Decisions") or stripped == "**Decisions:**":
                in_decisions_section = True
                continue
            # Exit section on next header (### or **bold:**)
            if in_decisions_section and (
                stripped.startswith("### ")
                or stripped.startswith("## ")
                or (stripped.startswith("**") and stripped.endswith(":**"))
            ):
                in_decisions_section = False
                continue
            # Lines in Decisions sections
            if in_decisions_section and stripped.startswith("- ") and stripped != "- (none)":
                entry = stripped[2:].strip()
                if len(entry) > 15 and not _is_noise_entry(entry):
                    decisions.append(entry[:200])
            # Lines elsewhere that match decision patterns
            elif stripped.startswith("- ") and _DECISION_PATTERNS.search(stripped):
                entry = stripped[2:].strip()
                if len(entry) > 15 and not _is_noise_entry(entry):
                    decisions.append(entry[:200])
        return decisions[:10]  # Cap to prevent MEMORY.md bloat

    @staticmethod
    def _extract_lessons(body: str) -> list[str]:
        """Extract lesson-worthy lines from DailyActivity body.

        Three extraction modes:
        1. Section-based: all items under **Lessons:** section (new format)
        2. Process reflection: the **Process Reflection:** line (LLM meta-analysis)
        3. Pattern-based: any line matching lesson patterns (both formats)
        """
        lessons = []
        in_lessons_section = False

        # Mode 2: Extract process reflection (single line, not a section)
        for line in body.splitlines():
            stripped = line.strip()
            if stripped.startswith("**Process Reflection:**"):
                entry = stripped.removeprefix("**Process Reflection:**").strip()
                if len(entry) > 20:
                    lessons.append(entry[:200])
                break  # Only one per session

        for line in body.splitlines():
            stripped = line.strip()
            # Track Lessons section (new format)
            if stripped == "**Lessons:**":
                in_lessons_section = True
                continue
            if in_lessons_section and (
                stripped.startswith("## ")
                or (stripped.startswith("**") and stripped.endswith(":**"))
            ):
                in_lessons_section = False
                continue
            # All items in Lessons section are lessons
            if in_lessons_section and stripped.startswith("- "):
                entry = stripped[2:].strip()
                if len(entry) > 15 and entry != "(none)":
                    lessons.append(entry[:200])
                continue
            # Pattern-based extraction (works on any format)
            if stripped.startswith("- ") and _LESSON_PATTERNS.search(stripped):
                entry = stripped[2:].strip()
                if len(entry) > 15 and entry != "(none)":
                    lessons.append(entry[:200])
        return lessons[:5]  # Cap to prevent MEMORY.md bloat

    @staticmethod
    def _extract_corrections(body: str) -> list[str]:
        """Extract user corrections of agent behavior from DailyActivity body.

        Looks for the **Corrections:** section written by the enriched
        DailyActivity format.
        """
        corrections = []
        in_corrections_section = False
        for line in body.splitlines():
            stripped = line.strip()
            if stripped == "**Corrections:**":
                in_corrections_section = True
                continue
            if in_corrections_section and (
                stripped.startswith("## ")
                or (stripped.startswith("**") and stripped.endswith(":**"))
            ):
                in_corrections_section = False
                continue
            if in_corrections_section and stripped.startswith("- "):
                entry = stripped[2:].strip()
                if len(entry) > 10 and entry != "(none)":
                    corrections.append(entry[:200])
        return corrections[:10]  # Cap

    @staticmethod
    def _extract_competence(body: str) -> list[str]:
        """Extract competence entries from DailyActivity body.

        Competence = "now I know how to X" — positive capability knowledge.
        Sources: **Lessons:** section items matching competence patterns,
        plus any line in the body matching competence patterns.

        Excludes items that also match lesson patterns (corrective/negative)
        to avoid double-counting.
        """
        competence = []
        in_lessons_section = False
        for line in body.splitlines():
            stripped = line.strip()
            if stripped == "**Lessons:**":
                in_lessons_section = True
                continue
            if in_lessons_section and (
                stripped.startswith("## ")
                or (stripped.startswith("**") and stripped.endswith(":**"))
            ):
                in_lessons_section = False
                continue
            if not stripped.startswith("- "):
                continue
            entry = stripped[2:].strip()
            if len(entry) <= 15 or entry == "(none)":
                continue
            # Must match competence pattern
            if not _COMPETENCE_PATTERNS.search(entry):
                continue
            # Skip purely corrective entries ("should have", "next time avoid")
            # but allow entries that are both competence AND lesson-like
            if re.search(r"(?:mistake was|should have|next time (?:avoid|don't))", entry, re.IGNORECASE):
                continue
            competence.append(entry[:200])
        return competence[:5]  # Cap

    @staticmethod
    def _supersede_by_topic(entries: list[str]) -> list[str]:
        """Keep only the newest entry per topic when multiple entries reference the same subject.

        Entries have the format ``- YYYY-MM-DD: <content>``.  When two
        entries share enough key terms (implementation, status, feature
        names) to be about the same topic, only the chronologically
        latest one survives.  This prevents stale implementation claims
        from older sessions persisting alongside newer, more accurate ones.

        The algorithm:
        1. Extract topic fingerprint from each entry (lowercased noun
           phrases, stripped of dates and filler words).
        2. Group entries whose fingerprints overlap by ≥50%.
        3. Within each group, keep only the entry with the latest date.

        Returns the de-duped list in original order.
        """
        if len(entries) <= 1:
            return entries

        # Stop words that don't contribute to topic identity
        _STOP = frozenset({
            "a", "an", "the", "is", "was", "are", "were", "be", "been",
            "to", "of", "in", "for", "on", "at", "by", "with", "from",
            "and", "or", "not", "no", "but", "that", "this", "it",
            "all", "now", "have", "has", "had", "do", "does", "did",
            "will", "would", "should", "could", "can", "may", "might",
            "our", "we", "i", "my", "me", "us", "its", "their",
            "also", "just", "still", "only", "very", "too", "so",
        })

        def _fingerprint(text: str) -> set[str]:
            """Extract meaningful words as a topic fingerprint.

            Handles both English words and CJK character sequences (LL06 fix:
            Chinese entries had empty fingerprints because \\w+ regex only
            matched English, causing duplicates to never be grouped).
            """
            # Strip leading "- YYYY-MM-DD: " prefix
            cleaned = re.sub(r"^-\s*\d{4}-\d{2}-\d{2}:\s*", "", text)
            # Strip markdown bold
            cleaned = cleaned.replace("**", "")
            # English words (existing)
            words = re.findall(r"[a-zA-Z][a-zA-Z0-9_.-]+", cleaned.lower())
            en_tokens = {w for w in words if w not in _STOP and len(w) > 2}
            # CJK tokens: extract runs of 2+ CJK characters as fingerprint units
            cjk_tokens = set(re.findall(r"[一-鿿]{2,}", cleaned))
            return en_tokens | cjk_tokens

        def _extract_date(text: str) -> str:
            """Extract YYYY-MM-DD from entry prefix."""
            m = re.match(r"^-\s*(\d{4}-\d{2}-\d{2})", text)
            return m.group(1) if m else "0000-00-00"

        # Build fingerprints
        fps = [(_fingerprint(e), _extract_date(e), i, e) for i, e in enumerate(entries)]

        # Greedy grouping: assign each entry to first overlapping group
        groups: list[list[int]] = []  # list of entry-index lists
        assigned: set[int] = set()

        for idx in range(len(fps)):
            if idx in assigned:
                continue
            group = [idx]
            assigned.add(idx)
            fp_i = fps[idx][0]
            if not fp_i:
                groups.append(group)
                continue

            for jdx in range(idx + 1, len(fps)):
                if jdx in assigned:
                    continue
                fp_j = fps[jdx][0]
                if not fp_j:
                    continue
                overlap = fp_i & fp_j
                # ≥30% overlap of the SMALLER fingerprint → same topic.
                # Threshold tuned for real entries: "Proactive Intelligence"
                # appearing in both entries (2 words) should match even when
                # surrounding details differ (version numbers, status verbs).
                # Guard: both fingerprints must have ≥3 significant words
                # for English. CJK tokens are much more specific (a single
                # 5-char Chinese phrase like "两套系统做同一件事" is equivalent
                # to 3+ English words in specificity). Count CJK tokens as 3x.
                min_size = min(len(fp_i), len(fp_j))
                # Effective size: CJK tokens count triple (high specificity)
                has_cjk_overlap = any(
                    len(t) >= 4 and ord(t[0]) >= 0x4E00 for t in overlap
                )
                effective_threshold = 1 if has_cjk_overlap else 3
                if min_size >= effective_threshold and len(overlap) >= max(1 if has_cjk_overlap else 2, min_size * 0.3):
                    group.append(jdx)
                    assigned.add(jdx)
            groups.append(group)

        # Within each group, keep only the entry with the latest date
        keep_indices: set[int] = set()
        for group in groups:
            if len(group) == 1:
                keep_indices.add(group[0])
            else:
                # Sort by date descending, keep newest.
                # Tiebreaker: higher index = appended later = newer.
                best = max(group, key=lambda i: (fps[i][1], i))
                keep_indices.add(best)

        return [e for i, e in enumerate(entries) if i in keep_indices]

    @staticmethod
    def _passes_frequency_gate(
        entry: str,
        da_files: list[dict],
        min_mentions: int = 2,
    ) -> bool:
        """Check whether an entry topic appears in enough DailyActivity files.

        An entry must be mentioned in ``min_mentions`` (default 2) distinct
        DailyActivity files before it qualifies for promotion to MEMORY.md.
        One-off observations stay in DailyActivity; recurring themes graduate.

        Uses the same fingerprinting approach as ``_supersede_by_topic``:
        extract significant words, check overlap with each DA body.

        Match threshold: ``max(2, min(4, len(fp) * 0.3))`` — at least 2
        words must match, capped at 4 so rich entries (10+ words) don't
        need unreasonably many matches.

        Cold-start safety: when only 1 DA file exists (first sessions),
        the gate passes unconditionally — don't block onboarding.

        Marathon session bypass: a DailyActivity file with >30 session
        sections is a marathon (e.g., CMHK full-day sprint). Entries from
        marathon files pass unconditionally — they represent intensive,
        high-value work that shouldn't wait for a second mention.

        Returns ``True`` when the gate passes (entry may be promoted).
        """
        if not da_files or len(da_files) < min_mentions:
            return True  # Not enough history → let through (cold start)

        # Marathon bypass: if any source DA file has >30 session sections,
        # it represents a marathon session — entries pass unconditionally.
        for df in da_files:
            section_count = len(_DA_SECTION_RE.findall(df.get("body", "")))
            if section_count > 30:
                return True  # Marathon session → bypass frequency gate

        # Build fingerprint — reuses _supersede_by_topic stop words / logic
        _STOP = frozenset({
            "a", "an", "the", "is", "was", "are", "were", "be", "been",
            "to", "of", "in", "for", "on", "at", "by", "with", "from",
            "and", "or", "not", "no", "but", "that", "this", "it",
            "all", "now", "have", "has", "had", "do", "does", "did",
            "will", "would", "should", "could", "can", "may", "might",
            "our", "we", "i", "my", "me", "us", "its", "their",
            "also", "just", "still", "only", "very", "too", "so",
        })
        cleaned = re.sub(r"^-?\s*\d{4}-\d{2}-\d{2}[:\s]*", "", entry)
        cleaned = cleaned.replace("**", "")
        fp = {
            w
            for w in re.findall(r"[a-zA-Z][a-zA-Z0-9_.-]+", cleaned.lower())
            if w not in _STOP and len(w) > 2
        }
        if len(fp) < 2:
            return True  # Too short to fingerprint → let through

        # Capped threshold: at least 2 words, at most 4.
        # Without the cap, a 14-word fingerprint needs 4.2 matches — too
        # strict for entries where DA files describe the same topic with
        # different wording.
        threshold = max(2, min(4, len(fp) * 0.3))

        mention_count = 0
        for df in da_files:
            body_lower = df["body"].lower()
            matches = sum(1 for term in fp if term in body_lower)
            if matches >= threshold:
                mention_count += 1
                if mention_count >= min_mentions:
                    return True
        return False

    @staticmethod
    def _get_source_repo_path() -> Path | None:
        """Resolve a source repo path for git verification.

        Scans ALL projects' TECH.md for repo paths (not just SwarmAI).
        No hardcoded paths — discovery is purely config-driven.
        Returns the first valid git repo found, or None.
        """
        candidates: list[Path] = []

        try:
            ws = Path(initialization_manager.get_cached_workspace_path())
            projects_dir = ws / "Projects"
            if projects_dir.is_dir():
                for project_dir in sorted(projects_dir.iterdir()):
                    tech_md = project_dir / "TECH.md"
                    if not tech_md.is_file():
                        continue
                    try:
                        content = tech_md.read_text(encoding="utf-8")
                        candidates.extend(_extract_repo_paths(content))
                    except (OSError, UnicodeDecodeError):
                        continue
        except Exception:
            pass

        for path in candidates:
            if path.is_dir() and (path / ".git").exists():
                return path
        return None

    @staticmethod
    def _verify_claim_against_git(claim: str, repo_path: Path) -> bool:
        """Check if an implementation claim is supported by git history.

        Non-implementation claims (no implementation keywords) return True
        immediately — they don't need git verification.

        For implementation claims, extracts key subject words and searches
        git log. Returns True if ANY matching commit is found.

        Args:
            claim: The text of the decision/lesson being promoted.
            repo_path: Path to the git repository.

        Returns:
            True if claim is verified (or not an implementation claim),
            False if claim makes an unverifiable implementation assertion.
        """
        # Only verify claims that contain implementation keywords
        if not _IMPLEMENTATION_KEYWORDS.search(claim):
            return True

        # Extract subject words for git search (remove common filler)
        _FILLER = frozenset({
            "a", "an", "the", "is", "was", "were", "are", "be", "been",
            "to", "of", "in", "for", "on", "at", "by", "with", "from",
            "and", "or", "not", "no", "but", "that", "this", "it",
            "all", "now", "have", "has", "had", "fully", "successfully",
            "we", "our", "i", "my", "also", "just", "still", "only",
        })
        # Strip date prefix and markdown
        cleaned = re.sub(r"^\d{4}-\d{2}-\d{2}:\s*", "", claim)
        cleaned = cleaned.replace("**", "").replace("`", "")
        words = re.findall(r"[a-zA-Z][a-zA-Z0-9_.-]+", cleaned.lower())
        subject_words = [w for w in words if w not in _FILLER and len(w) > 2
                         and w not in {kw.rstrip("edsing") for kw in
                         ["implemented", "shipped", "built", "created",
                          "completed", "fixed", "deployed", "launched",
                          "refactored", "migrated", "rewritten",
                          "deleted", "removed", "replaced"]}]

        if not subject_words:
            return True  # No meaningful subject to search for

        # Search git log for each significant subject word
        # Use the top 3 most distinctive words (longer = more distinctive)
        search_words = sorted(subject_words, key=len, reverse=True)[:3]

        for word in search_words:
            try:
                result = subprocess.run(
                    ["git", "log", "--all", f"--grep={word}",
                     "--oneline", "--max-count=3", "-i"],
                    cwd=str(repo_path),
                    capture_output=True, text=True, timeout=5,
                )
                if result.returncode == 0 and result.stdout.strip():
                    return True  # Found matching commit
            except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
                return True  # Can't verify — give benefit of the doubt

        # Tier 2: check file names — feature exists but commit messages don't mention it
        for word in search_words[:2]:
            try:
                result = subprocess.run(
                    ["git", "ls-files", f"*{word}*"],
                    cwd=str(repo_path),
                    capture_output=True, text=True, timeout=5,
                )
                if result.returncode == 0 and result.stdout.strip():
                    return True
            except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
                return True

        # Tier 3: search code content — class names, function defs, module refs.
        # git grep is fast (uses index) and catches cases where neither commit
        # messages nor file names mention the subject explicitly.
        for word in search_words[:2]:
            try:
                result = subprocess.run(
                    ["git", "grep", "-l", "-i", "--max-count=1", word],
                    cwd=str(repo_path),
                    capture_output=True, text=True, timeout=5,
                )
                if result.returncode == 0 and result.stdout.strip():
                    return True
            except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
                return True

        return False

    def _tag_unverified_claims(
        self,
        entries: list[str],
        repo_path: Path,
    ) -> list[str]:
        """Tag implementation claims that fail git verification with [UNVERIFIED].

        Non-implementation entries pass through unchanged.
        """
        tagged: list[str] = []
        for entry in entries:
            # Extract the claim text (after "- YYYY-MM-DD: " prefix)
            claim_match = re.match(r"^(-\s*\d{4}-\d{2}-\d{2}:\s*)(.*)", entry)
            if not claim_match:
                tagged.append(entry)
                continue

            prefix, claim = claim_match.group(1), claim_match.group(2)
            if self._verify_claim_against_git(claim, repo_path):
                tagged.append(entry)
            else:
                tagged.append(f"{prefix}[UNVERIFIED] {claim}")
                logger.info("Tagged unverified claim: %s", claim[:80])

        return tagged

    def _write_competence(
        self,
        evolution_path: Path,
        competence_entries: list[tuple[str, str]],
    ) -> None:
        """Write competence entries to EVOLUTION.md under 'Competence Learned'.

        Each entry gets a K-prefixed sequential ID. All entries are batched
        into a single locked_write call.
        """
        try:
            content = evolution_path.read_text(encoding="utf-8")
            existing_ids = re.findall(r"### K(\d+)", content)
            next_id = max((int(x) for x in existing_ids), default=0) + 1
        except Exception:
            next_id = 1

        blocks: list[str] = []
        for file_date, entry in competence_entries:
            entry_id = f"K{next_id:03d}"
            blocks.append(
                f"### {entry_id} | {file_date}\n"
                f"- **Competence**: {entry}\n"
                f"- **Status**: active\n"
            )
            next_id += 1

        if blocks:
            self._run_locked_write(
                evolution_path, "Competence Learned", "\n".join(blocks)
            )

        logger.info(
            "Wrote %d competence entries to EVOLUTION.md", len(competence_entries)
        )

    def _write_corrections(
        self,
        evolution_path: Path,
        corrections: list[tuple[str, str]],
    ) -> None:
        """Write correction entries to EVOLUTION.md under 'Corrections Captured'.

        Each correction gets a C-prefixed sequential ID. All entries are
        batched into a single locked_write call.
        """
        # Read current EVOLUTION.md to find next C-ID
        try:
            content = evolution_path.read_text(encoding="utf-8")
            existing_ids = re.findall(r"### C(\d+)", content)
            next_id = max((int(x) for x in existing_ids), default=0) + 1
        except Exception:
            next_id = 1

        blocks: list[str] = []
        for file_date, correction in corrections:
            entry_id = f"C{next_id:03d}"
            blocks.append(
                f"### {entry_id} | {file_date}\n"
                f"- **Correction**: {correction}\n"
                f"- **Status**: active\n"
            )
            next_id += 1

        if blocks:
            self._run_locked_write(
                evolution_path, "Corrections Captured", "\n".join(blocks)
            )

        logger.info(
            "Wrote %d correction entries to EVOLUTION.md", len(corrections)
        )

    @staticmethod
    def _signal_evolution_corrections(ws_path: Path, count: int) -> None:
        """Write signal file for evolution_maintenance_hook.

        When ≥5 corrections are pending, evolution hook will trigger the
        cycle early (without waiting for the 7-day timer). This closes
        the feedback loop: corrections → skill optimization faster.
        """
        signal_path = ws_path / ".evolution_corrections_pending"
        try:
            existing = {}
            if signal_path.exists():
                existing = json.loads(signal_path.read_text(encoding="utf-8"))
            existing["count"] = existing.get("count", 0) + count
            existing["last_updated"] = datetime.now(timezone.utc).isoformat()
            signal_path.write_text(json.dumps(existing), encoding="utf-8")
        except Exception as exc:
            logger.debug("Failed to write evolution signal: %s", exc)

    @staticmethod
    def _check_steeringify_proposals(
        evolution_path: Path,
        ws_path: Path,
    ) -> None:
        """Check if recurring corrections should become STEERING.md rules.

        Uses steeringify v2: structured Pattern-field extraction with
        cross-reference graph (no keyword clustering).

        Writes proposals to a JSON file for session briefing pickup.
        Does NOT auto-write to STEERING.md — user approval required.
        """
        try:
            from skills.s_steeringify.steeringify import (
                extract_corrections,
                group_and_propose,
            )

            evolution_text = evolution_path.read_text(encoding="utf-8")
            entries = extract_corrections(evolution_text)
            if not entries:
                return

            # Load existing STEERING.md for dedup + violation detection
            steering_path = ws_path / ".context" / "STEERING.md"
            steering_text = ""
            if steering_path.exists():
                steering_text = steering_path.read_text(encoding="utf-8")

            proposals = group_and_propose(
                entries,
                min_group_size=2,
                steering_text=steering_text,
            )

            # Only surface proposals that aren't already in STEERING.md
            new_proposals = [p for p in proposals if not p.already_in_steering]
            if not new_proposals:
                return

            # Write proposals JSON for briefing pickup
            proposals_path = ws_path / ".context" / "steeringify_proposals.json"
            data = [
                {
                    "title": p.title,
                    "body": p.body,
                    "source_ids": p.source_ids,
                    "confidence": p.confidence,
                    "already_in_agent": p.already_in_agent,
                    "violates_existing": p.violates_existing,
                }
                for p in new_proposals
            ]
            proposals_path.write_text(
                json.dumps(data, indent=2, ensure_ascii=False),
                encoding="utf-8",
            )
            logger.info(
                "Steeringify: %d rule proposals written to %s",
                len(new_proposals),
                proposals_path,
            )
        except Exception as exc:
            # Steeringify is best-effort — never block distillation
            logger.debug("Steeringify check failed: %s", exc)

    @staticmethod
    def _run_locked_write(
        memory_path: Path,
        section: str,
        text: str,
    ) -> None:
        """Write to MEMORY.md via flock + _modify_content (single lock).

        Deduplicates entries before writing: acquires the lock first, then
        reads existing content and skips entries whose first 120 chars
        already appear.  This prevents double-writes when the distilled
        frontmatter update fails after content extraction succeeds.

        Calls ``_modify_content`` directly under the same flock instead
        of ``locked_read_modify_write`` to avoid a nested-lock deadlock
        (flock is per-open-file-description on POSIX).
        """
        from utils.file_lock import flock_exclusive, flock_unlock
        from scripts.locked_write import _modify_content

        # MemoryGuard: sanitize content before any file I/O
        try:
            from core.memory_guard import MemoryGuard, MemoryGuardError
            _guard = MemoryGuard()
            text = _guard.sanitize(text)
        except MemoryGuardError:
            logger.warning("MemoryGuard rejected distillation write to %s", section)
            return
        except ImportError:
            pass  # memory_guard not available yet

        lock_path = memory_path.with_suffix(memory_path.suffix + ".lock")
        lock_path.parent.mkdir(parents=True, exist_ok=True)
        fd = None
        try:
            fd = open(lock_path, "w")  # noqa: SIM115
            flock_exclusive(fd)

            # Read current content under lock
            if memory_path.exists():
                existing = memory_path.read_text(encoding="utf-8")
            else:
                existing = ""

            # Dedup: filter out entries already present using TWO strategies:
            # (1) 120-char prefix match (existing, catches exact duplicates)
            # (2) Bold-title match (new, catches entries with same topic but
            #     different detail — e.g. [LL06] and [LL10] both titled
            #     "CJK 没有词边界..." are the same lesson with different IDs)
            if existing:
                existing_lines_lower = {
                    ln.strip()[:120].lower()
                    for ln in existing.splitlines()
                    if ln.strip()
                }
                # Extract bold titles from existing: "**Some Title**" → "some title"
                existing_titles = set()
                for ln in existing.splitlines():
                    m = re.search(r"\*\*(.+?)\*\*", ln)
                    if m:
                        existing_titles.add(m.group(1).strip().lower())

                new_lines = []
                for line in text.splitlines():
                    entry_key = line.strip()[:120].lower()
                    # Strategy 1: exact prefix match
                    if entry_key and entry_key in existing_lines_lower:
                        continue
                    # Strategy 2: bold title match (catches reworded duplicates)
                    title_match = re.search(r"\*\*(.+?)\*\*", line)
                    if title_match:
                        title = title_match.group(1).strip().lower()
                        if title and title in existing_titles:
                            logger.debug(
                                "distillation dedup: skipping duplicate title '%s'",
                                title[:60],
                            )
                            continue
                        # Also add to existing_titles for intra-batch dedup
                        existing_titles.add(title)
                    new_lines.append(line)
                if not new_lines:
                    return  # all entries already present
                text = "\n".join(new_lines)

            # Modify + write under the same lock
            new_content = _modify_content(existing, section, text, "prepend")
            memory_path.parent.mkdir(parents=True, exist_ok=True)
            memory_path.write_text(new_content, encoding="utf-8")
        except LockedWriteError as e:
            logger.warning("locked_write failed for section %s: %s", section, e)
        except Exception as e:
            logger.warning("locked_write failed for section %s: %s", section, e)
        finally:
            if fd is not None:
                try:
                    flock_unlock(fd)
                except OSError:
                    pass
                fd.close()

    # Regex for COE lines: **COE:** `signal` — topic
    # Uses regex instead of str.index() to avoid ValueError on malformed lines.
    _COE_LINE_RE = re.compile(
        r"\*\*COE:\*\*\s*`(candidate|resolution)`\s*[—\-]+\s*(.+)",
    )

    @staticmethod
    def _extract_coe_entries(body: str) -> list[tuple[str, str]]:
        """Extract COE signal and topic from DailyActivity body.

        Looks for lines like: ``**COE:** `resolution` — streaming not working``
        Returns list of (signal, topic) tuples.

        Uses regex instead of ``str.index()`` to gracefully handle
        malformed lines (missing backticks, wrong delimiters) without
        raising ``ValueError`` and losing the entire file's COE entries.
        """
        entries = []
        for line in body.splitlines():
            stripped = line.strip()
            m = DistillationTriggerHook._COE_LINE_RE.search(stripped)
            if m:
                signal = m.group(1)
                topic = m.group(2).strip()
                if topic:
                    entries.append((signal, topic))
        return entries

    def _write_coe_registry(
        self,
        memory_path: Path,
        entries: list[tuple[str, str, str]],
    ) -> None:
        """Write COE entries to MEMORY.md under '## COE Registry'.

        Groups by topic. If a topic has both candidate + resolution entries,
        marks it as resolved.
        """
        # Group by topic
        by_topic: dict[str, list[tuple[str, str]]] = {}
        for file_date, signal, topic in entries:
            key = topic.lower().strip()
            if key not in by_topic:
                by_topic[key] = []
            by_topic[key].append((file_date, signal))

        blocks: list[str] = []
        for topic_key, events in by_topic.items():
            # Use the original casing from the first entry
            original_topic = next(
                t for _, _, t in entries if t.lower().strip() == topic_key
            )
            dates = sorted(set(d for d, _ in events))
            has_resolution = any(s == "resolution" for _, s in events)
            status = "✅ Resolved" if has_resolution else "🔍 Investigating"

            blocks.append(
                f"- {dates[0]}: **{original_topic}** — {status}. "
                f"Sessions: {', '.join(dates)}"
            )

        if blocks:
            self._run_locked_write(
                memory_path, "COE Registry", "\n".join(blocks)
            )

        logger.info("Wrote %d COE registry entries to MEMORY.md", len(by_topic))

    def _update_open_threads(
        self,
        memory_path: Path,
        coe_entries: list[tuple[str, str, str]],
    ) -> None:
        """Auto-manage Open Threads from COE signals (code-enforced).

        For each COE topic:
        - If a matching thread exists: increment report count
        - If no match: create new P0 thread (COE candidates auto-promote)
        - If signal is 'resolution': mark thread as resolved

        Uses locked read-modify-write for the entire Open Threads section.
        """
        from utils.file_lock import flock_exclusive, flock_unlock

        lock_path = memory_path.with_suffix(memory_path.suffix + ".lock")
        lock_path.parent.mkdir(parents=True, exist_ok=True)
        fd = None
        try:
            fd = open(lock_path, "w")  # noqa: SIM115  — matches locked_write.py
            flock_exclusive(fd)
            try:
                content = memory_path.read_text(encoding="utf-8")
                updated = self._apply_open_thread_updates(content, coe_entries)
                if updated != content:
                    memory_path.write_text(updated, encoding="utf-8")
                    logger.info("Updated Open Threads with %d COE entries", len(coe_entries))
            finally:
                flock_unlock(fd)
        except Exception as exc:
            logger.warning("Failed to update Open Threads: %s", exc)
        finally:
            if fd is not None:
                fd.close()

    @staticmethod
    def _apply_open_thread_updates(
        content: str,
        coe_entries: list[tuple[str, str, str]],
    ) -> str:
        """Apply COE-driven updates to Open Threads section in MEMORY.md content.

        Matching logic: case-insensitive substring match of COE topic
        against existing thread titles.

        Report count format: ``(reported Nx: session1, session2)``
        """
        lines = content.split("\n")
        # Group COE entries by normalized topic
        by_topic: dict[str, list[tuple[str, str]]] = {}
        for file_date, signal, topic in coe_entries:
            key = topic.lower().strip()
            if key not in by_topic:
                by_topic[key] = []
            by_topic[key].append((file_date, signal))

        matched_topics: set[str] = set()

        for topic_key, events in by_topic.items():
            # Try to find a matching thread line
            found = False
            for i, line in enumerate(lines):
                # Match thread lines: "- 🔴 **title**" or "- 🟡 **title**"
                if not re.match(r"^- [🔴🟡🔵] \*\*", line):
                    continue
                # Extract title between ** **
                title_match = re.search(r"\*\*(.+?)\*\*", line)
                if not title_match:
                    continue
                title = title_match.group(1).lower()
                # Fuzzy match: any significant word overlap
                topic_words = set(topic_key.split())
                title_words = set(title.split())
                overlap = topic_words & title_words
                # Match if >50% of topic words appear in title, or substring match
                if (len(overlap) >= max(1, len(topic_words) // 2)
                        or topic_key in title
                        or title in topic_key):
                    # Found match — increment report count
                    count_match = re.search(r"\(reported (\d+)x:", line)
                    if count_match:
                        old_count = int(count_match.group(1))
                        new_count = old_count + len(events)
                        lines[i] = line.replace(
                            f"reported {old_count}x:",
                            f"reported {new_count}x:",
                        )
                    # Check if resolution — update status line
                    has_resolution = any(s == "resolution" for _, s in events)
                    if has_resolution and i + 1 < len(lines):
                        status_line = lines[i + 1]
                        if "Status:" in status_line and "resolved" not in status_line.lower():
                            lines[i + 1] = f"  Status: ~~{status_line.strip().removeprefix('Status:').strip()}~~ **RESOLVED**."
                    found = True
                    matched_topics.add(topic_key)
                    break

            if not found:
                # Create new P0 thread
                original_topic = next(
                    t for _, _, t in coe_entries if t.lower().strip() == topic_key
                )
                dates = sorted(set(d for d, _ in events))
                new_entry = (
                    f"- 🔴 **{original_topic}** (reported {len(events)}x: {', '.join(dates)})\n"
                    f"  Status: COE candidate — auto-promoted from DailyActivity."
                )
                # Find P0 section to insert
                p0_idx = None
                for i, line in enumerate(lines):
                    if line.strip() == "### P0 — Blocking":
                        p0_idx = i + 1
                        break
                if p0_idx is not None:
                    lines.insert(p0_idx, new_entry)
                else:
                    # No P0 section — find Open Threads and add P0
                    for i, line in enumerate(lines):
                        if line.strip() == "## Open Threads":
                            lines.insert(i + 1, f"\n### P0 — Blocking\n{new_entry}\n")
                            break
                matched_topics.add(topic_key)

        return "\n".join(lines)

    def _reconcile_open_threads_from_deliverables(
        self,
        memory_path: Path,
        da_files: list[Path],
    ) -> None:
        """Auto-resolve Open Threads when deliverables confirm work is done.

        Extracts deliverable lines from the DailyActivity files being
        distilled, then checks each P0/P1/P2 Open Thread title against
        them using fuzzy word overlap.  Matched threads are moved to the
        "Resolved (archive)" section with today's date.

        Uses the same lock pattern as ``_update_open_threads`` to prevent
        concurrent MEMORY.md corruption.
        """
        # Extract deliverables from all DA files being distilled
        deliverables: list[str] = []
        for da_file in da_files:
            try:
                content = da_file.read_text(encoding="utf-8")
            except (OSError, UnicodeDecodeError):
                continue
            in_deliverables = False
            for line in content.splitlines():
                stripped = line.strip()
                if stripped == "**Deliverables:**" or stripped.startswith("### Deliverables"):
                    in_deliverables = True
                    continue
                if in_deliverables and (
                    (stripped.startswith("**") and stripped.endswith(":**"))
                    or stripped.startswith("### ")
                    or stripped.startswith("## ")
                ):
                    in_deliverables = False
                    continue
                if in_deliverables and stripped.startswith("- "):
                    deliverables.append(stripped.lstrip("- ").strip().lower())

        if not deliverables:
            return

        from utils.file_lock import flock_exclusive, flock_unlock

        lock_path = memory_path.with_suffix(memory_path.suffix + ".lock")
        lock_path.parent.mkdir(parents=True, exist_ok=True)
        fd = None
        try:
            fd = open(lock_path, "w")  # noqa: SIM115
            flock_exclusive(fd)
            try:
                content = memory_path.read_text(encoding="utf-8")
                updated = self._apply_deliverable_reconciliation(content, deliverables)
                if updated != content:
                    memory_path.write_text(updated, encoding="utf-8")
                    logger.info("Reconciled Open Threads against %d deliverables", len(deliverables))
            finally:
                flock_unlock(fd)
        except Exception as exc:
            logger.warning("Failed to reconcile Open Threads from deliverables: %s", exc)
        finally:
            if fd is not None:
                fd.close()

    @staticmethod
    def _apply_deliverable_reconciliation(
        content: str,
        deliverables: list[str],
    ) -> str:
        """Move Open Threads to Resolved when deliverables confirm completion.

        Matching: for each thread title, tokenize into words and check
        against each deliverable line.  A thread is considered resolved
        when any deliverable has ≥50% word overlap with the title
        (same heuristic as COE matching) OR when the title is a
        substring of the deliverable (or vice versa).

        Returns the modified content string (unchanged if no matches).
        """
        lines = content.split("\n")
        today = date.today().isoformat().replace("-", "/")  # 2026/03/20
        resolved_entries: list[str] = []
        indices_to_remove: list[int] = []

        # Build deliverable word sets once
        deliv_word_sets = [set(d.split()) for d in deliverables]

        from core.session_utils import fuzzy_title_matches_deliverable

        for i, line in enumerate(lines):
            # Match thread lines: "- 🔴 **title**" or "- 🟡 **title**" or "- 🔵 **title**"
            if not re.match(r"^- [🔴🟡🔵] \*\*", line):
                continue
            title_match = re.search(r"\*\*(.+?)\*\*", line)
            if not title_match:
                continue
            title = title_match.group(1)

            # Check if any deliverable matches this thread
            matched = fuzzy_title_matches_deliverable(
                title, deliverables, deliv_word_sets,
            )

            if matched:
                # Extract short title for resolved archive entry
                resolved_entries.append(f"{title} ({today})")
                indices_to_remove.append(i)

        if not indices_to_remove:
            return content

        # Remove matched thread lines (reverse order to preserve indices)
        for idx in reversed(indices_to_remove):
            # Also remove trailing description/status lines that belong to this thread
            # (indented lines or empty lines immediately after)
            end = idx + 1
            while end < len(lines) and (
                lines[end].startswith("  ") or lines[end].strip() == ""
            ):
                # Don't eat into the next thread or section header
                if end + 1 < len(lines) and re.match(r"^- [🔴🟡🔵] \*\*|^###|^##", lines[end + 1]):
                    break
                end += 1
            # Check if removing this leaves an empty priority section
            # with just "_(None — all clear)_" remaining
            del lines[idx:end]

        # Append to Resolved archive
        resolved_section_idx = None
        for i, line in enumerate(lines):
            if line.strip().startswith("### Resolved"):
                resolved_section_idx = i
                break

        if resolved_section_idx is not None:
            # Find the "- ✅ ..." line in Resolved section and append
            for i in range(resolved_section_idx + 1, min(resolved_section_idx + 5, len(lines))):
                if lines[i].strip().startswith("- ✅"):
                    # Append new resolved items to existing line
                    existing = lines[i].strip()
                    new_items = ", ".join(resolved_entries)
                    lines[i] = f"{existing}, {new_items}"
                    break
            else:
                # No existing resolved line — create one
                lines.insert(
                    resolved_section_idx + 1,
                    f"- ✅ {', '.join(resolved_entries)}",
                )

        # Clean up empty priority sections: if a P section now has only
        # "_(None — all clear)_" or is empty, leave it as-is (harmless).

        logger.info(
            "Auto-resolved %d Open Threads from deliverables: %s",
            len(resolved_entries),
            ", ".join(resolved_entries),
        )
        return "\n".join(lines)

    @staticmethod
    def _archive_old_files(da_dir: Path, ws_path: Path) -> int:
        """Move DailyActivity files older than ARCHIVE_DAYS to Knowledge/Archives/.

        Only moves files that are already distilled (distilled: true in
        frontmatter).  Un-distilled old files are left in place as a
        safety net — they'll be distilled first on the next run.

        Returns the number of files archived.
        """
        cutoff = date.today() - timedelta(days=ARCHIVE_DAYS)
        archive_dir = ws_path / "Knowledge" / "Archives"
        archived = 0
        import shutil

        for f in da_dir.glob("*.md"):
            try:
                file_date = date.fromisoformat(f.stem)
                if file_date >= cutoff:
                    continue
            except ValueError:
                continue

            # Only archive distilled files
            try:
                content = f.read_text(encoding="utf-8")
            except (OSError, UnicodeDecodeError):
                continue
            if not _is_distilled(content):
                continue

            # Move to archive (shutil.move handles cross-device moves)
            archive_dir.mkdir(parents=True, exist_ok=True)
            dest = archive_dir / f.name
            if dest.exists():
                logger.debug("Archive target already exists, skipping: %s", dest)
                continue
            shutil.move(str(f), str(dest))
            archived += 1
            logger.debug("Archived %s → %s", f.name, dest)

        return archived

    # Protected keywords — entries containing these are never archived
    _RC_PROTECTED_KEYWORDS = ("Birthday", "GitHub", "repo", "architecture", "re-architecture")
    _RC_STALE_DAYS = 30

    def _archive_stale_rc_entries(self, memory_path: Path, ws_path: Path) -> None:
        """Archive stale RC entries from MEMORY.md Recent Context section.

        For each entry starting with ``- YYYY-MM-DD:``:
        - If age > 30 days AND key prefix is RC (not KD, LL, COE):
          - Skip if "protected" (contains Birthday, GitHub, repo, architecture, re-architecture)
          - Otherwise: append to Knowledge/Archives/MEMORY-archive-YYYY-MM.md,
            remove from MEMORY.md body

        Uses cross-platform file locking when writing to MEMORY.md.
        """
        from utils.file_lock import flock_exclusive, flock_unlock
        from scripts.locked_write import _find_section_range

        if not memory_path.exists():
            return

        lock_path = memory_path.with_suffix(memory_path.suffix + ".lock")
        lock_path.parent.mkdir(parents=True, exist_ok=True)
        fd = None
        try:
            fd = open(lock_path, "w")
            flock_exclusive(fd)

            content = memory_path.read_text(encoding="utf-8")
            section_range = _find_section_range(content, "Recent Context")
            if section_range is None:
                return

            header_end, next_header_pos = section_range
            section_text = content[header_end:next_header_pos]
            lines = section_text.split("\n")

            today = date.today()
            cutoff = today - timedelta(days=self._RC_STALE_DAYS)
            keep_lines: list[str] = []
            archived_lines: list[str] = []

            for line in lines:
                stripped = line.strip()
                if not stripped.startswith("- "):
                    keep_lines.append(line)
                    continue

                # Extract date
                date_match = re.match(r"^- (\d{4}-\d{2}-\d{2}):", stripped)
                if not date_match:
                    keep_lines.append(line)
                    continue

                try:
                    entry_date = date.fromisoformat(date_match.group(1))
                except ValueError:
                    keep_lines.append(line)
                    continue

                # Only archive RC entries (not KD, LL, COE)
                has_rc_prefix = bool(re.search(r"\[RC\d", stripped))
                is_non_rc = bool(re.search(r"\[(KD|LL|COE)\d", stripped))

                if not has_rc_prefix or is_non_rc:
                    keep_lines.append(line)
                    continue

                # Check age
                if entry_date >= cutoff:
                    keep_lines.append(line)
                    continue

                # Check protected keywords (word-boundary match)
                if any(re.search(r'\b' + re.escape(kw) + r'\b', stripped) for kw in self._RC_PROTECTED_KEYWORDS):
                    keep_lines.append(line)
                    continue

                # Archive this entry
                archived_lines.append(stripped)

            if not archived_lines:
                return

            # Rebuild content
            new_section = "\n".join(keep_lines)
            new_content = content[:header_end] + new_section + content[next_header_pos:]
            memory_path.write_text(new_content, encoding="utf-8")

            # Write to archive file
            archive_dir = ws_path / "Knowledge" / "Archives"
            archive_dir.mkdir(parents=True, exist_ok=True)
            archive_name = f"MEMORY-archive-{today.strftime('%Y-%m')}.md"
            archive_path = archive_dir / archive_name

            archive_block = f"\n### Archived Recent Context ({today.isoformat()})\n"
            archive_block += "\n".join(archived_lines) + "\n"

            if archive_path.exists():
                existing = archive_path.read_text(encoding="utf-8")
                archive_path.write_text(existing + archive_block, encoding="utf-8")
            else:
                archive_path.write_text(
                    f"# Memory Archive -- {today.strftime('%Y-%m')}\n" + archive_block,
                    encoding="utf-8",
                )

            logger.info(
                "Archived %d stale RC entries from MEMORY.md",
                len(archived_lines),
            )
        except Exception as exc:
            logger.warning("RC archival failed: %s", exc)
        finally:
            if fd is not None:
                try:
                    flock_unlock(fd)
                except OSError:
                    pass
                fd.close()

    @staticmethod
    def _enforce_section_caps(memory_path: Path, ws_path: Path | None = None) -> None:
        """Trim MEMORY.md sections to SECTION_CAPS max entries.

        Reads the file, finds each capped section, counts ``- `` prefixed
        lines, and archives the oldest (bottom) entries that exceed the cap
        to ``Knowledge/Archives/MEMORY-archive-YYYY-MM.md``.

        This runs after distillation writes, so the newest entries are at
        the top of each section (prepend mode).  Oldest = bottom = trimmed.

        Args:
            memory_path: Path to MEMORY.md.
            ws_path: Workspace root (for archive directory). If None,
                     overflow entries are discarded instead of archived.
        """
        from utils.file_lock import flock_exclusive, flock_unlock
        from scripts.locked_write import _find_section_range

        if not memory_path.exists():
            return

        # Load usage data for smart eviction (lowest-usage first).
        # Falls back to oldest-first if no usage data exists.
        _KEY_RE = re.compile(r"\[([A-Z]{2,3}\d{2,3})\]")
        usage: dict[str, int] = {}
        if ws_path is not None:
            usage_path = ws_path / ".context" / ".memory-usage.json"
            if usage_path.exists():
                try:
                    usage = json.loads(usage_path.read_text(encoding="utf-8"))
                except (json.JSONDecodeError, OSError):
                    pass

        lock_path = memory_path.with_suffix(memory_path.suffix + ".lock")
        lock_path.parent.mkdir(parents=True, exist_ok=True)
        fd = None
        try:
            fd = open(lock_path, "w")  # noqa: SIM115  — matches locked_write.py
            flock_exclusive(fd)
            try:
                content = memory_path.read_text(encoding="utf-8")
                modified = False
                archived_sections: dict[str, list[str]] = {}

                for section_name, cap in SECTION_CAPS.items():
                    section_range = _find_section_range(content, section_name)
                    if section_range is None:
                        continue

                    header_end, next_header_pos = section_range
                    section_text = content[header_end:next_header_pos]
                    lines = section_text.splitlines()

                    # Count entry lines (start with "- ")
                    entry_indices = [
                        i for i, line in enumerate(lines)
                        if line.strip().startswith("- ")
                    ]

                    if len(entry_indices) <= cap:
                        continue

                    # Smart eviction: sort entries by usage (lowest first),
                    # then by position (bottom = oldest = lower priority).
                    # Entries with 0 usage get evicted before used entries.
                    def _entry_usage(idx: int) -> int:
                        m = _KEY_RE.search(lines[idx])
                        return usage.get(m.group(1), 0) if m else 0

                    if usage:
                        # Sort by usage ascending — lowest usage evicted first
                        eviction_order = sorted(entry_indices, key=_entry_usage)
                    else:
                        # No usage data: fall back to oldest-first (bottom of section)
                        eviction_order = list(reversed(entry_indices))
                    overflow_count = len(entry_indices) - cap
                    evict_set = set(eviction_order[:overflow_count])

                    overflow_indices = evict_set
                    overflow_lines = [
                        lines[i] for i in entry_indices if i in evict_set
                    ]
                    if overflow_lines:
                        archived_sections[section_name] = overflow_lines

                    # Remove oldest entries (bottom of section)
                    trimmed_lines = [
                        line for i, line in enumerate(lines)
                        if i not in overflow_indices
                    ]
                    # Add archive reference
                    today = date.today()
                    archive_ref = f"- [Archived] See Knowledge/Archives/MEMORY-archive-{today.strftime('%Y-%m')}.md"
                    # Check if reference already exists
                    if not any("[Archived]" in l for l in trimmed_lines):
                        trimmed_lines.append(archive_ref)

                    # Preserve original trailing whitespace between sections
                    new_section = "\n".join(trimmed_lines)
                    if section_text.endswith("\n"):
                        new_section += "\n"
                    content = (
                        content[:header_end]
                        + new_section
                        + content[next_header_pos:]
                    )
                    modified = True
                    removed = len(entry_indices) - cap
                    logger.info(
                        "Capped %s: removed %d oldest entries (cap=%d), archived",
                        section_name, removed, cap,
                    )

                if modified:
                    memory_path.write_text(content, encoding="utf-8")

                # Write overflow to archive file
                if archived_sections and ws_path is not None:
                    today = date.today()
                    archive_dir = ws_path / "Knowledge" / "Archives"
                    archive_dir.mkdir(parents=True, exist_ok=True)
                    archive_name = f"MEMORY-archive-{today.strftime('%Y-%m')}.md"
                    archive_path = archive_dir / archive_name

                    # Append to existing archive or create new
                    archive_content = ""
                    if archive_path.exists():
                        archive_content = archive_path.read_text(encoding="utf-8")

                    new_blocks: list[str] = []
                    for sec_name, entries in archived_sections.items():
                        new_blocks.append(f"\n## {sec_name} (archived {today.isoformat()})\n")
                        for entry in entries:
                            new_blocks.append(entry + "\n")

                    archive_content += "".join(new_blocks)
                    archive_path.write_text(archive_content, encoding="utf-8")
                    logger.info(
                        "Archived overflow entries to %s", archive_path,
                    )

            finally:
                flock_unlock(fd)
        except Exception as exc:
            logger.warning("Section cap enforcement failed: %s", exc)
        finally:
            if fd is not None:
                fd.close()

    @staticmethod
    def _write_flag(da_dir: Path, count: int) -> None:
        """Write the .needs_distillation flag file as a fallback."""
        flag_path = da_dir / FLAG_FILENAME
        flag_path.write_text(
            f"undistilled_count={count}\n"
            f"flagged_at={datetime.now().isoformat()}\n",
            encoding="utf-8",
        )


def _is_distilled(content: str) -> bool:
    """Parse YAML frontmatter and check distilled field."""
    if not content.startswith("---"):
        return False
    end = content.find("---", 3)
    if end == -1:
        return False
    frontmatter = content[3:end].strip()
    return "distilled: true" in frontmatter
