Shape the piped output for the program that reads it

CHANGELOG.md

@@ -13,6 +13,18 @@ migration in the same line.
 
 ### Changed
 
+- **Piped output is shaped for the program reading it.** `-n` and `read` are what a coding
+  agent sees, and an agent pays per character for what a terminal gets free. Behind the same
+  not-a-terminal test the colour seam already uses: session ids shorten to the shortest prefix
+  that still tells every indexed session apart (git's rule, floored at 12 because Codex writes
+  time-ordered uuidv7 and 8 characters collide), column padding is dropped, and the output ends
+  by naming `agsearch read`. A 20-result search goes from 5748 to 5111 bytes, and further in
+  tokens. `read` accepts any unambiguous id prefix and reports how many sessions an ambiguous
+  one matched. A terminal sees exactly what it saw before.
+- **A piped `read` caps at 12k characters**, keeping the opening turns and as many closing ones
+  as fit, because a session is read to learn what the work was and where it stopped. The
+  elision names what it dropped and how to get it. `--full`, and any terminal, is uncapped.
+
 - **`-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

agsearch

@@ -587,16 +587,44 @@ def _agent_tag(source):
     return "cx" if source == "codex" else "cc"
 
 
-def _match_entry(f, matched, total, texts, keys):
+AGENT_ID_MIN = 12       # git's short-hash rule; see _short_id_len for why 8 is not enough
+
+
+def _short_id_len(sids):
+    """Shortest prefix that still tells every indexed session apart, floored at AGENT_ID_MIN.
+
+    Git's rule, for the same reason: the full id is the one field a reader has to copy, and a
+    36-char uuid is the most expensive thing on the line for the one consumer that pays per
+    character. The floor is not cosmetic. Codex writes uuidv7, whose leading bytes are a
+    timestamp, so sessions recorded near each other share 8-char prefixes: on a 741-session
+    corpus 8 collided 36 times and 12 collided none.
+    """
+    uniq = set(sids)
+    for n in (13, 18, 23, 36):          # uuid group boundaries: a cut mid-group reads as noise
+        if n >= AGENT_ID_MIN and len({s[:n] for s in uniq}) == len(uniq):
+            return n
+    return 36
+
+
+def _match_entry(f, matched, total, texts, keys, idlen=36, pad=True):
     """One result: the session's fields on one line, then the line that matched, indented.
 
     The session id leads because it is the only field another program needs — it is the handle
     `agsearch read <sid>` takes.
+
+    A terminal gets padded columns because eyes track a ragged left edge badly. A pipe gets
+    neither the padding nor the full id: alignment buys an agent nothing and every run of
+    spaces costs it a token.
     """
     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]}")
+    proj = short_proj(f[C_CWD])[:20]
+    if pad:
+        head = (f"{f[C_SID]}  {f[C_DATE]}  {tag:<4} {badge:<4} "
+                f"\033[36m{proj:<20}\033[0m  {f[C_TITLE][:60]}")
+    else:
+        head = (f"{f[C_SID][:idlen]} {f[C_DATE]} {tag} {badge} "
+                f"\033[36m{proj}\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]
@@ -631,15 +659,46 @@ def print_matches(lines, query, limit=20):
     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)
+    pad = _isatty(sys.stdout)
+    idlen = 36 if pad else _short_id_len([f[C_SID] for f in rows])
+    _emit("\n".join(_match_entry(f, m, len(qterms), texts.get(f[C_SID], []), keys, idlen, pad)
                     for m, f in hits[:limit]) + "\n")
     extra = len(hits) - limit
     if extra > 0:
         print(f"... and {extra} more (narrow the query, or run agsearch for the TUI).",
               file=sys.stderr)
+    if not pad:
+        # Whoever is reading this is a program, and the next thing it wants is one of these
+        # sessions. Naming the command here means it does not have to be told anywhere else.
+        print("open one: agsearch read <id>  (the id is the first field above)")
     return 0
 
 
+def resolve_sid(sid):
+    """Full session id for `sid`, which may be any unambiguous prefix of one.
+
+    `-n` prints shortened ids, so `read` has to accept what `-n` printed. Git again: a prefix
+    is a handle until it stops being unique, and then the ambiguity is worth saying out loud
+    rather than guessing at.
+
+    Returns (sid, None) or (None, message).
+    """
+    sid = (sid or "").strip()
+    try:
+        index = json.load(open(INDEX_PATH))
+    except (OSError, json.JSONDecodeError):
+        return sid, None                    # no index to check against; let the caller try
+    if sid in index:
+        return sid, None
+    hits = sorted(k for k in index if k.startswith(sid))
+    if len(hits) == 1:
+        return hits[0], None
+    if not hits:
+        return None, f"no session id starts with {sid!r}. Run a search first: agsearch -n \"...\""
+    listed = ", ".join(hits[:5]) + (" ..." if len(hits) > 5 else "")
+    return None, f"{sid!r} matches {len(hits)} sessions: {listed}. Use more characters."
+
+
 def _session_path(sid):
     if os.path.isdir(PROJECTS_DIR):
         for root, _dirs, files in os.walk(PROJECTS_DIR):
@@ -952,13 +1011,41 @@ def resume_line(sid, cwd):
     return resume_command(target or cwd, argv)
 
 
-def render_transcript(sid, thinking="0", query="", color=None):
+AGENT_READ_CHARS = 12_000       # what a piped `read` spends before it starts eliding
+OPENING_TURNS = 2               # kept whatever the budget: the opening states the goal
+
+
+def _elide(blocks, budget):
+    """Keep the opening turns and as many closing turns as fit, drop the middle.
+
+    A session is read to answer one of two questions: what were we doing, and where did we
+    stop. The first lives in the opening turns and the second in the closing ones, so a
+    transcript too long to print whole loses least from the middle. Returns (blocks, dropped).
+    """
+    if sum(len(b) for b in blocks) <= budget:
+        return blocks, 0
+    head = blocks[:OPENING_TURNS]
+    spent = sum(len(b) for b in head)
+    tail = []
+    for b in reversed(blocks[OPENING_TURNS:]):
+        if spent + len(b) > budget:
+            break
+        tail.insert(0, b)
+        spent += len(b)
+    return head + tail, len(blocks) - len(head) - len(tail)
+
+
+def render_transcript(sid, thinking="0", query="", color=None, budget=None):
     """The whole conversation, readable, without resuming it.
 
     Resuming to read costs a CLI start, a context load, and a session you then
     have to leave. Usually you only wanted to check this is the right session or
     lift one answer out of it. Matched turns are marked with a bar so the pager
     can jump between them.
+
+    `budget` caps the printed characters for a reader that pays for them. A terminal has a
+    scrollback and a pager and so gets the whole thing; a pipe gets the opening and the
+    ending, and a line saying what it did not get and how to ask for it.
     """
     # Full text, not the index cap: this view exists to be read.
     source, tagged = load_session_rows(sid, _flag(thinking), limit=100_000)
@@ -976,13 +1063,23 @@ def render_transcript(sid, thinking="0", query="", color=None):
            + (f" · {n_sub} subagent" if n_sub else "") + fork_line(sid) + "\033[0m",
            f"\033[2m{resume_line(sid, r0[1])}\033[0m\n"]
 
+    blocks = []
     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 "  "
-        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)
+        blocks.append(mark + _turn_header(r[4], source, sub) + "\n"
+                      + f"\033[2m{r[3][11:16]}\033[0m  " + _snippet(r[7], keys, 100_000) + "\n")
+
+    dropped = 0
+    if budget:
+        blocks, dropped = _elide(blocks, budget)
+    if dropped:
+        note = (f"[{dropped} messages elided from the middle. "
+                f"To see them, search inside this session: "
+                f'agsearch -n "<terms>" | grep {sid[:13]}, '
+                f"or read it whole: agsearch read {sid[:13]} --full]")
+        blocks.insert(OPENING_TURNS, "\033[2m" + note + "\033[0m\n")
+    _emit("\n".join(out + blocks) + "\n", color)
     return 0
 
 
@@ -1606,7 +1703,14 @@ def main(argv):
         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:]))
+        rest = [a for a in argv[2:] if a != "--full"]
+        full = "--full" in argv[2:] or _isatty(sys.stdout)
+        sid, err = resolve_sid(argv[1])
+        if err:
+            print(err, file=sys.stderr)
+            return 1
+        return render_transcript(sid, "0", " ".join(rest),
+                                 budget=None if full else AGENT_READ_CHARS)
     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_output.py

@@ -0,0 +1,112 @@
+"""`-n` piped is read by a program, and a program pays per character.
+
+A terminal gets aligned columns and the whole session id. A pipe gets neither: alignment buys
+an agent nothing, every run of spaces costs it a token, and a 36-char uuid is the most
+expensive field on the line for the one reader that has to copy it. These tests pin the
+shortened id, the missing padding, the line that names the next command, and the prefix
+resolution that makes the shortened id usable.
+"""
+
+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()
+
+UUID4 = ["b94c7836-09ee-4716-878a-26a1a407df6f", "fa131384-8585-461b-a951-07a822c00547"]
+# Codex writes uuidv7. The leading bytes are a timestamp, so sessions recorded in the same
+# period agree on their first 8 characters and disagree only later.
+UUID7 = ["019ebbc6-2fbe-7902-abf4-378d68a24645", "019ebbc6-96de-7503-9c1b-b66beb6b88e5"]
+
+
+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):
+    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()
+
+
+def resolve(sids, prefix):
+    with tempfile.TemporaryDirectory() as d:
+        ag.INDEX_PATH = os.path.join(d, "index.json")
+        json.dump({s: {"source": "cc"} for s in sids}, open(ag.INDEX_PATH, "w"))
+        return ag.resolve_sid(prefix)
+
+
+class ShortIdTests(unittest.TestCase):
+    def test_eight_characters_is_not_enough_for_codex_ids(self):
+        """The reason the floor exists. Truncate uuidv7 at 8 and two sessions become one."""
+        self.assertEqual(len({s[:8] for s in UUID7}), 1)
+
+    def test_the_short_id_still_tells_every_session_apart(self):
+        n = ag._short_id_len(UUID4 + UUID7)
+        self.assertEqual(len({s[:n] for s in UUID4 + UUID7}), 4)
+
+    def test_the_short_id_is_never_shorter_than_the_floor(self):
+        self.assertGreaterEqual(ag._short_id_len(["a" * 36]), ag.AGENT_ID_MIN)
+
+    def test_the_cut_lands_on_a_uuid_group_boundary(self):
+        """A prefix cut mid-group reads as noise rather than as an identifier."""
+        self.assertIn(ag._short_id_len(UUID4 + UUID7), (13, 18, 23, 36))
+
+    def test_a_corpus_that_needs_the_whole_id_gets_it(self):
+        near = ["a" * 35 + "0", "a" * 35 + "1"]
+        self.assertEqual(ag._short_id_len(near), 36)
+
+
+class PipedRowTests(unittest.TestCase):
+    def test_the_piped_row_carries_no_column_padding(self):
+        _code, out = run([msg("sid1", "please migrate the billing database")], "billing")
+        self.assertNotIn("  ", out.splitlines()[0])
+
+    def test_a_terminal_still_gets_aligned_columns(self):
+        f = msg("sid1", "please migrate the billing database").split(ag.SEP)
+        head = ag._match_entry(f, 1, 1, [], [], 36, True).splitlines()[0]
+        self.assertIn("  ", head)
+
+    def test_the_piped_id_is_shortened(self):
+        _code, out = run([msg(UUID4[0], "please migrate the billing database")], "billing")
+        self.assertTrue(out.startswith(UUID4[0][:13] + " "), out)
+        self.assertNotIn(UUID4[0], out)
+
+    def test_the_output_names_the_command_that_opens_a_hit(self):
+        """The next thing a program wants is one of these sessions. Say so in the output."""
+        _code, out = run([msg("sid1", "please migrate the billing database")], "billing")
+        self.assertIn("agsearch read", out)
+
+
+class ResolveSidTests(unittest.TestCase):
+    def test_a_whole_id_resolves_to_itself(self):
+        self.assertEqual(resolve(UUID4, UUID4[0]), (UUID4[0], None))
+
+    def test_a_unique_prefix_resolves(self):
+        self.assertEqual(resolve(UUID4, UUID4[0][:13]), (UUID4[0], None))
+
+    def test_an_ambiguous_prefix_says_how_many_and_what_to_do(self):
+        sid, err = resolve(UUID7, UUID7[0][:8])
+        self.assertIsNone(sid)
+        self.assertIn("2 sessions", err)
+        self.assertIn("more characters", err)
+
+    def test_an_unknown_prefix_points_at_search(self):
+        sid, err = resolve(UUID4, "zzzz")
+        self.assertIsNone(sid)
+        self.assertIn("agsearch -n", err)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_noninteractive_rows.py

@@ -45,7 +45,7 @@ def test_a_session_appears_once_however_many_messages_match(self):
     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)
+        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")