Put -n on the ranker the list already uses

CHANGELOG.md

@@ -11,8 +11,26 @@ migration in the same line.
 
 ## [Unreleased]
 
+### Changed
+
+- **`-n` now runs the same ranker as the interactive list.** It was a separate path: an
+  AND of raw substrings over *message* rows, sorted by date. That meant no BM25, no
+  stemming, no typo tier, no demotion of SDK-spawned runs, and one session repeated once
+  per matching message. `-n` is also the README's zero-install first command and the
+  fallback when fzf is missing, so the surface most new users met was the unranked one.
+  It now returns ranked sessions, one entry each. Output shape changed with it: the
+  session id leads the line, and the matching text sits indented below it. Cost of the
+  shared path is ~0.7s per `-n` run against a 400-session corpus, up from ~0.25s,
+  because it builds the same per-session index the list uses.
+
 ### Added
 
+- **`agsearch read <session-id>`** prints a whole conversation without resuming it. This
+  was already there as the TUI's <kbd>Ctrl-O</kbd>, reachable only as an internal
+  subcommand; it is now a documented command, so a search hit can actually be opened.
+- **Colour only when a terminal is reading.** `-n` and `read` emit plain text when stdout
+  is a pipe, or when `NO_COLOR` is set. Escape sequences quoted inside a transcript are
+  stripped too, including ones the snippet cut in half.
 - **Forked sessions are marked `fork`** ahead of the title, in the list and in the preview
   and `read` headers alike, and those headers also name the branch a fork came from and
   the message the two split at. Claude Code forks a

README.md

