KARL Integration — Evolution³ / Stage 1: PATH D

Load registry (mtime-cached)
    --> compile trigger regex per active forged skill
    --> match prompt.search(pattern) for each skill
    --> first match wins
    --> load SKILL.md, strip frontmatter, inject via stdout
    --> write invocation_record to entries.jsonl

def build_skill_embedding_text(skill_name: str, skill_md: str) -> str:
    """Construct embedding input for a skill — captures semantic surface area."""
    sections = []

    # 1. Strip frontmatter, extract intent + workflow
    body = strip_frontmatter(skill_md)
    intent = extract_section(body, "## Intent")
    workflow = extract_section(body, "## Workflow")
    gotchas = extract_section(body, "## Gotchas")

    # 2. Pull top-5 historical trigger prompts from invocation_records
    historical = get_top_prompts_for_skill(skill_name, limit=5)

    sections = [
        f"Skill: {skill_name}",
        f"Intent: {intent[:200]}",
        f"Workflow: {workflow[:300]}",
        f"Gotchas summary: {gotchas[:200]}",
        f"Used for prompts like: {'; '.join(historical)}",
    ]
    return "\n".join(s for s in sections if s)

def build_prompt_embedding_text(prompt: str, cwd: str = "") -> str:
    project = Path(cwd).name if cwd else ""
    if project:
        return f"[project:{project}] {prompt}"
    return prompt

-- In Supabase (reuses existing pgvector schema)
CREATE TABLE skill_embeddings (
    skill_name      TEXT PRIMARY KEY,
    embedding       vector(768),
    embedding_text  TEXT,           -- the text that was embedded
    updated_at      TIMESTAMPTZ DEFAULT NOW(),
    version         INT DEFAULT 1,  -- bump when skill content changes
    trajectory_weight FLOAT DEFAULT 1.0  -- modified by outcome learning
);

CREATE INDEX ON skill_embeddings
    USING ivfflat (embedding vector_cosine_ops)
    WITH (lists = 10);  -- 10 skills = small index, fast

# [home-path]

CACHE_PATH = Path.home() / ".claude" / "cortex" / "skill_embeddings.pkl"
CACHE_TTL_SECONDS = 3600  # 1 hour

def load_skill_embeddings() -> dict[str, np.ndarray]:
    """Load cached skill embeddings. Recompute if stale."""
    if CACHE_PATH.exists():
        age = time.time() - CACHE_PATH.stat().st_mtime
        if age < CACHE_TTL_SECONDS:
            with open(CACHE_PATH, "rb") as f:
                return pickle.load(f)

    # Cache miss — recompute from pgvector
    embeddings = fetch_embeddings_from_pgvector()
    with open(CACHE_PATH, "wb") as f:
        pickle.dump(embeddings, f)
    return embeddings

sim_weighted(prompt, skill) = cosine_sim(embed(prompt), embed(skill)) * w_s

ALPHA = 0.1   # learning rate — slow to prevent noise domination
GAMMA = 0.9   # decay — old signals matter less

def update_weight(current_weight: float, outcome: float, alpha: float = ALPHA) -> float:
    """
    outcome: +1.0 = success, 0.0 = neutral, -1.0 = failure/correction
    Maps outcome to [0.5, 1.5] scaling range.
    """
    target = 1.0 + (outcome * 0.5)  # outcome=+1 -> target=1.5, outcome=-1 -> target=0.5
    return current_weight * (1 - alpha) + target * alpha

def check_post_injection_correction(
    invocation_ts: str,
    session_id: str,
    window_prompts: int = 3,
) -> bool:
    """Returns True if a correction was detected after this skill injection."""
    entries = load_entries(entry_type="correction")
    inv_dt = datetime.fromisoformat(invocation_ts)

    for correction in entries:
        if correction.session_id != session_id:
            continue
        corr_dt = datetime.fromisoformat(correction.ts)
        if corr_dt > inv_dt:
            # Count prompts between injection and correction
            prompts_between = count_prompts_in_window(session_id, inv_dt, corr_dt)
            if prompts_between <= window_prompts:
                return True
    return False

SUCCESS_INDICATORS = [
    r"^commit this",
    r"^push it",
    r"^ship it",
    r"^looks good",
    r"^perfect",
    r"^nice",
    r"^done",
    r"^deploy it",
]

def check_post_injection_success(
    invocation_ts: str,
    session_id: str,
    window_prompts: int = 5,
) -> bool:
    """Returns True if a success-indicator prompt followed this injection."""
    next_prompts = get_prompts_after(session_id, invocation_ts, limit=window_prompts)
    for p in next_prompts:
        for pattern in SUCCESS_INDICATORS:
            if re.match(pattern, p["text"], re.IGNORECASE):
                return True
    return False

