---
name: ayf:ayf-brain
description: |
  Search the project's accumulated knowledge (decisions, handoffs, agent log,
  learnings) with SQLite FTS5 and return the top ranked entries with file+line
  context. Falls back to grep automatically when sqlite3/FTS5 is unavailable.
allowed-tools:
  - Bash
  - Read
  - Grep
---

# /ayf-brain -- Knowledge Search

Decisions, handoffs, and the agent log pile up in `.ay/` but nothing searches
them, so agents keep re-learning the same lessons. `/ayf-brain <query>` builds a
throwaway SQLite FTS5 index from those files at query time and returns the top 5
relevant entries, ranked by bm25, each with its `file:line` location.

If the Python `sqlite3` module or its FTS5 extension is unavailable, the command
**automatically falls back to a grep-style line scan** over the same files — no
manual step needed.

## Input

The human provides a free-text query:

- `/ayf-brain atomic lock`
- `/ayf-brain handoff schema validation`
- `/ayf-brain "redis connection leak"`

If no query is given, ask: "What do you want to search the project brain for?"

## Sources indexed

| Source       | File                          | Unit indexed                              |
|--------------|-------------------------------|-------------------------------------------|
| `decision`   | `.ay/tracking/DECISIONS.md`   | one section per `#`..`######` heading     |
| `handoff`    | `.ay/tracking/HANDOFFS.md`    | one block per `---` separator or heading  |
| `agents-log` | `.ay/tracking/AGENTS-LOG.md`  | one block per `##` heading or `[timestamp]`|
| `learning`   | `.ay/learnings.jsonl`         | one JSON line (topic + insight + tags)    |

Missing files are skipped silently -- a project with only handoffs still works.
Every indexed unit tracks its starting line number so results point at
`file:line`.

## Run

Pass the human's query words as arguments after `python3 -` (they become the
search terms). The script locates `.ay/` by walking up from the current
directory, so it works from anywhere in the repo.

```bash
python3 - <QUERY WORDS HERE> <<'AYF_BRAIN_PY'
import json
import os
import re
import sys


def find_ay(start="."):
    d = os.path.abspath(start)
    while True:
        if os.path.isdir(os.path.join(d, ".ay")):
            return os.path.join(d, ".ay")
        parent = os.path.dirname(d)
        if parent == d:
            return None
        d = parent


def chunk_file(path, source, boundary_re):
    """Yield (source, path, start_line, ref, body) blocks, tracking line numbers.

    A new block starts at each line matching boundary_re. The ref label is the
    boundary line (cleaned), or the block's first non-empty line when the
    boundary was a bare separator.
    """
    rows = []
    if not os.path.exists(path):
        return rows
    with open(path, encoding="utf-8") as fh:
        lines = fh.readlines()
    blocks = []  # [start_line, header_or_None, [body_lines]]
    cur = None
    for i, raw in enumerate(lines, 1):
        if re.match(boundary_re, raw):
            if cur:
                blocks.append(cur)
            cur = [i, raw.rstrip("\n"), []]
        else:
            if cur is None:
                cur = [i, None, []]
            cur[2].append(raw.rstrip("\n"))
    if cur:
        blocks.append(cur)
    for start_line, header, body_lines in blocks:
        body = "\n".join(body_lines).strip()
        ref = re.sub(r"^[#\s>\-]+", "", header).strip() if header else ""
        if not ref:
            for ln in body_lines:
                if ln.strip():
                    ref = ln.strip()
                    break
        ref = (ref or source)[:80]
        full = ((header or "") + "\n" + body).strip()
        if full:
            rows.append((source, path, start_line, ref, full))
    return rows


def load_learnings(path):
    rows = []
    if not os.path.exists(path):
        return rows
    with open(path, encoding="utf-8") as fh:
        for n, line in enumerate(fh, 1):
            line = line.strip()
            if not line:
                continue
            try:
                obj = json.loads(line)
            except json.JSONDecodeError:
                continue
            topic = str(obj.get("topic", "")).strip()
            ts = str(obj.get("timestamp", "")).strip()
            ref = " / ".join(p for p in (ts, topic) if p) or "learning"
            tags = obj.get("tags", [])
            tags = " ".join(map(str, tags)) if isinstance(tags, list) else str(tags)
            body = " ".join(str(obj.get(k, "")) for k in ("topic", "insight", "source"))
            rows.append(("learning", path, n, ref[:80], (body + " " + tags).strip()))
    return rows


def gather(ay):
    tracking = os.path.join(ay, "tracking")
    rows = []
    rows += chunk_file(os.path.join(tracking, "DECISIONS.md"), "decision", r"^#{1,6}\s")
    rows += chunk_file(os.path.join(tracking, "HANDOFFS.md"), "handoff", r"^(-{3,}\s*$|#{1,6}\s)")
    rows += chunk_file(os.path.join(tracking, "AGENTS-LOG.md"), "agents-log", r"^(#{1,6}\s|\[)")
    rows += load_learnings(os.path.join(ay, "learnings.jsonl"))
    return rows


def build_match(raw):
    terms = re.findall(r"[A-Za-z0-9_]+", raw)
    if not terms:
        return None, []
    return " OR ".join('"%s"' % t for t in terms), terms


def rel(ay, path):
    try:
        return os.path.relpath(path, os.path.dirname(ay))
    except ValueError:
        return path


def grep_fallback(ay, terms, query):
    """Automatic fallback when sqlite3/FTS5 is unavailable: scan files for terms."""
    print(f'🧠 ayf-brain (grep fallback — sqlite3/FTS5 unavailable): "{query}"\n')
    tracking = os.path.join(ay, "tracking")
    files = [os.path.join(tracking, f) for f in
             ("DECISIONS.md", "HANDOFFS.md", "AGENTS-LOG.md")]
    pat = re.compile("|".join(re.escape(t) for t in terms), re.IGNORECASE)
    hits = 0
    for path in files:
        if not os.path.exists(path):
            continue
        with open(path, encoding="utf-8") as fh:
            for n, line in enumerate(fh, 1):
                if pat.search(line):
                    print(f"{rel(ay, path)}:{n}: {line.strip()[:160]}")
                    hits += 1
                    if hits >= 20:
                        print("\n… (truncated at 20 hits — narrow the query)")
                        return 0
    if not hits:
        print(f'no matches for "{query}"')
    return 0


def main(argv):
    query = " ".join(argv[1:]).strip()
    if not query:
        print("usage: ayf-brain <query>", file=sys.stderr)
        return 2

    ay = find_ay()
    if not ay:
        print("No .ay/ directory found. Is the AY Framework installed here?", file=sys.stderr)
        return 1

    match, terms = build_match(query)
    if not match:
        print("Query has no searchable terms.", file=sys.stderr)
        return 2

    rows = gather(ay)
    if not rows:
        print(f"🧠 ayf-brain: no knowledge sources found under {ay}")
        return 0

    # Try SQLite FTS5; fall back to grep automatically if it is not available.
    try:
        import sqlite3
        db = sqlite3.connect(":memory:")
        db.execute(
            "CREATE VIRTUAL TABLE brain USING fts5("
            "source, path UNINDEXED, line UNINDEXED, ref, body, "
            "tokenize='porter unicode61')"
        )
    except Exception:
        return grep_fallback(ay, terms, query)

    db.executemany(
        "INSERT INTO brain(source, path, line, ref, body) VALUES (?,?,?,?,?)", rows
    )
    try:
        results = db.execute(
            "SELECT source, path, line, ref, "
            "snippet(brain, 4, '[', ']', '…', 14) AS snip, bm25(brain) AS score "
            "FROM brain WHERE brain MATCH ? ORDER BY score LIMIT 5",
            (match,),
        ).fetchall()
        total = db.execute(
            "SELECT count(*) FROM brain WHERE brain MATCH ?", (match,)
        ).fetchone()[0]
    except sqlite3.OperationalError:
        return grep_fallback(ay, terms, query)

    if not results:
        print(f'🧠 ayf-brain: no matches for "{query}" across {len(rows)} entries.')
        return 0

    icon = {"learning": "📘", "decision": "⚖️", "handoff": "🤝", "agents-log": "📓"}
    print(f'🧠 ayf-brain: "{query}" — top {len(results)} of {total} matches\n')
    for i, (source, path, line, ref, snip, score) in enumerate(results, 1):
        snip = re.sub(r"\s+", " ", snip).strip()
        print(f"{i}. {icon.get(source, '•')} [{source}] {ref}")
        print(f"   {rel(ay, path)}:{line}")
        print(f"   {snip}\n")
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))
AYF_BRAIN_PY
```

