fix: clean up ai chat/attach live output

Default pretty output was a firehose: empty assistant/user/result
frames, plumbing system subtypes (init_received, hook_started,
autoinit_loaded, ...), raw session_status JSON, full ISO timestamps,
and a stray ">" caret that landed on the next streamed line.

- Filter noisy message types (session_status, usage_delta) and system
  subtypes (bridge + Claude SDK plumbing) by default.
- Skip assistant frames with only tool_use blocks, user frames that
  wrap a tool_result, and result pings without a summary.
- Shorten timestamps to HH:MM:SS; add --verbose/-v to both
  `ai session attach` and `ai chat` to restore the firehose.
- Fix user-content extraction: server sends {"content": [blocks]},
  not {"text": ...}, so the "user:" line was always blank.
- Strip leading/trailing whitespace and whitespace-only blocks in
  assistant text so Claude's "\n\n..." openers stop rendering as
  empty indented lines.
- Drop the eager "> " input caret — under concurrent streaming it
  interleaved with output; the terminal echoes keystrokes anyway.
- Share the noise set with `ai session history` / `ai chats history`.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: maximelb

limacharlie/commands/_ai_attach.py

@@ -42,13 +42,18 @@ def run_attach(sdk: AISDK, session_id: str, *,
                interactive: bool,
                show_history: bool,
                raw: bool,
+               verbose: bool = False,
                initial_prompt: str | None = None) -> int:
     """Run the attach session loop.
 
     ``initial_prompt`` is sent as the first message after the WebSocket
     connects.  Used by ``ai chat`` to seed a new session; attach leaves
     it ``None``.
 
+    ``verbose`` disables the default noise filter (plumbing system
+    subtypes, empty assistant/user/result frames, session_status pings)
+    and also prints full ISO timestamps instead of HH:MM:SS.
+
     Returns the process exit code: 0 on clean disconnect, 1 on error.
     """
     try:
@@ -58,6 +63,7 @@ def run_attach(sdk: AISDK, session_id: str, *,
             interactive=interactive,
             show_history=show_history,
             raw=raw,
+            verbose=verbose,
             initial_prompt=initial_prompt,
         ))
     except KeyboardInterrupt:
@@ -73,6 +79,7 @@ async def _attach(sdk: AISDK, session_id: str, *,
                   interactive: bool,
                   show_history: bool,
                   raw: bool,
+                  verbose: bool = False,
                   initial_prompt: str | None = None) -> int:
     # Choose the endpoint.  If the user didn't force --read-only and the
     # owner endpoint refuses us with 403, fall back transparently.
@@ -112,6 +119,7 @@ async def _attach(sdk: AISDK, session_id: str, *,
                 interactive=interactive and not read_only,
                 show_history=show_history,
                 raw=raw,
+                verbose=verbose,
             ))
             writer: asyncio.Task | None = None
             if interactive and not read_only:
@@ -140,7 +148,8 @@ async def _attach(sdk: AISDK, session_id: str, *,
 async def _reader_loop(att: SessionAttachment, *,
                        interactive: bool,
                        show_history: bool,
-                       raw: bool) -> None:
+                       raw: bool,
+                       verbose: bool = False) -> None:
     async for msg in att.messages():
         if raw:
             click.echo(json.dumps(msg))