def check_bash_exit_codes(
    invocation_ts: str,
    session_id: str,
    skill_name: str,
) -> Optional[float]:
    """Returns 1.0 if relevant build commands succeeded, -1.0 if failed, None if not applicable."""
    SKILL_COMMANDS = {
        "ops:ios": ["xcodebuild", "xcode-select"],
        "ops:deploy": ["systemctl", "docker compose"],
        "ops:docker": ["docker"],
        "ops:prefect": ["prefect"],
        "ops:git": ["git commit", "git push"],
    }
    relevant_cmds = SKILL_COMMANDS.get(skill_name, [])
    if not relevant_cmds:
        return None  # not applicable for this skill

    bash_records = load_bash_records_after(session_id, invocation_ts)
    for record in bash_records:
        cmd = record.get("command", "")
        if any(c in cmd for c in relevant_cmds):
            exit_code = record.get("exit_code", -1)
            return 1.0 if exit_code == 0 else -0.5
    return None

def compute_trajectory_score(
    invocation: CortexEntry,
    all_entries: list[CortexEntry],
) -> float:
    """
    Returns a score in [-1, +1] for this invocation.
    Priority: bash exit codes > correction detection > success indicators > neutral.
    """
    # Try bash exit codes first (highest signal quality)
    exit_score = check_bash_exit_codes(
        invocation.ts, invocation.session_id, invocation.skill
    )
    if exit_score is not None:
        return exit_score

    # Check for corrections (strong negative signal)
    if check_post_injection_correction(invocation.ts, invocation.session_id):
        return -1.0

    # Check for success indicators (moderate positive signal)
    if check_post_injection_success(invocation.ts, invocation.session_id):
        return 0.5   # not +1.0 — success indicators are noisy

    # No signal available — treat as weakly positive (injection didn't cause harm)
    return 0.1

# [home-path]
# Fires on Stop event, budget: 200ms

def update_weights_from_session(session_id: str) -> None:
    """
    For all invocations in this session, compute outcome scores
    and apply EMA weight updates to skill_embeddings.
    """
    invocations = [
        e for e in load_entries(entry_type="invocation_record")
        if e.session_id == session_id
    ]
    if not invocations:
        return

    all_entries = load_entries()
    for inv in invocations:
        score = compute_trajectory_score(inv, all_entries)
        if abs(score) < 0.05:
            continue  # skip neutral, save write

        current_weight = fetch_weight(inv.skill)  # from local pkl cache
        new_weight = update_weight(current_weight, score)
        persist_weight(inv.skill, new_weight)  # update pkl + Supabase async

# flows/feed-hub/skill_embedding_refresh.py
@flow(name="skill-embedding-refresh")
def refresh_skill_embeddings(force_all: bool = False) -> dict:
    """
    Daily: re-embed skills with changed content or significant weight drift.
    """
    registry = load_registry()
    stats = {"reembedded": 0, "skipped": 0}

    for skill_name, info in registry.get("skills", {}).items():
        if info.get("status") != "active":
            continue

        skill_md_path = SKILLS_DIR / skill_name / "SKILL.md"
        if not skill_md_path.exists():
            continue

        if not force_all:
            # Check if recomputation is needed
            current_emb = fetch_embedding_record(skill_name)
            if current_emb:
                skill_mtime = skill_md_path.stat().st_mtime
                emb_mtime = current_emb["updated_at"].timestamp()
                weight_drift = abs(current_emb["trajectory_weight"] - 1.0)
                if skill_mtime < emb_mtime and weight_drift < 0.15:
                    stats["skipped"] += 1
                    continue

        skill_text = build_skill_embedding_text(skill_name, skill_md_path.read_text())
        embedding = embed_via_ragpp(skill_text)  # POST to :8000
        upsert_skill_embedding(skill_name, embedding, skill_text)
        stats["reembedded"] += 1

    return stats

[User prompt arrives]
        |
        v
[ops_trigger_v2.py] -- embed prompt (RAG++ :8000, ~80ms)
        |
        v
[load skill_embeddings.pkl] -- in-memory, <1ms
        |
        v
[compute weighted similarities] -- 10 skills x 768 dims, numpy, <1ms
        |
        v
[inject top-k skill(s)] -- SKILL.md content injection
        |
        v
[session continues... tools called... bash exits... corrections...]
        |
        v