@@ -73,7 +73,7 @@ pipx install agsearch
 The interactive interface needs [`fzf`](https://github.com/junegunn/fzf#installation) **0.35 or
 newer** — that is the release which added the `start` event agsearch binds. Some distributions
 package an older one; `fzf`'s own install script is the fallback. Without fzf,
-`agsearch -n "query"` still prints ranked matches.
+`agsearch -n "query"` still prints ranked sessions.
 
 ### Install script
 
@@ -91,7 +91,8 @@ agsearch requires Python 3.9 or newer and has no Python package dependencies.
 ```sh
 agsearch                       # browse all sessions in the interactive interface
 agsearch "stripe tax id"       # open with an initial query
-agsearch -n "stripe tax id"    # print ranked matches without fzf
+agsearch -n "stripe tax id"    # print ranked sessions as plain text, no fzf
+agsearch read <session-id>     # print a whole session, without resuming it
 agsearch --here "webhook"      # search only the current project
 agsearch -p myapp "migration"  # search projects whose path contains "myapp"
 agsearch --thinking "query"    # include assistant thinking blocks
@@ -100,6 +101,19 @@ agsearch --reindex             # rebuild the transcript cache
 agsearch --version             # print the installed version
 ```
 
+### Scripts and coding agents
+
+`-n` prints one entry per session as plain text, led by the session id, and drops colour
+whenever it is not writing to a terminal. That makes the search loop scriptable:
+
+```sh
+agsearch -n "webhook retry backoff"        # ranked sessions, one entry each
+agsearch read 3f2a1c4e-...                 # the whole conversation, no resume, no tokens
+```
+
+Ranking is the same as the interactive list, so a term you half-remember or mistype finds
+the same session either way.
+
 ### Interactive keys
 
 | Key | Action |

agsearch

@@ -10,14 +10,19 @@ drops you straight back into the session with `claude --resume` or `codex resume
 Usage:
     agsearch                       # interactive fuzzy TUI (needs fzf)
     agsearch "stripe tax id"       # open the TUI pre-filtered to this query
-    agsearch -n "stripe tax id"    # non-interactive: print ranked matches, no fzf
+    agsearch -n "stripe tax id"    # non-interactive: ranked sessions as plain text, no fzf
+    agsearch read <session-id>     # print a whole session, without resuming it
     agsearch --here "..."          # only sessions from the current directory's project
     agsearch --project myapp       # only sessions whose path matches 'myapp'
     agsearch --thinking            # also index assistant thinking blocks
     agsearch --reindex             # force a full rebuild of the cache
     agsearch --version             # print the installed version and exit
     agsearch _preview <sid> <seq> <thinking> <query...>   # (internal) fzf preview
 
+`-n` prints one entry per session, led by its session id, and drops colour when it is
+not writing to a terminal — so a script or a coding agent can search, then read a hit
+with `agsearch read <session-id>`.
+
 In the TUI the right pane previews the matched session, auto-scrolled to your match
 (marked ▶) with a "match N of M" header. Enter resumes the session (and copies your
 query to the clipboard, so ⌘F → ⌘V → Enter jumps to it inside the replayed transcript).
@@ -423,6 +428,25 @@ class _IndexProgress:
         self._width = 0
 
 
+# Any CSI sequence, not just the colours we emit: transcripts quote terminal output, so a
+# message can carry escapes of its own, and a program reading this wants none of them.
+_ANSI_RE = re.compile(r"\x1b\[[0-9;?]*[a-zA-Z]|\x1b\[?")   # trailing alt: a code the snippet cut in half
+
+
+def _use_color():
+    """Colour is for a human at a terminal. A pipe (an agent, a script, a test) wants text.
+
+    The TUI paths never ask: fzf reads the escapes itself, and its preview and pager are pipes.
+    """
+    return _isatty(sys.stdout) and not os.environ.get("NO_COLOR")
+
+
+def _emit(text, color=None):
+    """Print a rendered block, dropping ANSI unless colour is wanted (auto-detect by default)."""
+    sys.stdout.write(text if (_use_color() if color is None else color)
+                     else _ANSI_RE.sub("", text))
+
+
 def _isatty(stream):
     try:
         return bool(stream.isatty())
@@ -533,14 +557,6 @@ def apply_scope(lines, here=False, project=None):
     return lines
 
 
-def rank_matches(lines, query):
-    """AND-of-terms substring filter, ranked by recency (iso date desc)."""
-    terms = [t.lower() for t in query.split()]
-    hits = [l for l in lines if all(t in l.lower() for t in terms)]
-    hits.sort(key=lambda l: l.split(SEP)[3], reverse=True)
-    return hits
-
-
 # ------------------------------------------------------------------ rendering
 
 def short_proj(cwd):
@@ -559,7 +575,7 @@ def _highlight(text, terms, code="\033[1;30;43m"):
     return text
 
 
-ROW_TEXT_WIDTH = 160        # keeps -n rows one terminal line each, however big the message was
+ROW_TEXT_WIDTH = 160        # the "why it matched" line under a result, however big the message was
 
 
 def _agent_tag(source):
@@ -571,36 +587,56 @@ def _agent_tag(source):
     return "cx" if source == "codex" else "cc"
 
 
-def _source_map():
-    """sid -> source, from the index build_index() writes. Missing/unreadable reads as Claude,
-    which is what the rest of the tool defaults to."""
-    try:
-        index = json.load(open(INDEX_PATH))
-    except (OSError, json.JSONDecodeError):
-        return {}
-    return {sid: info.get("source", "cc") for sid, info in index.items()}
-
-
-def fmt_row(l, terms=(), source="cc"):
-    f = (l.split(SEP) + [""] * 8)[:8]
-    sid, cwd, branch, ts, role, seq, title, text = f
-    tag = {"user": "you", "assistant": _agent_tag(source), "thinking": "th"}.get(role, role)
-    body = _snippet(text, [t for t in terms if t], ROW_TEXT_WIDTH)
-    return f"{ts[:10]}  {short_proj(cwd)[:18]:<18}  {tag:<3} {title[:32]:<32} │ {body}"
-
+def _match_entry(f, matched, total, texts, keys):
+    """One result: the session's fields on one line, then the line that matched, indented.
 
-def print_matches(lines, query, limit=40):
-    hits = rank_matches(lines, query) if query else lines
+    The session id leads because it is the only field another program needs — it is the handle
+    `agsearch read <sid>` takes.
+    """
+    tag = "auto" if f[C_KIND] == "auto" else _agent_tag(f[C_SOURCE])
+    badge = f"{matched}/{total}" if total else ""
+    head = (f"{f[C_SID]}  {f[C_DATE]}  {tag:<4} {badge:<4} "
+            f"\033[36m{short_proj(f[C_CWD])[:20]:<20}\033[0m  {f[C_TITLE][:60]}")
+    idx = best_matching(texts, keys)[0] if (keys and texts) else ()
+    # Matched on the title alone (or no query at all): show what the session opened with.
+    body = texts[idx[0]] if idx else f[C_FIRST]
+    return head + "\n    " + _snippet(body, keys, ROW_TEXT_WIDTH)
+
+
+def print_matches(lines, query, limit=20):
+    """Non-interactive results: the same ranked sessions the TUI lists, one entry each.
+
+    This is the surface a pipe reads — an agent, a script, an install without fzf — so it runs
+    the same rank_sessions the list does instead of a filter of its own. It used to be an
+    AND-of-substrings sorted by date over *message* rows, which meant no BM25, no stemming, no
+    typo tier, no demotion of automation, and one session repeated once per matching message.
+    """
+    rows = build_sessions(lines)
+    if not rows:
+        print("No indexed sessions found.", file=sys.stderr)
+        return 1
+    qterms = parse_query(query) if query.strip() else []
+    if qterms:
+        hits = [(m, f) for _score, m, f in rank_sessions(rows, qterms, _usage_counts())]
+    else:                              # no query: newest first, yours ahead of automation
+        rows.sort(key=lambda f: f[C_DATE], reverse=True)
+        rows.sort(key=lambda f: f[C_KIND] == "auto")
+        hits = [(0, f) for f in rows]
     if not hits:
         print("No matches.", file=sys.stderr)
         return 1
-    terms = [t.lower() for t in query.split()]
-    sources = _source_map()
-    for l in hits[:limit]:
-        print(fmt_row(l, terms, sources.get(l.split(SEP, 1)[0], "cc")))
+
+    keys = query_keys(qterms)
+    texts = {}
+    for l in lines:                    # sid -> its messages, for the "why it matched" line
+        f = l.split(SEP)
+        texts.setdefault(f[0], []).append(f[7])
+    _emit("\n".join(_match_entry(f, m, len(qterms), texts.get(f[C_SID], []), keys)
+                    for m, f in hits[:limit]) + "\n")
     extra = len(hits) - limit
     if extra > 0:
-        print(f"... and {extra} more (narrow the query or use the fzf TUI).", file=sys.stderr)
+        print(f"... and {extra} more (narrow the query, or run agsearch for the TUI).",
+              file=sys.stderr)
     return 0
 
 
@@ -916,7 +952,7 @@ def resume_line(sid, cwd):
     return resume_command(target or cwd, argv)
 
 
-def render_transcript(sid, thinking="0", query=""):
+def render_transcript(sid, thinking="0", query="", color=None):
     """The whole conversation, readable, without resuming it.
 
     Resuming to read costs a CLI start, a context load, and a session you then
@@ -927,25 +963,27 @@ def render_transcript(sid, thinking="0", query=""):
     # Full text, not the index cap: this view exists to be read.
     source, tagged = load_session_rows(sid, _flag(thinking), limit=100_000)
     if not tagged:
-        print("(session not found)")
-        return
+        print("(session not found)")      # stdout: the TUI reads this through a pager
+        return 1
     rows = [r for r, _ in tagged]
     keys = query_keys(parse_query(query)) if query.strip() else []
 
     r0 = rows[0]
     title = r0[6] or next((r[7] for r, _ in tagged if r[4] == "user"), "") or "(untitled)"
     n_sub = sum(1 for _, sub in tagged if sub)
-    print(fork_mark(sid) + f"\033[1m{title[:80]}\033[0m")
-    print(f"\033[2m{short_proj(r0[1])} · {r0[3][:10]} · {len(rows)} msgs"
-          + (f" · {n_sub} subagent" if n_sub else "") + fork_line(sid) + "\033[0m")
-    print(f"\033[2m{resume_line(sid, r0[1])}\033[0m\n")
+    out = [fork_mark(sid) + f"\033[1m{title[:80]}\033[0m",
+           f"\033[2m{short_proj(r0[1])} · {r0[3][:10]} · {len(rows)} msgs"
+           + (f" · {n_sub} subagent" if n_sub else "") + fork_line(sid) + "\033[0m",
+           f"\033[2m{resume_line(sid, r0[1])}\033[0m\n"]
 
     for r, sub in tagged:
         hit = bool(keys) and all(k in r[7].lower() for k in keys)
         mark = "\033[1;33m▶\033[0m " if hit else "  "
-        print(mark + _turn_header(r[4], source, sub))
-        print(f"\033[2m{r[3][11:16]}\033[0m  " + _snippet(r[7], keys, 100_000))
-        print()
+        out.append(mark + _turn_header(r[4], source, sub))
+        out.append(f"\033[2m{r[3][11:16]}\033[0m  " + _snippet(r[7], keys, 100_000))
+        out.append("")
+    _emit("\n".join(out) + "\n", color)
+    return 0
 
 
 def render_preview(sid, thinking, query):
@@ -1186,10 +1224,10 @@ def build_sessions(lines):
         # Stored lowercased: ranking is the only reader, and it would otherwise re-lower the
         # whole corpus on every keystroke. Nothing displays this column.
         blob = _single_line(" · ".join(s["texts"]), 2_000_000).lower()
-        out.append(SEP.join([s["sid"], s["cwd"], s["date"][:10], source, s["kind"],
-                             title, first, blob]))
-    with open(SESSIONS_PATH, "w") as fh:
-        fh.write("\n".join(out))
+        out.append([s["sid"], s["cwd"], s["date"][:10], source, s["kind"], title, first, blob])
+    with open(SESSIONS_PATH, "w") as fh:                 # what _filter reads, one line each
+        fh.write("\n".join(SEP.join(r) for r in out))
+    return out
 
 
 # Common words that add noise, not signal, to a search ("migration OF the database").
@@ -1560,10 +1598,15 @@ def main(argv):
     if argv and argv[0] == "_filter":
         cmd_filter(argv[1:])
         return 0
-    if argv and argv[0] == "_transcript":
+    if argv and argv[0] == "_transcript":    # fzf ctrl-o, piped to a pager: keep the colour
         render_transcript(argv[1] if len(argv) > 1 else "",
-                          argv[2] if len(argv) > 2 else "0", " ".join(argv[3:]))
+                          argv[2] if len(argv) > 2 else "0", " ".join(argv[3:]), color=True)
         return 0
+    if argv and argv[0] == "read":
+        if len(argv) < 2 or not argv[1].strip():
+            print("usage: agsearch read <session-id> [query]", file=sys.stderr)
+            return 1
+        return render_transcript(argv[1], "0", " ".join(argv[2:]))
     if argv and argv[0] == "_copy":
         sid = argv[1] if len(argv) > 1 else ""
         cwd = argv[2] if len(argv) > 2 else ""

tests/test_agent_label.py

@@ -38,27 +38,16 @@ def test_unknown_source_falls_back_to_cc(self):
         self.assertEqual(ag._agent_tag(None), "cc")
 
 
-class FmtRowTests(unittest.TestCase):
-    def test_assistant_row_is_named_after_the_source(self):
-        self.assertIn("cx ", strip(ag.fmt_row(row("assistant", "the reply"), (), "codex")))
-        self.assertIn("cc ", strip(ag.fmt_row(row("assistant", "the reply"), (), "cc")))
-
-    def test_source_only_renames_the_agent_side(self):
-        for source in ("cc", "codex"):
-            self.assertIn("you", strip(ag.fmt_row(row("user", "the ask"), (), source)))
-            self.assertIn("th ", strip(ag.fmt_row(row("thinking", "musing"), (), source)))
-
-    def test_defaults_to_claude_when_no_source_is_passed(self):
-        self.assertIn("cc ", strip(ag.fmt_row(row("assistant", "the reply"))))
-
-
 class LabelTests(unittest.TestCase):
     """The preview names the agent in the turn gutter, so these assert the
     rendered gutter rather than a label helper. The list and `-n` paths keep the
-    two-character `cc`/`cx` form; only the preview spells the agent out."""
+    two-character `cc`/`cx` form; only the preview spells the agent out.
+
+    `-n` lists sessions rather than messages, so it has no per-turn role to name; its `cc`/`cx`
+    comes from the session's source via _agent_tag, covered by AgentTagTests above."""
 
     def _gutter(self, source):
-        # _preview_lines takes split rows, not the SEP-joined strings fmt_row takes.
+        # _preview_lines takes split rows, not the SEP-joined strings row() builds.
         turns = [(row("user", "the ask").split(ag.SEP), False),
                  (row("assistant", "the reply").split(ag.SEP), False)]
         return "\n".join(strip(l) for l in ag._preview_lines(turns, [], source))

tests/test_noninteractive_rows.py

@@ -0,0 +1,66 @@
+"""`-n` is the surface a pipe reads: a script, a coding agent, or an install without fzf.
+
+It used to be its own thing — an AND of raw substrings over *message* rows, sorted by date —
+so it had none of the ranking the TUI has and repeated a session once per matching message.
+These tests pin the contract that replaced it: same ranker as the list, one entry per session,
+led by the session id, and no ANSI unless a terminal is reading.
+"""
+
+import io
+import json
+import os
+import tempfile
+import unittest
+from contextlib import redirect_stdout, redirect_stderr
+
+from load_agsearch import load_agsearch
+
+ag = load_agsearch()
+
+
+def msg(sid, text, role="user", ts="2026-08-19T00:00:00", title="Billing migration"):
+    return ag.SEP.join([sid, "/repo/app", "main", ts, role, "0", title, text, "cli"])
+
+
+def run(lines, query):
+    """print_matches against a throwaway cache, returning what a pipe would have read."""
+    out, err = io.StringIO(), io.StringIO()
+    with tempfile.TemporaryDirectory() as d:
+        ag.CACHE_DIR = d
+        ag.SESSIONS_PATH = os.path.join(d, "sessions.tsv")
+        ag.INDEX_PATH = os.path.join(d, "index.json")
+        json.dump({}, open(ag.INDEX_PATH, "w"))
+        with redirect_stdout(out), redirect_stderr(err):
+            code = ag.print_matches(lines, query)
+    return code, out.getvalue()
+
+
+class NonInteractiveTests(unittest.TestCase):
+    def test_a_session_appears_once_however_many_messages_match(self):
+        lines = [msg("sid1", "please migrate the billing database"),
+                 msg("sid1", "the billing database migration is done", role="assistant")]
+        _code, out = run(lines, "billing")
+        self.assertEqual(out.count("sid1"), 1, out)
+
+    def test_the_entry_leads_with_the_session_id(self):
+        """The id is the handle `agsearch read <sid>` takes. Without it a hit is unopenable."""
+        _code, out = run([msg("sid1", "please migrate the billing database")], "billing")
+        self.assertTrue(out.startswith("sid1  "), out)
+
+    def test_piped_output_carries_no_ansi(self):
+        _code, out = run([msg("sid1", "please migrate the billing database")], "billing")
+        self.assertNotIn("\x1b", out)
+
+    def test_a_typo_still_finds_the_session(self):
+        """Raw substring could not do this, and it is what the README's demo promises."""
+        _code, out = run([msg("sid1", "please migrate the billing database")], "databse")
+        self.assertIn("sid1", out)
+
+    def test_no_match_says_so_and_exits_nonzero(self):
+        code, out = run([msg("sid1", "please migrate the billing database")], "kubernetes")
+        self.assertEqual(code, 1)
+        self.assertEqual(out, "")
+
+
+if __name__ == "__main__":
+    unittest.main()