## Output

Present the script's output verbatim -- it is already formatted for the terminal.
Each hit shows its source (`⚖️ decision`, `🤝 handoff`, `📓 agents-log`,
`📘 learning`), a reference label, its `file:line`, and a snippet with matched
terms wrapped in `[brackets]`:

```
🧠 ayf-brain: "atomic lock" — top 2 of 11 matches

1. 📓 [agents-log] Task 60: Atomic lock acquisition + session-id ownership check
   .ay/tracking/AGENTS-LOG.md:35
   …[Atomic] [lock] acquisition + session-id ownership check in pre-commit hook…

2. 🤝 [handoff] [core-platform → all] Task 60 DONE — atomic lock
   .ay/tracking/HANDOFFS.md:172
   …Task 60 DONE — [atomic] [lock] acquisition + session-id ownership…
```

After showing the results, offer a one-line takeaway if the hits answer the
human's underlying question (e.g. "So the canonical lock path is `.ay/tracking/locks/`.").

## Notes

- **Ranking:** bm25 (lower score = more relevant). The index is rebuilt every run,
  so results always reflect the current files -- there is no stale index to maintain.
- **File+line context:** every result prints the source file and the starting line
  of the matched block, so you can jump straight to it.
- **Query matching:** terms are OR'd and stemmed (`porter`), so `signatures` matches
  `signature`. Punctuation and FTS operators in the query are stripped, so an
  arbitrary phrase can never produce a MATCH syntax error.
- **Automatic grep fallback:** if `import sqlite3` fails, or the SQLite build lacks
  FTS5 (the `CREATE VIRTUAL TABLE` raises `OperationalError`), the script switches to
  a case-insensitive line scan over DECISIONS/HANDOFFS/AGENTS-LOG and prints
  `file:line` hits. No dependency beyond the Python standard library either way.
- **Optional gbrain MCP:** if a `gbrain` (or equivalent semantic-search) MCP server is
  connected, prefer it for fuzzy/semantic recall and use `/ayf-brain` as the offline,
  zero-dependency fallback. This command never requires the MCP to be present.
- **No matches:** the script reports how many entries it searched so the human knows
  the brain was consulted and simply had nothing relevant.