[Stop event fires]
        |
        v
[weight_updater.py] -- score trajectory, EMA update weights, persist
        |
        v
[weights.pkl updated] -- next prompt uses updated space
        |
        v (daily)
[skill_embedding_refresh Prefect flow] -- re-embed with historical prompts

#!/usr/bin/env python3
"""Ops-Trigger V2 — learned vector routing for skill injection.

Replaces regex-based skill matching with trajectory-weighted cosine similarity.
Embedding model: Gemini text-embedding-004 via RAG++ gateway (:8000).
Weights stored in: [home-path]
Full design: Desktop/evo-cube-output/karl-trajectory-intelligence/stage1-path-d.md

Performance budget: <500ms (80ms embed + <5ms retrieval + <50ms load). SIGALRM hard cap.
"""

from __future__ import annotations

import json
import os
import pickle
import signal
import sys
import time
from pathlib import Path
from typing import Optional

import numpy as np

if __name__ == "__main__" or "pytest" not in sys.modules:
    signal.signal(signal.SIGALRM, lambda *_: sys.exit(0))
    signal.setitimer(signal.ITIMER_REAL, 0.5)

SKILLS_DIR = Path.home() / ".claude" / "skills"
CACHE_PATH = Path.home() / ".claude" / "cortex" / "skill_embeddings.pkl"
ENTRIES_FILE = Path.home() / ".claude" / "cortex" / "entries.jsonl"
RAG_EMBED_URL = "http://localhost:8000/api/rag/embed"  # SSH tunnel :8000

# Similarity thresholds
MIN_SIMILARITY = 0.60       # below this: no injection
MULTI_SKILL_THRESHOLD = 0.80  # above this for multiple skills: compose
TRAJECTORY_WEIGHT_FLOOR = 0.3  # safety floor to prevent total suppression


def _embed_prompt(text: str, cwd: str = "") -> Optional[np.ndarray]:
    """Embed a prompt text via RAG++ gateway. Returns None on failure."""
    import urllib.request
    payload = json.dumps({
        "text": f"[project:{Path(cwd).name}] {text}" if cwd else text
    }).encode()
    try:
        req = urllib.request.Request(
            RAG_EMBED_URL,
            data=payload,
            headers={"Content-Type": "application/json"},
            method="POST",
        )
        with urllib.request.urlopen(req, timeout=0.35) as resp:
            data = json.loads(resp.read())
            return np.array(data["embedding"], dtype=np.float32)
    except Exception:
        return None


def _load_embeddings() -> dict[str, dict]:
    """Load skill embeddings from local pickle cache."""
    if not CACHE_PATH.exists():
        return {}
    try:
        with open(CACHE_PATH, "rb") as f:
            return pickle.load(f)
    except Exception:
        return {}


def _cosine_sim(a: np.ndarray, b: np.ndarray) -> float:
    """Cosine similarity between two unit vectors."""
    norm_a = np.linalg.norm(a)
    norm_b = np.linalg.norm(b)
    if norm_a == 0 or norm_b == 0:
        return 0.0
    return float(np.dot(a, b) / (norm_a * norm_b))


def _rank_skills(
    prompt_emb: np.ndarray,
    skill_embeddings: dict[str, dict],
) -> list[tuple[str, float]]:
    """
    Rank skills by trajectory-weighted cosine similarity.
    Returns list of (skill_name, weighted_score) sorted descending.
    """
    ranked = []
    for skill_name, record in skill_embeddings.items():
        emb = record.get("embedding")
        if emb is None:
            continue
        raw_sim = _cosine_sim(prompt_emb, np.array(emb, dtype=np.float32))
        weight = max(record.get("trajectory_weight", 1.0), TRAJECTORY_WEIGHT_FLOOR)
        weighted = raw_sim * weight
        ranked.append((skill_name, weighted))

    return sorted(ranked, key=lambda x: -x[1])


def _check_claims(skill_name: str) -> bool:
    """Check pane domain claims — unchanged from v1."""
    claims_path = Path.home() / ".claude" / "state" / "pane_claims.json"
    if not claims_path.exists():
        return True
    try:
        with open(claims_path) as f:
            claims = json.load(f)
    except Exception:
        return True
    my_tty = os.environ.get("TTY", "")
    from .ops_trigger import _load_registry
    registry = _load_registry()
    domains = registry.get("skills", {}).get(skill_name, {}).get("domains", [])
    domain = domains[0] if domains else None
    if not domain:
        return True
    for claim in claims.get("active", []):
        if claim.get("domain") == domain and claim.get("tty") != my_tty:
            return False
    return True