@@ -156,7 +165,9 @@ async def _reader_loop(att: SessionAttachment, *,
                 messages = payload.get("messages") or []
                 _print_banner(f"History ({len(messages)} messages)")
                 for m in messages:
-                    _render_message(m)
+                    if _should_skip_message(m, verbose):
+                        continue
+                    _render_message(m, verbose=verbose)
                 _print_banner("Live stream")
             continue
 
@@ -168,7 +179,8 @@ async def _reader_loop(att: SessionAttachment, *,
             await _handle_question(att, payload)
             continue
 
-        _render_message(msg)
+        if not _should_skip_message(msg, verbose):
+            _render_message(msg, verbose=verbose)
 
         if mtype == MSG_SESSION_END:
             return
@@ -214,14 +226,16 @@ async def _input_loop(att: SessionAttachment) -> None:
 def _read_line_or_eof() -> str | None:
     """Blocking stdin read returning ``None`` on EOF.
 
-    Printed to stderr so it doesn't interleave with stdout messages
-    when the terminal is being driven by another task.
+    We intentionally do NOT print a prompt character before reading.
+    The reader task streams server messages to the terminal
+    concurrently, so any caret we print eagerly ends up on the same
+    line as the next incoming frame (``> [ts] assistant:``) or gets
+    overwritten — the result was inconsistent noise rather than a
+    helpful cue.  The terminal already echoes keystrokes, so users
+    can see what they're typing without a visible prompt; getting a
+    true persistent prompt would need a line-editor library such as
+    ``prompt_toolkit``.
     """
-    try:
-        sys.stderr.write("> ")
-        sys.stderr.flush()
-    except Exception:
-        pass
     line = sys.stdin.readline()
     if not line:
         return None
@@ -309,11 +323,113 @@ async def _handle_question(att: SessionAttachment, payload: dict) -> None:
 # Rendering
 # ---------------------------------------------------------------------------
 
-def _render_message(msg: dict[str, Any]) -> None:
+# System subtypes that are internal plumbing — noisy in the default
+# live stream.  The list matches what the ai-sessions sdk_bridge and
+# the Claude Code SDK emit as "this is what I just configured"
+# housekeeping.  ``ai.py`` also reuses this set for history filtering.
+_NOISY_SYSTEM_SUBTYPES: frozenset[str] = frozenset({
+    # ai-sessions bridge plumbing
+    "credential_diagnostics",
+    "init_received",
+    "claude_md_loaded",
+    "claude_md_error",
+    "user_mcp_servers_loaded",
+    "mcp_config_debug",
+    "mcp_servers_loaded",
+    "mcp_servers_set",
+    "oid_added_to_system_prompt",
+    "ttl_added_to_system_prompt",
+    "unknown_plugin",
+    "plugins_resolved",
+    "autoinit_loaded",
+    "autoinit_error",
+    "resuming_sdk_session",
+    "model_set",
+    "max_turns_set",
+    "max_budget_set",
+    "task_budget_set",
+    "one_shot_mode_set",
+    "permission_mode_set",
+    "tools_configured",
+    "system_prompt_set",
+    "sdk_session_id",
+    "session_patterns_loaded",
+    # Claude SDK passthrough noise
+    "hook_started",
+    "hook_response",
+    "hook_matched",
+})
+
+# Top-level message types treated as plumbing (not user-facing).
+_NOISY_TYPES: frozenset[str] = frozenset({
+    "session_status",
+    "usage_delta",
+    "sdk_session_id",
+})
+
+
+def _should_skip_message(msg: dict[str, Any], verbose: bool) -> bool:
+    """Return True when ``msg`` is plumbing noise we hide by default.
+
+    ``verbose=True`` disables every filter so the user can see the raw
+    firehose.  Skips:
+
+    * Known plumbing message types (session_status, usage_delta, ...).
+    * ``system`` messages whose subtype is in the noisy set.
+    * ``assistant`` frames with no extractable text (tool-use-only
+      turns — the accompanying ``tool_use`` message already renders).
+    * ``user`` frames with no extractable text (tool_result wrappers
+      — the accompanying ``tool_result`` message already renders).
+    * ``result`` frames without a human-readable summary/message.
+    """
+    if verbose:
+        return False
+
     mtype = msg.get("type")
     payload = msg.get("payload") or {}
-    ts = msg.get("timestamp")
-    prefix = f"[{ts}] " if ts else ""
+
+    if mtype in _NOISY_TYPES:
+        return True
+
+    if mtype == MSG_SYSTEM:
+        subtype = payload.get("subtype", "")
+        return subtype in _NOISY_SYSTEM_SUBTYPES
+
+    if mtype == MSG_ASSISTANT:
+        return not _extract_assistant_text(payload).strip()
+
+    if mtype == MSG_USER:
+        return not _extract_user_text(payload).strip()
+
+    if mtype == MSG_RESULT:
+        summary = payload.get("summary") or payload.get("message") or ""
+        return not str(summary).strip()
+
+    return False
+
+
+def _format_timestamp(ts: str | None, verbose: bool) -> str:
+    """Render the incoming ISO timestamp as a prefix.
+
+    Default: ``[HH:MM:SS]`` — enough to eyeball ordering without
+    swamping the line.  ``verbose=True`` preserves the full string so
+    precise correlation with server logs remains possible.
+    """
+    if not ts:
+        return ""
+    if verbose:
+        return f"[{ts}] "
+    # ISO-8601: "YYYY-MM-DDTHH:MM:SS[.frac][+-ZZ:ZZ|Z]".  Grab HH:MM:SS.
+    if "T" in ts:
+        time_part = ts.split("T", 1)[1]
+        return f"[{time_part[:8]}] "
+    return f"[{ts}] "
+
+
+def _render_message(msg: dict[str, Any], *, verbose: bool = False) -> None:
+    mtype = msg.get("type")
+    payload = msg.get("payload") or {}
+    prefix = _format_timestamp(msg.get("timestamp"), verbose)
 
     if mtype == MSG_ASSISTANT:
         text = _extract_assistant_text(payload)
@@ -323,7 +439,7 @@ def _render_message(msg: dict[str, Any]) -> None:
         return
 
     if mtype == MSG_USER:
-        text = payload.get("text", "")
+        text = _extract_user_text(payload)
         click.echo(click.style(f"{prefix}user:", fg="green", bold=True))
         if text:
             click.echo(_indent(text))
@@ -405,19 +521,45 @@ def _render_message(msg: dict[str, Any]) -> None:
 
 
 def _extract_assistant_text(payload: dict) -> str:
+    """Collapse an assistant/user content blocks array into text.
+
+    Returns the concatenation of ``text`` blocks; non-text blocks
+    (tool_use, tool_result, thinking) contribute nothing — a frame
+    that carries only those yields an empty string, which the noise
+    filter uses to drop the redundant "assistant:" header.
+
+    Whitespace-only blocks are dropped and leading/trailing whitespace
+    is stripped from the final result: Claude's first reply of a turn
+    often opens with a couple of ``\\n`` characters that otherwise
+    render as blank indented lines under the ``assistant:`` header.
+    """
     content = payload.get("content")
     if isinstance(content, str):
-        return content
+        return content.strip()
     if isinstance(content, list):
         parts: list[str] = []
         for block in content:
             if isinstance(block, dict):
                 if block.get("type") == "text":
                     parts.append(block.get("text", ""))
-                elif "text" in block:
+                elif "text" in block and block.get("type") not in {"tool_use", "tool_result"}:
                     parts.append(str(block["text"]))
-        return "\n".join(p for p in parts if p)
-    return str(content or "")
+        return "\n".join(p for p in parts if p.strip()).strip()
+    return str(content or "").strip()
+
+
+def _extract_user_text(payload: dict) -> str:
+    """Pull the human-readable text from a ``user`` payload.
+
+    The server wraps prompt turns as ``{"content": [blocks...]}`` (same
+    shape as assistant messages), but older tests and a few code paths
+    still pass ``{"text": "..."}``; accept both so nothing regresses.
+    Tool-result wrappers (no text blocks) collapse to ``""``, which
+    the noise filter uses to drop them.
+    """
+    if "text" in payload and "content" not in payload:
+        return str(payload.get("text") or "")
+    return _extract_assistant_text(payload)
 
 
 # ---------------------------------------------------------------------------

limacharlie/commands/ai.py

@@ -589,31 +589,20 @@ def session_terminate(ctx, session_id) -> None:
 """
 register_explain("ai.session.history", _EXPLAIN_SESSION_HISTORY)
 
-# System subtypes that are internal plumbing, not useful conversation.
-_SYSTEM_INIT_SUBTYPES = frozenset({
-    "credential_diagnostics",
-    "init_received",
-    "claude_md_loaded",
-    "mcp_config_debug",
-    "mcp_servers_set",
-    "model_set",
-    "max_turns_set",
-    "max_budget_set",
-    "oid_added_to_system_prompt",
-    "permission_mode_set",
-    "tools_configured",
-    "system_prompt_set",
-    "sdk_session_id",
-})
+def _filter_history(messages: list[dict]) -> list[dict]:
+    """Remove internal system init messages from history.
 
+    Reuses the single source-of-truth noise set defined alongside the
+    live-stream renderer in ``_ai_attach`` so both surfaces hide the
+    same plumbing subtypes.
+    """
+    from ._ai_attach import _NOISY_SYSTEM_SUBTYPES
 
-def _filter_history(messages: list[dict]) -> list[dict]:
-    """Remove internal system init messages from history."""
     filtered = []
     for m in messages:
         if m.get("type") == "system":
             payload = m.get("payload", {})
-            if isinstance(payload, dict) and payload.get("subtype") in _SYSTEM_INIT_SUBTYPES:
+            if isinstance(payload, dict) and payload.get("subtype") in _NOISY_SYSTEM_SUBTYPES:
                 continue
         filtered.append(m)
     return filtered
@@ -684,8 +673,10 @@ def session_history(ctx, session_id, raw) -> None:
               help="Don't render the initial history block on connect.")
 @click.option("--raw", is_flag=True, default=False,
               help="Print raw JSON messages instead of pretty formatting.")
+@click.option("--verbose", "-v", is_flag=True, default=False,
+              help="Show plumbing system/status messages and full ISO timestamps.")
 @pass_context
-def session_attach(ctx, session_id, read_only, interactive, no_history, raw) -> None:
+def session_attach(ctx, session_id, read_only, interactive, no_history, raw, verbose) -> None:
     from ._ai_attach import run_attach
 
     org = _get_org(ctx)
@@ -696,6 +687,7 @@ def session_attach(ctx, session_id, read_only, interactive, no_history, raw) ->
         interactive=interactive,
         show_history=not no_history,
         raw=raw,
+        verbose=verbose,
     )
     ctx.exit(exit_code)
 
@@ -959,10 +951,12 @@ def claude_auth_logout(ctx) -> None:
               help="Repeatable. Plugin names to enable.")
 @click.option("--idempotent-key", default=None,
               help="Deduplication key for session creation.")
+@click.option("--verbose", "-v", is_flag=True, default=False,
+              help="Show plumbing system/status messages and full ISO timestamps.")
 @pass_context
 def chat(ctx, prompt, name, model, max_turns, max_budget_usd,
          task_budget_tokens, permission_mode,
-         allowed_tools, denied_tools, plugins, idempotent_key) -> None:
+         allowed_tools, denied_tools, plugins, idempotent_key, verbose) -> None:
     from ._ai_attach import run_attach
 
     org = _get_org(ctx)
@@ -1020,6 +1014,7 @@ def chat(ctx, prompt, name, model, max_turns, max_budget_usd,
         interactive=True,
         show_history=False,
         raw=False,
+        verbose=verbose,
         initial_prompt=initial_prompt,
     )
     ctx.exit(exit_code)