def _load_skill_content(skill_name: str) -> Optional[str]:
    """Load SKILL.md body — unchanged from v1."""
    skill_path = SKILLS_DIR / skill_name / "SKILL.md"
    if not skill_path.exists():
        return None
    content = skill_path.read_text()
    if content.startswith("---"):
        end = content.find("---", 3)
        if end > 0:
            content = content[end + 3:].strip()
    return content


def _write_invocation(
    skill_name: str,
    prompt: str,
    similarity: float,
    composed_with: Optional[list[str]] = None,
) -> None:
    """Write invocation record with similarity metadata."""
    import uuid
    from datetime import datetime, timezone
    entry = {
        "id": uuid.uuid4().hex[:8],
        "type": "invocation_record",
        "ts": datetime.now(timezone.utc).isoformat(),
        "machine": "mac1",
        "pane": os.environ.get("TTY", ""),
        "skill": skill_name,
        "trigger_prompt": prompt[:100],
        "routing_method": "embedding_v2",
        "similarity_score": round(similarity, 4),
        "composed_with": composed_with or [],
    }
    ENTRIES_FILE.parent.mkdir(parents=True, exist_ok=True)
    with open(ENTRIES_FILE, "a") as f:
        f.write(json.dumps(entry, separators=(",", ":")) + "\n")


def main():
    try:
        hook_input = json.loads(sys.stdin.read())
    except (json.JSONDecodeError, EOFError):
        sys.exit(0)

    prompt = hook_input.get("prompt", "")
    cwd = hook_input.get("cwd", "")
    if not prompt or len(prompt) < 5:
        sys.exit(0)

    # Step 1: Embed the prompt
    prompt_emb = _embed_prompt(prompt, cwd)
    if prompt_emb is None:
        # Embedding failed (timeout/network) — fall back to legacy regex
        # Import and call v1 main() as fallback
        from .ops_trigger import main as legacy_main
        legacy_main()
        return

    # Step 2: Load skill embeddings from cache
    skill_embeddings = _load_embeddings()
    if not skill_embeddings:
        sys.exit(0)  # cold start — no embeddings yet

    # Step 3: Rank by weighted similarity
    ranked = _rank_skills(prompt_emb, skill_embeddings)
    if not ranked or ranked[0][1] < MIN_SIMILARITY:
        sys.exit(0)  # no skill is relevant enough

    # Step 4: Determine injection strategy
    top_skill, top_score = ranked[0]
    if not _check_claims(top_skill):
        sys.exit(0)

    # Check for multi-skill composition
    composed = []
    for skill_name, score in ranked[1:3]:  # consider 2nd and 3rd
        if score >= MULTI_SKILL_THRESHOLD and _check_claims(skill_name):
            composed.append((skill_name, score))

    # Step 5: Inject
    if not composed:
        # Single skill injection
        content = _load_skill_content(top_skill)
        if content:
            print(f"<system-reminder>\n[Cortex v2] Skill: {top_skill} (sim={top_score:.3f})\n{content}\n</system-reminder>")
            _write_invocation(top_skill, prompt, top_score)
    else:
        # Multi-skill composition
        _inject_composite(top_skill, top_score, composed, prompt)

    sys.exit(0)


def _inject_composite(
    primary: str,
    primary_score: float,
    secondary: list[tuple[str, float]],
    prompt: str,
) -> None:
    """Inject a composite skill context for multi-domain prompts."""
    parts = [f"[Cortex v2] Multi-skill: {primary} (sim={primary_score:.3f})"]
    all_skills = [primary]

    primary_content = _load_skill_content(primary)
    if primary_content:
        parts.append(f"## {primary}\n{primary_content}")

    for skill_name, score in secondary:
        content = _load_skill_content(skill_name)
        if content:
            # Extract only the Gotchas section for secondary skills to avoid bloat
            gotchas = _extract_section(content, "## Gotchas")
            if gotchas:
                parts.append(f"## {skill_name} — Gotchas\n{gotchas}")
                all_skills.append(skill_name)

    combined = "\n\n".join(parts)
    print(f"<system-reminder>\n{combined}\n</system-reminder>")

    for skill_name in all_skills:
        score = primary_score if skill_name == primary else dict(secondary)[skill_name]
        _write_invocation(skill_name, prompt, score, composed_with=all_skills)


def _extract_section(content: str, section_header: str) -> str:
    """Extract a markdown section by header."""
    start = content.find(section_header)
    if start == -1:
        return ""
    end = content.find("\n## ", start + len(section_header))
    if end == -1:
        return content[start:]
    return content[start:end]


if __name__ == "__main__":
    main()

"router": {
    "version": "v1",  // change to "v2" when embeddings are bootstrapped
    "fallback": "regex",  // v2 falls back to v1 on embedding failure
    "min_similarity": 0.60
}

# [home-path]
# Run once to initialize skill_embeddings.pkl

def bootstrap_from_invocation_records():
    """
    Phase 1: Embed all 13 active skills using current SKILL.md content
             + historical prompts from prompts-all.jsonl.
    Phase 2: Set initial trajectory weights from domain keyword frequency
             (a cheap proxy for "how often does this domain matter here").
    Phase 3: Persist to pkl + pgvector.
    """
    # Phase 1: Gather top prompts per domain
    prompts = load_all_prompts()
    domain_prompts = group_prompts_by_domain(prompts)  # uses DOMAIN_KEYWORDS from extractor.py

    skill_records = {}
    for skill_name, info in load_active_skills().items():
        skill_md = (SKILLS_DIR / skill_name / "SKILL.md").read_text()
        domain = info.get("domains", ["debug"])[0]
        historical = domain_prompts.get(domain, [])[:5]

        embedding_text = build_skill_embedding_text(skill_name, skill_md, historical)
        embedding = embed_via_ragpp(embedding_text)

        # Phase 2: Initial weight from domain frequency
        # Skills invoked more often start with slight advantage
        domain_count = len(domain_prompts.get(domain, []))
        initial_weight = 1.0 + min(domain_count / 1000, 0.2)  # max 1.2 initial advantage

        skill_records[skill_name] = {
            "embedding": embedding.tolist(),
            "embedding_text": embedding_text,
            "trajectory_weight": initial_weight,
            "version": 1,
            "bootstrapped_at": datetime.now(timezone.utc).isoformat(),
        }

    # Phase 3: Persist
    with open(CACHE_PATH, "wb") as f:
        pickle.dump(skill_records, f)
    upsert_all_to_pgvector(skill_records)

    print(f"Bootstrap complete: {len(skill_records)} skills embedded")
    return skill_records


def group_prompts_by_domain(prompts: list[dict]) -> dict[str, list[str]]:
    """Group prompt texts by the operational domain they touch."""
    from cortex.forge.extractor import DOMAIN_KEYWORDS, _detect_domains
    result = {d: [] for d in DOMAIN_KEYWORDS}
    for p in prompts:
        for domain in _detect_domains(p["text"]):
            if len(result[domain]) < 50:
                result[domain].append(p["text"])
    return result

Primary skill (highest similarity):  FULL SKILL.md content injected
Secondary skills (>0.80 threshold):  GOTCHAS SECTION ONLY injected
Tertiary and beyond:                 SKILL NAME MENTIONED ONLY ("also relevant: ops:deploy")

[Cortex v2] Multi-skill: ops:prefect (sim=0.87) + ops:deploy (sim=0.83) + ops:supabase (sim=0.81)

## ops:prefect
[full SKILL.md content]

## ops:deploy — Gotchas
- SSH heredoc: Mangles ${...} — write scripts locally, scp to VM, then execute
- VM hangs: ssh cloud-vm hangs? → gcloud compute instances reset...
[gotchas only]

## ops:supabase — Gotchas
- Anon key allows writes: RLS policies use auth.role() = 'authenticated' but anon key works for CRUD...
[gotchas only]

{
  "type": "invocation_record",
  "skill": "ops:prefect",
  "trigger_prompt": "deploy the flow",
  "routing_method": "embedding_v2",     // NEW — "regex_v1" | "embedding_v2" | "fallback_v1"
  "similarity_score": 0.847,            // NEW — cosine sim after trajectory weighting
  "composed_with": ["ops:deploy"]       // NEW — other skills in composite injection
}

CREATE TABLE skill_embeddings (
    skill_name         TEXT PRIMARY KEY,
    embedding          vector(768),
    embedding_text     TEXT,
    trajectory_weight  FLOAT NOT NULL DEFAULT 1.0,
    updated_at         TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    version            INT NOT NULL DEFAULT 1,
    invocation_count   INT NOT NULL DEFAULT 0,
    success_count      INT NOT NULL DEFAULT 0,
    failure_count      INT NOT NULL DEFAULT 0
);

ANCHOR_TEXT = "deploy the Prefect flow to cloud-vm"
# Compare embed(ANCHOR_TEXT) monthly. If cos_sim < 0.90 vs. previous: re-embed all.