clanker-saddle/test/test_task_log_js.py

#!/usr/bin/env python3
"""Static + logic tests for the task event-log page's client JS.

The app binary cannot be built in every container (statum-mkpstm /
libstatum are build-time dependencies), so this suite verifies the
page (src/pages/task_log.html) and its client JS resource
(src/pages/task_log.js, loaded via pstm-res-src) two ways:

1. Statically - the script must be ASCII-only (the page compiler
   HTML-escapes inline script bodies, so any non-ASCII literal is
   re-served as literal entity text like "·" after the
   browser's one decode pass; the resource keeps the discipline),
   must wire the SSE stream through `onmessage` (the server sends
   id:/data: frames with no `event:` name, so named listeners can
   never fire), must not listen for an 'append' event name, and its
   delimiter structure must balance after strings and comments are
   stripped.

2. Behaviourally - the ingest / render-order / scroll-stick /
   load-more logic is ported verbatim below as pure python functions
   and unit-tested: oldest-first chronology, id dedupe, the 48px
   bottom-stick rule, prepend-mode scroll anchoring, and the
   newest-first-page reversal with the `more` -> exhausted flag.

Runnable standalone: python3 test/test_task_log_js.py
"""

import json
import os
import re
import unittest

PAGE = os.path.join(os.path.dirname(os.path.abspath(__file__)),
                    os.pardir, "src", "pages", "task_log.html")
SCRIPT = os.path.join(os.path.dirname(os.path.abspath(__file__)),
                      os.pardir, "src", "pages", "task_log.js")
# Goal 41 T7: the interactive shell (toolbar + feed) lives in the
# task-log-feed partial - shared by this page and the task page's
# Log tab - so markup invariants are asserted against it.
SHELL = os.path.join(os.path.dirname(os.path.abspath(__file__)),
                     os.pardir, "src", "pages", "task_log_feed.html")


def script_body():
    # The client JS lives in its own resource file; the page markup
    # carries no inline script.
    with open(SCRIPT, encoding="utf-8") as handle:
        return handle.read()


# ---------------------------------------------------------------------------
# Static checks
# ---------------------------------------------------------------------------

class ScriptSourceTests(unittest.TestCase):
    """What the served bytes must look like after compilation."""

    def setUp(self):
        self.body = script_body()

    def test_script_is_ascii_only(self):
        """Non-ASCII in a script body comes back as literal entity
        text ("·") because the compiler escapes script bodies
        and the browser decodes exactly once. ASCII (with \\uXXXX
        escapes for the characters we want) is immune."""
        offenders = [(index, hex(ord(char)))
                     for index, char in enumerate(self.body) if ord(char) > 127]
        self.assertEqual(offenders, [])

    def test_page_loads_the_resource_and_carries_no_inline_script(self):
        with open(PAGE, encoding="utf-8") as handle:
            markup = handle.read()
        self.assertNotIn("<script>", markup,
                         "the client JS lives in task_log.js, not inline")
        self.assertIn('pstm-res-src="task_log.js"', markup)

    def test_unicode_escapes_used_for_separators(self):
        """The separator characters must still render - as escapes,
        not as raw bytes and not as HTML entities."""
        for escape in ("\\u00b7", "\\u2026", "\\u2014"):
            self.assertIn(escape, self.body)
        self.assertNotIn("&middot;", self.body)
        self.assertNotIn("&mdash;", self.body)

    def test_stream_wired_through_onmessage(self):
        self.assertRegex(self.body, r"stream\.onmessage\s*=")
        self.assertNotIn("addEventListener('append'", self.body)
        self.assertNotIn('addEventListener("append"', self.body)

    def test_heartbeat_ping_is_skipped(self):
        self.assertRegex(self.body, r"message\.data\s*===\s*'ping'")

    def test_load_more_pages_before_oldest(self):
        self.assertRegex(self.body, r"/events\?before='\s*\+\s*\(state\.oldest")

    def test_page_is_reversed_before_ingest(self):
        self.assertIn("slice().reverse().forEach(ingest)", self.body)

    def test_exhausted_set_when_no_more(self):
        self.assertIn("state.exhausted = true", self.body)

    def test_more_button_rides_inside_the_feed(self):
        # The load-older control is the feed's FIRST row - it
        # scrolls away with the history instead of sitting above
        # the pane. Search the shell markup only (the script's
        # comments also mention [data-log-feed]).
        # Goal 41 T7: the shell moved into the task-log-feed
        # partial (task_log_feed.html) - the SAME markup the task
        # page's Log tab embeds - so the invariant is asserted
        # there.
        with open(SHELL, encoding="utf-8") as handle:
            markup = handle.read().split("</script>", 1)[-1]
        feed = markup.index("data-log-feed")
        more = markup.index("data-log-more", feed)
        close = markup.index("</div>", feed)
        self.assertLess(feed, more)
        self.assertLess(more, close,
                        "the load-older control must live inside the feed container")
        self.assertLess(more, markup.index("<p", feed),
                        "the load-older control must be the feed's first row")

    def test_delimiters_balance(self):
        """A cheap structural stand-in for a JS parser (none is
        installed here): after stripping comments and string literals
        every bracket must close."""
        body = self.body
        out = []
        i = 0
        mode = None  # None | 'line' | 'block' | "'" | '"'
        while i < len(body):
            char = body[i]
            nxt = body[i + 1] if i + 1 < len(body) else ""
            if mode is None:
                if char == "/" and nxt == "/":
                    mode = "line"
                    i += 2
                    continue
                if char == "/" and nxt == "*":
                    mode = "block"
                    i += 2
                    continue
                if char in "'\"":
                    mode = char
                    i += 1
                    continue
                out.append(char)
            elif mode == "line":
                if char == "\n":
                    mode = None
                    out.append("\n")
            elif mode == "block":
                if char == "*" and nxt == "/":
                    mode = None
                    i += 2
                    continue
            else:  # inside a string literal
                if char == "\\":
                    i += 2
                    continue
                if char == mode:
                    mode = None
            i += 1
        self.assertEqual(mode, None, "unterminated comment or string literal")
        pairs = {")": "(", "]": "[", "}": "{"}
        stack = []
        for char in out:
            if char in "([{":
                stack.append(char)
            elif char in pairs:
                self.assertTrue(stack and stack[-1] == pairs[char],
                                "unbalanced %r at depth %s" % (char, stack))
                stack.pop()
        self.assertEqual(stack, [], "unclosed delimiters %s" % stack)


# ---------------------------------------------------------------------------
# Port of the client logic (keep in step with task_log.html)
# ---------------------------------------------------------------------------

STICK_THRESHOLD = 48  # px that still counts as "reading the tail"
PAGE_SIZE = 30


class Feed(object):
    """Scroll-state stand-in for [data-log-feed]."""

    def __init__(self, scroll_height=0, client_height=600, scroll_top=0):
        self.scroll_height = scroll_height
        self.client_height = client_height
        self.scroll_top = scroll_top


def at_bottom(feed):
    """PORT: a feed shorter than its viewport counts as at-bottom."""
    return feed.scroll_height - feed.scroll_top - feed.client_height \
        <= STICK_THRESHOLD


class State(object):
    """PORT: the module-level `state` object."""

    def __init__(self):
        self.events = []
        self.oldest = 0
        self.exhausted = False
        self.loading = False
        self.seeded = False


def ingest(state, evt):
    """PORT: keep state.events oldest-first (ascending id); skip ids
    already held (SSE replay after a reconnect resends a frame)."""
    i = 0
    while i < len(state.events) and state.events[i]["id"] < evt["id"]:
        i += 1
    if i < len(state.events) and state.events[i]["id"] == evt["id"]:
        return
    state.events.insert(i, evt)
    if evt["id"] < state.oldest or state.oldest == 0:
        state.oldest = evt["id"]


def render(state, feed, mode=None, height_after=None):
    """PORT of render(mode): decide the post-render scroll position.
    height_after stands in for the DOM rebuild's new scrollHeight."""
    was_at_bottom = at_bottom(feed)
    height_before = feed.scroll_height
    scroll_top_before = feed.scroll_top
    if height_after is not None:
        feed.scroll_height = height_after
    if not state.seeded and len(state.events) > 0:
        state.seeded = True
        feed.scroll_top = feed.scroll_height  # scrollToBottom()
    elif mode == "prepend":
        if was_at_bottom:
            feed.scroll_top = feed.scroll_height
        else:
            feed.scroll_top = scroll_top_before + (feed.scroll_height
                                                   - height_before)
    elif was_at_bottom:
        feed.scroll_top = feed.scroll_height
    else:
        feed.scroll_top = scroll_top_before
    return feed.scroll_top


def server_page(newest, oldest, more):
    """The endpoint's wire shape: newest-first within the page."""
    return {"events": [{"id": i} for i in range(newest, oldest - 1, -1)],
            "more": more}


def load_more(state, feed, page, height_after):
    """PORT of loadMore's success path: reverse the newest-first page
    into chronological order, ingest it, set exhausted, render."""
    rows = page["events"]
    for evt in reversed(rows):
        ingest(state, evt)
    if not page["more"] or len(rows) == 0:
        state.exhausted = True
    return render(state, feed, mode="prepend", height_after=height_after)


def on_stream_message(state, feed, data, height_after):
    """PORT of onStreamMessage: skip heartbeats, ingest one row."""
    if data == "ping":
        return feed.scroll_top
    ingest(state, {"id": int(data)})
    return render(state, feed, mode="append", height_after=height_after)


# ---------------------------------------------------------------------------
# Behaviour tests
# ---------------------------------------------------------------------------

def ids(state):
    return [evt["id"] for evt in state.events]


class ChronologyTests(unittest.TestCase):

    def test_initial_window_renders_oldest_first(self):
        """The SSE replay / first JSON page arrives newest-first
        ([30..1]); the feed must render [1..30] - oldest at the top."""
        state = State()
        for evt in server_page(30, 1, True)["events"]:
            ingest(state, evt)
        self.assertEqual(ids(state), list(range(1, 31)))
        self.assertEqual(state.oldest, 1)

    def test_live_appends_land_in_order(self):
        state = State()
        for i in range(30, 0, -1):
            ingest(state, {"id": i})
        ingest(state, {"id": 31})
        ingest(state, {"id": 33})
        ingest(state, {"id": 32})  # out of order arrival
        self.assertEqual(ids(state)[-3:], [31, 32, 33])

    def test_duplicate_ids_are_dropped(self):
        state = State()
        ingest(state, {"id": 5})
        ingest(state, {"id": 5})  # EventSource reconnect replay
        self.assertEqual(ids(state), [5])


class ScrollStickTests(unittest.TestCase):

    def test_first_render_with_rows_seeds_at_the_bottom(self):
        state = State()
        feed = Feed(scroll_height=0, client_height=600)
        for evt in server_page(30, 1, True)["events"]:
            ingest(state, evt)
        render(state, feed, height_after=3000)
        self.assertTrue(state.seeded)
        self.assertEqual(feed.scroll_top, 3000)

    def test_append_sticks_when_reading_the_tail(self):
        state = State()
        feed = Feed(scroll_height=3000, client_height=600)
        state.seeded = True
        feed.scroll_top = 2400  # exactly at the bottom
        self.assertTrue(at_bottom(feed))
        ingest(state, {"id": 31})
        render(state, feed, mode="append", height_after=3100)
        self.assertEqual(feed.scroll_top, 3100)

    def test_append_within_threshold_still_sticks(self):
        feed = Feed(scroll_height=3000, client_height=600, scroll_top=2360)
        self.assertTrue(at_bottom(feed))  # 40px shy: close enough

    def test_just_past_threshold_is_not_the_tail(self):
        feed = Feed(scroll_height=3000, client_height=600, scroll_top=2351)
        self.assertFalse(at_bottom(feed))  # 49px shy: reading history

    def test_scrolled_up_reader_is_never_yanked(self):
        state = State()
        feed = Feed(scroll_height=3000, client_height=600, scroll_top=500)
        state.seeded = True
        self.assertFalse(at_bottom(feed))
        ingest(state, {"id": 31})
        render(state, feed, mode="append", height_after=3100)
        self.assertEqual(feed.scroll_top, 500)

    def test_short_feed_counts_as_at_bottom(self):
        """A feed shorter than its viewport cannot be scrolled away
        from the newest row."""
        feed = Feed(scroll_height=100, client_height=600, scroll_top=0)
        self.assertTrue(at_bottom(feed))

    def test_heartbeat_does_not_touch_anything(self):
        state = State()
        feed = Feed(scroll_height=3000, client_height=600, scroll_top=2400)
        state.seeded = True
        before = (list(ids(state)), feed.scroll_top)
        on_stream_message(state, feed, "ping", height_after=3100)
        self.assertEqual((list(ids(state)), feed.scroll_top), before)


class LoadMoreTests(unittest.TestCase):

    def test_older_page_is_reversed_and_prepended(self):
        """before=<oldest>=1 returns [60..31] newest-first on the
        wire; the display must gain [31..60] above [1..30]."""
        state = State()
        feed = Feed(scroll_height=3000, client_height=600, scroll_top=500)
        for evt in server_page(30, 1, True)["events"]:
            ingest(state, evt)
        state.seeded = True
        render(state, feed, height_after=3000)
        feed.scroll_top = 500  # the reader scrolled up into history
        load_more(state, feed, server_page(60, 31, True), height_after=6000)
        self.assertEqual(ids(state), list(range(1, 61)))
        self.assertEqual(state.oldest, 1)
        self.assertFalse(state.exhausted)

    def test_prepend_keeps_the_viewport_anchored(self):
        """Content added above must not move the rows under the
        reader's cursor: scrollTop grows by the added height."""
        state = State()
        feed = Feed(scroll_height=3000, client_height=600, scroll_top=500)
        state.seeded = True
        self.assertFalse(at_bottom(feed))
        delta = 3000
        load_more(state, feed, server_page(60, 31, True),
                  height_after=3000 + delta)
        self.assertEqual(feed.scroll_top, 500 + delta)

    def test_prepend_while_at_bottom_stays_at_bottom(self):
        state = State()
        feed = Feed(scroll_height=3000, client_height=600, scroll_top=2400)
        state.seeded = True
        self.assertTrue(at_bottom(feed))
        load_more(state, feed, server_page(60, 31, True), height_after=6000)
        self.assertEqual(feed.scroll_top, 6000)

    def test_exhausted_when_more_is_false(self):
        state = State()
        feed = Feed(scroll_height=3000, client_height=600, scroll_top=2400)
        state.seeded = True
        load_more(state, feed, server_page(60, 31, False), height_after=6000)
        self.assertTrue(state.exhausted)

    def test_empty_page_exhausts_too(self):
        """A defensive guard: a page with no rows means no older
        history, whatever the flag says."""
        state = State()
        feed = Feed(scroll_height=3000, client_height=600, scroll_top=2400)
        state.seeded = True
        ingest(state, {"id": 1})
        load_more(state, feed, {"events": [], "more": True}, height_after=3000)
        self.assertTrue(state.exhausted)


class EndToEndSequenceTests(unittest.TestCase):
    """The whole first-minute story: replay, live tail, history."""

    def test_replay_then_append_then_history(self):
        state = State()
        feed = Feed(scroll_height=0, client_height=600)
        # 1. SSE replay: most-recent-30 oldest-first on the wire.
        for evt in server_page(30, 1, True)["events"]:
            ingest(state, evt)
        render(state, feed, height_after=3000)
        self.assertEqual(ids(state), list(range(1, 31)))
        self.assertEqual(feed.scroll_top, 3000)  # seeded at the tail

        # 2. Two live rows arrive; the reader is at the bottom.
        on_stream_message(state, feed, "31", height_after=3100)
        on_stream_message(state, feed, "32", height_after=3200)
        self.assertEqual(ids(state)[-2:], [31, 32])
        self.assertEqual(feed.scroll_top, 3200)

        # 3. A heartbeat is ignored, then the reader scrolls up.
        on_stream_message(state, feed, "ping", height_after=3200)
        feed.scroll_top = 100
        on_stream_message(state, feed, "33", height_after=3300)
        self.assertEqual(feed.scroll_top, 100)  # never yanked

        # 4. Load older: before=1 -> [60..31] newest-first wire page.
        self.assertEqual(state.oldest, 1)
        load_more(state, feed, server_page(60, 31, False), height_after=6300)
        self.assertEqual(ids(state), list(range(1, 61)))
        self.assertTrue(state.exhausted)
        self.assertEqual(feed.scroll_top, 100 + 3000)  # anchored


# ---------------------------------------------------------------------------
# Payload contract: the daemon JSON-encodes args/result into STRINGS
# ---------------------------------------------------------------------------

def parse_maybe_json(value):
    """PORT of parseMaybeJson: strings decode, objects pass through,
    a non-JSON string answers None so the caller can fall back to
    rendering the raw text."""
    if not isinstance(value, str):
        return value
    try:
        return json.loads(value)
    except ValueError:
        return None


def is_plain_object(value):
    """PORT of isPlainObject."""
    return isinstance(value, dict)


def fmt_bytes(n):
    """PORT of fmtBytes."""
    if n < 1024:
        return "%d B" % n
    if n < 1024 * 1024:
        return "%.1f KiB" % (n / 1024)
    return "%.1f MiB" % (n / (1024 * 1024))


def str_text(value):
    """PORT of strText."""
    return value if isinstance(value, str) else ""


def filter_of(kind):
    """PORT of filterOf: which filter bucket owns a kind. The yeehaw
    celebration lands in the comments bucket (goal 9 task 4b)."""
    if kind == "tool_call":
        return "commands"
    if kind in ("tool_result", "command_output"):
        return "results"
    if kind in ("comment", "answer", "verdict", "yeehaw"):
        return "comments"
    if kind == "usage":
        return "usage"
    return "transitions"


def yeehaw_render(evt):
    """PORT of eventNode's yeehaw branch -> (span_class, message)."""
    payload = evt.get("payload") or {}
    return "yeehaw-message", str_text(payload.get("message"))


def fmt_duration(ms):
    """PORT of fmtDuration."""
    if ms < 1000:
        return "%d ms" % ms
    return "%.1f s" % (ms / 1000)


def str_bytes(value):
    """PORT of strBytes."""
    return len(value) if isinstance(value, str) else 0


def result_envelope(evt):
    """PORT of resultEnvelope: the parsed result string when it
    decodes to an object, else the object-shaped payload itself."""
    payload = evt.get("payload") or {}
    parsed = parse_maybe_json(payload.get("result"))
    if is_plain_object(parsed):
        return parsed
    if is_plain_object(payload):
        return payload
    return {}


RESULT_PLUMBING = ("ok", "exit_code", "stdout", "stderr", "bytes_out",
                   "bytes_err", "truncated", "duration_ms", "turn", "tool")


def result_extras(envelope):
    """PORT of resultExtras: the envelope minus command plumbing
    and minus the note members (rendered separately)."""
    return dict((key, value) for key, value in envelope.items()
                if key not in RESULT_PLUMBING
                and key not in ("error", "note", "timeout"))


def result_bits(evt):
    """PORT of resultBits -> the metric bits shared by the expander
    summary and the inline line."""
    envelope = result_envelope(evt)
    payload = evt.get("payload") or {}
    bits = []
    if envelope.get("exit_code") is not None:
        bits.append("exit %s" % envelope["exit_code"])
    elif envelope.get("ok") is False:
        bits.append("failed")
    if isinstance(envelope.get("bytes_out"), int) \
            and isinstance(envelope.get("bytes_err"), int):
        bytes_count = envelope["bytes_out"] + envelope["bytes_err"]
    else:
        bytes_count = str_bytes(envelope.get("stdout")) \
            + str_bytes(envelope.get("stderr"))
    if bytes_count > 0:
        bits.append(fmt_bytes(bytes_count))
    if envelope.get("truncated"):
        bits.append("truncated")
    if isinstance(envelope.get("duration_ms"), int):
        bits.append(fmt_duration(envelope["duration_ms"]))
    if payload.get("turn") is not None:
        bits.append("turn %s" % payload["turn"])
    if not bits and isinstance(payload.get("result"), str) \
            and not is_plain_object(parse_maybe_json(payload["result"])):
        bits.append(fmt_bytes(len(payload["result"])))
    return bits


INLINE_RESULT_LIMIT = 220


def result_view(evt, attached=False):
    """PORT of resultView -> (inline, bits_text, body_text): a short
    body renders inline under its metric bits; a long one collapses
    behind the expander summary - UNLESS it rides attached to its
    call, where the output is always visible (CSS clamps oversized
    bodies)."""
    out, err, note, extras, no_output = result_output(evt)
    parts = [part for part in (out, err, note, extras) if part]
    if no_output:
        parts = ["(no output)"]
    body_text = "\n".join(parts).strip()
    inline = attached or len(body_text) <= INLINE_RESULT_LIMIT
    return inline, " \u00b7 ".join(result_bits(evt)), body_text


def result_summary(evt):
    """PORT of resultSummary -> the summary line's text."""
    bits = result_bits(evt)
    return "result" + (" (" + ", ".join(bits) + ")" if bits else "")


def result_output(evt):
    """PORT of resultBody's text decisions -> (stdout_text,
    stderr_text, note_text, extras_text, no_output_flag). The extras
    go through the tailored result body first (a read window, a
    replacement count, a task list, an ok ack), then raw JSON."""
    envelope = result_envelope(evt)
    payload = evt.get("payload") or {}
    parsed = parse_maybe_json(payload.get("result"))
    out = err = ""
    if is_plain_object(parsed):
        out = str_text(envelope.get("stdout"))
        err = str_text(envelope.get("stderr"))
    elif isinstance(parsed, str):
        out = parsed  # a JSON string literal
    elif isinstance(payload.get("result"), str):
        out = payload["result"]
    else:
        out = str_text(envelope.get("stdout"))
        err = str_text(envelope.get("stderr"))
    note = ""
    extras = ""
    if not out and not err:
        note = str_text(envelope.get("error")) or str_text(envelope.get("note")) \
            or ("timed out" if envelope.get("timeout") else "")
        if not note:
            remainder = result_extras(envelope)
            if remainder:
                tailored = tailored_result_text(
                    str_text(payload.get("tool")), remainder)
                if tailored is not None:
                    extras = tailored
                else:
                    compact = json.dumps(remainder)
                    extras = compact if len(compact) <= 120 \
                        else json.dumps(remainder, indent=2)
    return out, err, note, extras, (not out and not err and not note and not extras)


def task_list_text(items):
    """PORT of taskList -> one '#id title status' line per item."""
    lines = []
    for item in items:
        if not is_plain_object(item):
            continue
        line = "#%s  %s" % (item.get("id", "?"), str_text(item.get("title")))
        if str_text(item.get("status")):
            line += " " + str_text(item.get("status"))
        lines.append(line)
    return "\n".join(lines)


def tailored_result_text(tool, extras):
    """PORT of tailoredResultBody -> the rendered text, or None for
    the generic JSON view."""
    if str_text(extras.get("content")) or isinstance(extras.get("total_lines"), int):
        meta = []
        if str_text(extras.get("path")):
            meta.append(str_text(extras.get("path")))
        if isinstance(extras.get("total_lines"), int):
            start = extras.get("line_start") \
                if isinstance(extras.get("line_start"), int) else 1
            end = extras.get("line_end") \
                if isinstance(extras.get("line_end"), int) else extras["total_lines"]
            meta.append("lines %d-%d of %d"
                        % (start, end, extras["total_lines"]))
            if isinstance(extras.get("bytes"), int):
                meta.append(fmt_bytes(extras["bytes"]))
        else:
            if str_text(extras.get("representation")):
                meta.append(str_text(extras.get("representation")))
            if isinstance(extras.get("bytes_read"), int):
                window = fmt_bytes(extras["bytes_read"])
                if isinstance(extras.get("total_bytes"), int):
                    window += " of " + fmt_bytes(extras["total_bytes"])
                meta.append(window)
            if isinstance(extras.get("offset"), int) and extras["offset"] > 0:
                meta.append("at " + fmt_bytes(extras["offset"]))
        if extras.get("truncated"):
            meta.append("truncated")
        parts = [" \u00b7 ".join(meta)]
        if str_text(extras.get("content")):
            # The content rides behind an expander: a read is
            # reference material, not the action's output.
            parts.append("[fold] content \u00b7 "
                         + fmt_bytes(len(str_text(extras.get("content")))))
        return "\n".join(parts)
    if tool == "replace_in_file" and isinstance(extras.get("replacements"), int):
        count = extras["replacements"]
        written = extras.get("bytes_written") \
            if isinstance(extras.get("bytes_written"), int) else 0
        return "%d replacement%s in %s \u00b7 %s" % (
            count, "" if count == 1 else "s",
            str_text(extras.get("path")), fmt_bytes(written))
    if tool == "write_file" and isinstance(extras.get("bytes"), int):
        return "wrote %s \u00b7 %s" % (str_text(extras.get("path")),
                                       fmt_bytes(extras["bytes"]))
    if tool == "overwrite_file" and isinstance(extras.get("bytes_written"), int):
        text = "rewrote %s \u00b7 %s" % (str_text(extras.get("path")),
                                         fmt_bytes(extras["bytes_written"]))
        if isinstance(extras.get("previous_bytes"), int):
            text += " (was %s)" % fmt_bytes(extras["previous_bytes"])
        return text
    if tool == "create_task" and isinstance(extras.get("task_id"), int):
        text = "created task #%d" % extras["task_id"]
        if str_text(extras.get("status")):
            text += " \u00b7 " + str_text(extras.get("status"))
        return text
    if tool == "get_task" and "id" in extras:
        return task_list_text([extras])
    for key in ("tasks", "goals"):
        if isinstance(extras.get(key), list):
            return task_list_text(extras[key])
    keys = list(extras)
    all_true = len(keys) > 0 and all(extras[key] is True for key in keys)
    all_scalars = 0 < len(keys) <= 4 and all(
        extras[key] is None or not isinstance(extras[key], (dict, list))
        for key in keys)
    if all_true:
        return "ok"
    if all_scalars:
        return " \u00b7 ".join("%s: %s" % (key, extras[key]) for key in keys)
    return None


def args_without(args, skip):
    """PORT of argsWithout: every member except `skip`, pretty-
    printed; '' when nothing is left."""
    rest = dict((key, value) for key, value in args.items() if key != skip)
    return json.dumps(rest, indent=2) if rest else ""


def tool_call_render(evt):
    """PORT of toolCallBody -> the rendered blocks as (kind, text)
    pairs: ('head', line) for a command-styled head, ('pre', text)
    for plain pre lines, ('para', text) for prose, ('fold', label)
    for collapsed payloads, ('diff', '- old') / ('diff', '+ new')
    for the replace_in_file diff. Bespoke bodies come from
    tailored_blocks; None falls back to the generic view."""
    payload = evt.get("payload") or {}
    tool = str_text(payload.get("tool")) or "tool"
    parsed = parse_maybe_json(payload.get("args"))
    args = parsed if is_plain_object(parsed) else None
    if args is not None:
        blocks = tailored_blocks(tool, args)
        if blocks is not None:
            return blocks
    blocks = [("head", tool)]
    if args is not None:
        rest = args_without(args, "command")
        if rest:
            blocks.append(("pre", rest))
    elif str_text(payload.get("args")):
        blocks.append(("pre", payload["args"]))
    elif payload.get("args") is not None and not isinstance(payload.get("args"), str):
        blocks.append(("pre", json.dumps(payload["args"], indent=2)))
    return blocks


def fmt_int(n):
    """PORT of fmtInt: thousands separators without locale."""
    digits = str(abs(int(round(n))))
    out = ""
    for i, char in enumerate(digits):
        if i > 0 and (len(digits) - i) % 3 == 0:
            out += ","
        out += char
    return ("-" if n < 0 else "") + out


def usage_line(payload):
    """PORT of eventNode's usage branch -> the one-line text. The
    daemon may report either token count alone."""
    bits = []
    if str_text(payload.get("model")):
        bits.append(payload["model"])
    in_tok = "%s in" % fmt_int(payload["prompt_tokens"]) \
        if isinstance(payload.get("prompt_tokens"), int) else ""
    out_tok = "%s out" % fmt_int(payload["completion_tokens"]) \
        if isinstance(payload.get("completion_tokens"), int) else ""
    if in_tok or out_tok:
        bits.append(in_tok + (" / " if in_tok and out_tok else "") + out_tok + " tok")
    if isinstance(payload.get("cached_tokens"), int):
        bits.append("%s cached" % fmt_int(payload["cached_tokens"]))
    if isinstance(payload.get("elapsed_ms"), int):
        bits.append(fmt_duration(payload["elapsed_ms"]))
    return " \u00b7 ".join(bits) if bits else "unreported"


def state_change_line(payload):
    """PORT of eventNode's state_change branch."""
    from_status = str_text(payload.get("from"))
    to_status = str_text(payload.get("to"))
    if from_status and to_status:
        return from_status + " \u2192 " + to_status
    return str_text(payload.get("text")) or json.dumps(payload)


# --- Quiet-call folding + result attachment (the live-tail polish):
# a successful side-effect tool whose content already renders as its
# own event folds its call/ack pair away; every other call carries
# its result as an attached block instead of a separate row. ---

FOLDABLE_TOOLS = {"update_progress": "progress", "add_comment": "comment",
                  "yeehaw": "yeehaw"}


def quiet_result(evt):
    """PORT of quietResult: an ack-shaped success with nothing to
    say - ok, no error/note/timeout, no output, extras only true/
    false/empty-string/row pointers (_id integers whose content
    renders elsewhere)."""
    envelope = result_envelope(evt)
    if envelope.get("ok") is False:
        return False
    if isinstance(envelope.get("exit_code"), int) and envelope["exit_code"] != 0:
        return False
    if str_text(envelope.get("error")) or str_text(envelope.get("note")):
        return False
    if envelope.get("timeout"):
        return False
    if str_text(envelope.get("stdout")) or str_text(envelope.get("stderr")):
        return False
    for key, value in result_extras(envelope).items():
        if value is True or value is False:
            continue
        if isinstance(value, str) and value == "":
            continue
        if isinstance(value, int) and key.endswith("_id"):
            continue
        return False
    return True


def same_session(a, b):
    """PORT of sameSession: equal, or either side unlabelled."""
    return a == b or a is None or b is None


def match_result_index(rows, index):
    """PORT of matchResultIndex: the first tool_result after the
    call is its own (one call per turn). The live wire's result
    payloads carry no turn of their own - matching rides on order,
    tool name and session; a turn mismatch (both present) is the
    only rejection."""
    evt = rows[index]
    payload = evt.get("payload") or {}
    if not isinstance(payload.get("turn"), int):
        return -1
    for j in range(index + 1, len(rows)):
        other = rows[j]
        if other.get("kind") == "tool_result":
            other_payload = other.get("payload") or {}
            other_turn = other_payload.get("turn")
            if (other_turn is None or other_turn == payload.get("turn")) \
                    and same_session(other.get("session_id"), evt.get("session_id")) \
                    and str_text(other_payload.get("tool")) == str_text(payload.get("tool")):
                return j
            return -1
        if other.get("kind") == "tool_call":
            return -1
    return -1


def folded_quiet_calls(rows):
    """PORT of foldedQuietCalls -> the folded row indices."""
    folded = set()
    for index, evt in enumerate(rows):
        if evt.get("kind") != "tool_call":
            continue
        payload = evt.get("payload") or {}
        if not isinstance(payload.get("turn"), int):
            continue
        companion = FOLDABLE_TOOLS.get(str_text(payload.get("tool")))
        if not companion:
            continue
        result_index = match_result_index(rows, index)
        if result_index < 0:
            continue
        has_companion = any(
            rows[k].get("kind") == companion
            and same_session(rows[k].get("session_id"), evt.get("session_id"))
            for k in range(index + 1, result_index))
        if has_companion and quiet_result(rows[result_index]):
            folded.add(index)
            folded.add(result_index)
    return folded


def attach_pairs(rows, folded):
    """PORT of render's attachment walk -> (calls_by_id,
    rides_by_id): every non-folded call carries its matching
    result; the result then never renders as its own row."""
    calls_by_id = {}
    rides_by_id = {}
    for index, evt in enumerate(rows):
        if index in folded or evt.get("kind") != "tool_call":
            continue
        result_index = match_result_index(rows, index)
        if result_index >= 0 and result_index not in folded:
            calls_by_id[evt["id"]] = rows[result_index]["id"]
            rides_by_id[rows[result_index]["id"]] = True
    return calls_by_id, rides_by_id


# --- The state badge: only a working (or assigned) task streams
# live; every other status names itself on the pill, still. ---

LIVE_STATUSES = {"working", "assigned"}
BADGE_LABELS = {"suspended": "parked", "review_requested": "review requested"}


def badge_line(status, stream_state="live"):
    """PORT of updateBadge -> (class, label)."""
    if status in LIVE_STATUSES:
        if stream_state == "retry":
            return "log-live retry", "reconnecting..."
        return "log-live", "live"
    label = BADGE_LABELS.get(status, status) or "idle"
    return "log-live still badge badge-" + status, label


DIFF_CONTEXT = 2
DIFF_MAX_LINES = 400


def _push_side(rows, lines, begin, end, marker):
    """PORT of pushSide."""
    count = end - begin
    half = DIFF_MAX_LINES // 2
    if count <= DIFF_MAX_LINES:
        for i in range(begin, end):
            rows.append([marker, lines[i]])
        return
    for j in range(begin, begin + half):
        rows.append([marker, lines[j]])
    rows.append([marker, "... %d more lines ..." % (count - half * 2)])
    for k in range(end - half, end):
        rows.append([marker, lines[k]])


def diff_rows(old_text, new_text):
    """PORT of diffRows: common head/tail become clamped context,
    the differing middle renders as -old/+new."""
    old_lines = old_text.split("\n")
    new_lines = new_text.split("\n")
    start = 0
    while start < len(old_lines) and start < len(new_lines) \
            and old_lines[start] == new_lines[start]:
        start += 1
    end_old, end_new = len(old_lines), len(new_lines)
    while end_old > start and end_new > start \
            and old_lines[end_old - 1] == new_lines[end_new - 1]:
        end_old -= 1
        end_new -= 1
    rows = []
    has_middle = end_old > start or end_new > start
    if has_middle:
        for i in range(max(0, start - DIFF_CONTEXT), start):
            rows.append([" ", old_lines[i]])
        _push_side(rows, old_lines, start, end_old, "-")
        _push_side(rows, new_lines, start, end_new, "+")
        for i in range(end_old, min(len(old_lines), end_old + DIFF_CONTEXT)):
            rows.append([" ", old_lines[i]])
    return rows


FILE_LANGUAGES = {".vala": "Vala", ".vapi": "Vapi", ".c": "C", ".h": "C",
                  ".py": "Python", ".js": "JavaScript", ".ts": "TypeScript",
                  ".html": "HTML", ".css": "CSS", ".json": "JSON",
                  ".md": "Markdown", ".sh": "Shell", ".bash": "Shell",
                  ".xml": "XML", ".yml": "YAML", ".yaml": "YAML",
                  ".sql": "SQL", ".toml": "TOML", ".ini": "INI",
                  ".cfg": "Config", ".diff": "Diff", ".patch": "Diff",
                  ".rst": "reST", ".txt": "Text"}


def language_of(path):
    """PORT of languageOf: the extension chip, with meson.build and
    makefile as whole-name matches; '' when nothing matches."""
    lower = path.lower()
    name = lower[lower.rfind("/") + 1:]
    if name == "meson.build":
        return "Meson"
    if name == "makefile":
        return "Make"
    dot = name.rfind(".")
    if dot < 1 or dot == len(name) - 1:
        return ""
    return FILE_LANGUAGES.get(name[dot:], "")


def content_bits(content):
    """PORT of contentBits: line count and byte size."""
    lines = content.count("\n") + 1
    return ["%d line%s" % (lines, "" if lines == 1 else "s"),
            fmt_bytes(len(content))]


def tailored_blocks(tool, args):
    """PORT of tailoredCallBody -> the bespoke blocks, or None to
    fall back to the generic JSON view (fields missing - e.g. a
    truncated args string that clamped mid-JSON)."""
    path = str_text(args.get("path"))
    command = str_text(args.get("command"))
    question = str_text(args.get("question"))
    bits = []

    def head(text, more=None):
        block = ("head", text)
        if more:
            block = ("head", text + "  \u00b7 " + " \u00b7 ".join(more))
        return block

    if tool == "run_command":
        if not command:
            return None
        if isinstance(args.get("timeout_ms"), int):
            bits.append("timeout %ds" % round(args["timeout_ms"] / 1000))
        return [head("$ " + command, bits)]
    if tool == "read_file":
        if not path:
            return None
        if isinstance(args.get("offset"), int) and args["offset"] > 1:
            bits.append("from line %d" % args["offset"])
        if isinstance(args.get("limit"), int):
            bits.append("%d lines" % args["limit"])
        return [head("read_file " + path, bits)]
    if tool in ("write_file", "overwrite_file", "temp_doc"):
        content = str_text(args.get("content"))
        if not content:
            return None
        if tool == "temp_doc":
            return [head("temp_doc", content_bits(content)), ("pre", content)]
        if not path:
            return None
        more = content_bits(content)
        language = language_of(path)
        if language:
            more.insert(0, language)
        return [head(tool + " " + path, more), ("pre", content)]
    if tool == "replace_in_file":
        old = str_text(args.get("old"))
        if not path or not old:
            return None
        if isinstance(args.get("expected_count"), int):
            bits.append("expect %d" % args["expected_count"])
        blocks = [head("replace_in_file " + path, bits)]
        rows = diff_rows(old, str_text(args.get("new")))
        if rows:
            blocks.extend(("diff", marker + " " + text) for marker, text in rows)
        return blocks
    if tool in ("enqueue_task", "delete_task", "get_task"):
        if not isinstance(args.get("task_id"), int):
            return None
        return [head(tool + " #" + str(args["task_id"]))]
    if tool in ("update_progress", "add_comment", "save_note"):
        text = str_text(args.get("text")) or str_text(args.get("body"))
        if not text:
            return None
        if tool == "save_note" and str_text(args.get("project")):
            bits.append(str_text(args.get("project")))
        return [head(tool, bits), ("para", text)]
    if tool == "ask_question":
        if not question:
            return None
        blocks = [head("ask_question"), ("para", "Q: " + question)]
        if isinstance(args.get("options"), list) and args["options"]:
            blocks.append(("options", [str(o) for o in args["options"]]))
        return blocks
    return None


def command_output_render(evt):
    """PORT of eventNode's command_output branch ->
    (command_prefix, output_text)."""
    payload = evt.get("payload") or {}
    command = str_text(payload.get("command"))
    text = payload.get("text") or payload.get("chunk") or ""
    return ("$ " + command if command else ""), text


# --- Fixtures: copied from the live ledger (web-db.sqlite, events
# table) and the test fake's object-shaped pushes - the real wire
# shapes, inner JSON spelled out as the daemon serialises it. ---

# The daemon's inner JSON, built with json.dumps so the escaping
# is exactly what the wire carries (a JSON string holding JSON).
inner_call = json.dumps({"command": "cd /workspace && ls"})
inner_progress = json.dumps(
    {"text": "Root-caused the SSE listener; next: the payload parse"})
inner_enqueue = json.dumps({"task_id": 3})
inner_result = json.dumps(
    {"ok": True, "exit_code": 0,
     "stdout": "find /workspace/task-1 -maxdepth 2",
     "stderr": "", "duration_ms": 11,
     "bytes_out": 6042, "bytes_err": 0, "truncated": True})
inner_error = json.dumps(
    {"ok": False,
     "error": "suspend refused: 3 draft task(s) were never enqueued"})
inner_empty = json.dumps({"ok": True, "exit_code": 0,
                          "stdout": "", "stderr": ""})

FIXTURE_RUN_COMMAND_CALL = json.loads(
    '{"turn": 52, "tool": "run_command", "args": ' + json.dumps(inner_call) + '}')

FIXTURE_UPDATE_PROGRESS_CALL = json.loads(
    '{"turn": 6, "tool": "update_progress", "args": '
    + json.dumps(inner_progress) + '}')

FIXTURE_ENQUEUE_CALL = json.loads(
    '{"turn": 5, "tool": "enqueue_task", "args": '
    + json.dumps(inner_enqueue) + '}')

FIXTURE_RUN_COMMAND_RESULT = json.loads(
    '{"tool": "run_command", "result": ' + json.dumps(inner_result)
    + ', "turn": 52}')

FIXTURE_ERROR_RESULT = json.loads(
    '{"tool": "suspend", "result": ' + json.dumps(inner_error) + '}')

FIXTURE_EMPTY_RESULT = json.loads(
    '{"tool": "run_command", "result": ' + json.dumps(inner_empty) + '}')

# Not valid JSON: the live daemon emitted result strings that are
# truncated raw output (72 rows in the production ledger).
FIXTURE_NOT_JSON_RESULT = {
    "tool": "run_command",
    "result": '{"ok":true,"exit_code":0,"stdout":"=== Spry ===\n'
              'MANIFEST.usm  README.md  meson.build',
}

# command_output: the harness fake's shape; `command` is a member.
FIXTURE_COMMAND_OUTPUT = {"command": "meson test -C builddir",
                          "text": "nobody:/task-1$ done\r\n"}

# The harness fake's legacy object-shaped payloads (plan 17's own
# tests push these): args/result as real objects.
FIXTURE_OBJECT_CALL = {"turn": 1, "tool": "run_command",
                       "args": {"command": "echo hello"}}
FIXTURE_OBJECT_RESULT = {"turn": 1, "tool": "run_command",
                         "stdout": "hello\n", "exit_code": 0}

# Tailored-rendering fixtures (plan 18): the file tools' arg shapes
# straight off the daemon's tool schemas.
FIXTURE_REPLACE_CALL = {"turn": 40, "tool": "replace_in_file", "args": {
    "path": "src/pages/app.css",
    "old": ".event-turn {\n    border-left: 3px solid #ddf4ff;\n}",
    "new": ".event-turn {\n    border-left: 3px solid #ddf4ff;\n    color: var(--ink-muted);\n}",
    "expected_count": 1}}
FIXTURE_WRITE_CALL = {"turn": 41, "tool": "write_file", "args": {
    "path": "notes.md", "content": "# Notes\n- first\n- second"}}
# Args clamped mid-JSON by the daemon's display clamp: parsing
# fails, so the bespoke renderer must step aside and the generic
# view shows the raw text.
FIXTURE_TRUNCATED_ARGS = {"turn": 42, "tool": "write_file", "args":
                          '{"path":"big.bin","content":"AAA'}


class PayloadParserTests(unittest.TestCase):

    def test_string_args_decode_to_an_object(self):
        parsed = parse_maybe_json(FIXTURE_RUN_COMMAND_CALL["args"])
        self.assertEqual(parsed, {"command": "cd /workspace && ls"})

    def test_non_json_string_answers_none(self):
        self.assertIsNone(parse_maybe_json(
            FIXTURE_NOT_JSON_RESULT["result"]))

    def test_objects_pass_through(self):
        self.assertEqual(parse_maybe_json({"command": "ls"}), {"command": "ls"})
        self.assertEqual(parse_maybe_json(None), None)


class ToolCallRenderTests(unittest.TestCase):
    """The command the agent ran must surface in EVERY render path,
    and each known tool renders through its tailored body."""

    def evt(self, payload):
        return {"kind": "tool_call", "payload": payload}

    def test_run_command_call_heads_with_the_dollar(self):
        blocks = tool_call_render(self.evt(FIXTURE_RUN_COMMAND_CALL))
        self.assertEqual(blocks[0], ("head", "$ cd /workspace && ls"))

    def test_run_command_timeout_rides_the_head(self):
        blocks = tool_call_render(self.evt({"tool": "run_command", "args":
                                            {"command": "make", "timeout_ms": 900000}}))
        self.assertEqual(blocks[0],
                         ("head", "$ make  \u00b7 timeout 900s"))

    def test_update_progress_call_renders_prose_not_json(self):
        """The old renderer dumped {"text": ...} as JSON; the
        tailored body renders the text itself."""
        blocks = tool_call_render(self.evt(FIXTURE_UPDATE_PROGRESS_CALL))
        self.assertEqual(blocks[0], ("head", "update_progress"))
        self.assertEqual(blocks[1],
                         ("para", "Root-caused the SSE listener; next: the payload parse"))
        self.assertNotIn('"text"', json.dumps(blocks))

    def test_enqueue_call_renders_the_id(self):
        blocks = tool_call_render(self.evt(FIXTURE_ENQUEUE_CALL))
        self.assertEqual(blocks, [("head", "enqueue_task #3")])

    def test_replace_in_file_renders_a_diff(self):
        blocks = tool_call_render(self.evt(FIXTURE_REPLACE_CALL))
        self.assertEqual(blocks[0],
                         ("head", "replace_in_file src/pages/app.css  \u00b7 expect 1"))
        diff_texts = [text for kind, text in blocks if kind == "diff"]
        joined = " ".join(diff_texts)
        self.assertIn("  .event-turn {", joined)
        # The untouched border line is context; only the added color
        # line is an addition - no phantom removals.
        self.assertIn("      border-left: 3px solid #ddf4ff;", joined)
        self.assertIn("+     color: var(--ink-muted);", joined)
        self.assertIn("  }", joined)
        self.assertFalse([t for t in diff_texts if t.startswith("- ")],
                         "an unchanged line must never render as a removal")

    def test_write_file_renders_language_lines_and_content(self):
        blocks = tool_call_render(self.evt(FIXTURE_WRITE_CALL))
        self.assertEqual(blocks[0],
                         ("head", "write_file notes.md  \u00b7 Markdown \u00b7 3 lines \u00b7 24 B"))
        self.assertEqual(blocks[1], ("pre", "# Notes\n- first\n- second"))

    def test_meson_build_gets_its_language(self):
        self.assertEqual(language_of("sub/meson.build"), "Meson")
        self.assertEqual(language_of("src/app.vala"), "Vala")
        self.assertEqual(language_of("Makefile".lower()), "Make")
        self.assertEqual(language_of("data/bin"), "")

    def test_truncated_args_fall_back_to_the_raw_view(self):
        blocks = tool_call_render(self.evt(FIXTURE_TRUNCATED_ARGS))
        self.assertEqual(blocks[0], ("head", "write_file"))
        self.assertEqual(blocks[1][0], "pre")
        self.assertIn('"path":"big.bin"', blocks[1][1])

    def test_unknown_tool_falls_back_to_json_args(self):
        blocks = tool_call_render(self.evt(
            {"tool": "brand_new_tool", "args": {"deep": {"a": 1}}}))
        self.assertEqual(blocks[0], ("head", "brand_new_tool"))
        self.assertIn('"deep"', blocks[1][1])

    def test_object_shaped_call_still_works(self):
        blocks = tool_call_render(self.evt(FIXTURE_OBJECT_CALL))
        self.assertEqual(blocks, [("head", "$ echo hello")])

    def test_non_json_args_render_raw(self):
        blocks = tool_call_render(self.evt(
            {"tool": "run_command", "args": "not json at all"}))
        self.assertEqual(blocks,
                         [("head", "run_command"), ("pre", "not json at all")])

    def test_no_args_at_all_renders_the_tool_name(self):
        blocks = tool_call_render(self.evt({"tool": "list_tasks"}))
        self.assertEqual(blocks, [("head", "list_tasks")])

    def test_bespoke_missing_fields_step_aside(self):
        """A replace_in_file without its old string (clamped args
        that still parse) falls back rather than lying."""
        blocks = tool_call_render(self.evt(
            {"tool": "replace_in_file", "args": {"path": "x.c"}}))
        self.assertEqual(blocks[0], ("head", "replace_in_file"))
        self.assertEqual(blocks[1][0], "pre")


class DiffRowTests(unittest.TestCase):
    """The replace_in_file diff: context clamping and honest
    replacement semantics."""

    def test_single_line_swap(self):
        rows = diff_rows("alpha\nbeta\ngamma", "alpha\nBETA\ngamma")
        self.assertEqual(rows, [[" ", "alpha"], ["-", "beta"],
                                ["+", "BETA"], [" ", "gamma"]])

    def test_context_is_clamped_to_two_lines_per_side(self):
        old = "\n".join(["ctx"] * 10 + ["old-line"] + ["ctx"] * 10)
        new = "\n".join(["ctx"] * 10 + ["new-line"] + ["ctx"] * 10)
        rows = diff_rows(old, new)
        context = [text for marker, text in rows if marker == " "]
        self.assertEqual(context, ["ctx"] * 4)

    def test_whole_file_rewrite_shows_no_context(self):
        rows = diff_rows("a\nb\nc", "x\ny\nz")
        self.assertEqual([m for m, _ in rows], ["-", "-", "-", "+", "+", "+"])

    def test_deletion_only_keeps_its_context(self):
        rows = diff_rows("keep\ndrop\nkeep", "keep\nkeep")
        self.assertEqual(rows, [[" ", "keep"], ["-", "drop"], [" ", "keep"]])

    def test_empty_new_text(self):
        rows = diff_rows("line", "")
        self.assertEqual(rows, [["-", "line"], ["+", ""]])

    def test_oversized_side_keeps_head_and_tail(self):
        old = "\n".join("line-%04d" % i for i in range(1000))
        rows = diff_rows(old, "changed")
        dels = [text for marker, text in rows if marker == "-"]
        self.assertEqual(len(dels), 401)
        self.assertIn("more lines", dels[200])
        self.assertEqual(dels[0], "line-0000")
        self.assertEqual(dels[-1], "line-0999")

    def test_identical_text_renders_nothing(self):
        self.assertEqual(diff_rows("same\nsame", "same\nsame"), [])


class UsageLineTests(unittest.TestCase):

    def test_full_payload_is_one_line(self):
        line = usage_line({"model": "glm-4.7", "prompt_tokens": 45200000,
                           "completion_tokens": 1100000, "cached_tokens": 4500,
                           "elapsed_ms": 3200})
        self.assertEqual(line,
                         "glm-4.7 \u00b7 45,200,000 in / 1,100,000 out tok "
                         "\u00b7 4,500 cached \u00b7 3.2 s")

    def test_unreported_when_no_numbers(self):
        self.assertEqual(usage_line({}), "unreported")

    def test_partial_payload_skips_missing_members(self):
        line = usage_line({"completion_tokens": 5, "elapsed_ms": 40})
        self.assertEqual(line, "5 out tok \u00b7 40 ms")


class StateChangeLineTests(unittest.TestCase):

    def test_from_to_arrow(self):
        self.assertEqual(
            state_change_line({"from": "working", "to": "blocked",
                               "text": "Status \u2192 blocked"}),
            "working \u2192 blocked")

    def test_text_fallback(self):
        self.assertEqual(
            state_change_line({"text": "Goal created (branch x)"}),
            "Goal created (branch x)")

    def test_json_fallback(self):
        self.assertEqual(state_change_line({}), "{}")


QUIET_ACK = json.dumps({"ok": True, "recorded": True})
FAILED_ACK = json.dumps({"ok": False, "error": "web callback unreachable"})


def progress_trio(result_inner=QUIET_ACK, with_companion=True,
                  result_turn=None):
    """The live-tail's repetitive trio: an update_progress call,
    the progress event carrying the same text, then the ack
    result. The live wire's result payloads carry NO turn of their
    own (the payload is {tool, result}); pass result_turn to model
    a turn-stamped result."""
    rows = [{"id": 1, "kind": "tool_call", "session_id": 201, "payload": {
        "turn": 1551, "tool": "update_progress",
        "args": json.dumps({"text": "Segfault located"})}}]
    if with_companion:
        rows.append({"id": 2, "kind": "progress", "session_id": 201,
                     "payload": {"text": "Segfault located"}})
    result_payload = {"tool": "update_progress", "result": result_inner}
    if result_turn is not None:
        result_payload["turn"] = result_turn
    rows.append({"id": 3, "kind": "tool_result", "session_id": 201,
                 "payload": result_payload})
    return rows


class QuietFoldingTests(unittest.TestCase):
    """A successful update_progress renders as its progress event
    ALONE; the duplicate call and the bare ack fold away. A failure
    unfolds both."""

    def test_quiet_trio_folds_call_and_result(self):
        """The live wire's shape: the result carries no turn."""
        folded = folded_quiet_calls(progress_trio())
        self.assertEqual(folded, {0, 2})

    def test_turn_stamped_result_also_folds(self):
        folded = folded_quiet_calls(progress_trio(result_turn=1551))
        self.assertEqual(folded, {0, 2})

    def test_turn_mismatch_never_folds(self):
        """A result stamped with another turn is not this call's
        answer."""
        self.assertEqual(
            folded_quiet_calls(progress_trio(result_turn=9999)), set())

    def test_add_comment_folds_behind_its_comment_event(self):
        """The add_comment result carries a comment_id row pointer -
        quiet - and the comment event between carries the body."""
        rows = [
            {"id": 1, "kind": "tool_call", "session_id": 201, "payload": {
                "turn": 60, "tool": "add_comment",
                "args": json.dumps({"body": "Merged and pushed."})}},
            {"id": 2, "kind": "comment", "session_id": 201, "actor": "model",
             "payload": {"body": "Merged and pushed."}},
            {"id": 3, "kind": "tool_result", "session_id": 201, "payload": {
                "tool": "add_comment",
                "result": json.dumps({"ok": True, "comment_id": 234})}},
        ]
        self.assertEqual(folded_quiet_calls(rows), {0, 2})

    def test_failed_ack_unfolds_the_pair(self):
        self.assertEqual(folded_quiet_calls(progress_trio(FAILED_ACK)), set())

    def test_missing_companion_unfolds(self):
        """Without the progress event between them there would be
        nothing to show instead - keep the pair."""
        self.assertEqual(folded_quiet_calls(progress_trio(with_companion=False)),
                         set())

    def test_missing_result_unfolds(self):
        rows = progress_trio()[:2]
        self.assertEqual(folded_quiet_calls(rows), set())

    def test_non_foldable_tool_never_folds(self):
        rows = [
            {"id": 1, "kind": "tool_call", "session_id": 201, "payload": {
                "turn": 5, "tool": "run_command",
                "args": json.dumps({"command": "ls"})}},
            {"id": 2, "kind": "tool_result", "session_id": 201, "payload": {
                "tool": "run_command", "result": QUIET_ACK}},
        ]
        self.assertEqual(folded_quiet_calls(rows), set())


class ResultAttachmentTests(unittest.TestCase):
    """A non-folded call carries its result as an attached block
    INSIDE its row; the result never renders as a separate row."""

    def test_call_and_result_pair_up(self):
        rows = [
            {"id": 10, "kind": "tool_call", "session_id": 7, "payload": {
                "turn": 5, "tool": "run_command",
                "args": json.dumps({"command": "make"})}},
            {"id": 11, "kind": "tool_result", "session_id": 7, "payload": {
                "tool": "run_command",
                "result": json.dumps({"ok": True, "exit_code": 0,
                                      "stdout": "built\n", "stderr": "",
                                      "duration_ms": 900})}},
        ]
        calls_by_id, rides_by_id = attach_pairs(rows, set())
        self.assertEqual(calls_by_id, {10: 11})
        self.assertEqual(rides_by_id, {11: True})

    def test_turn_less_live_result_pairs_up(self):
        """The live wire's shape: the result carries no turn, so
        matching rides on order, tool and session."""
        rows = [
            {"id": 20, "kind": "tool_call", "session_id": 201, "payload": {
                "turn": 1731, "tool": "run_command",
                "args": json.dumps({"command": "make test"})}},
            {"id": 21, "kind": "tool_result", "session_id": 201, "payload": {
                "tool": "run_command",
                "result": json.dumps({"ok": True, "exit_code": 0,
                                      "stdout": "ok\n", "stderr": "",
                                      "duration_ms": 25000})}},
        ]
        calls_by_id, rides_by_id = attach_pairs(rows, set())
        self.assertEqual(calls_by_id, {20: 21})
        self.assertEqual(rides_by_id, {21: True})

    def test_long_command_output_is_visible_when_attached(self):
        """A command's output must be VISIBLE riding with its call -
        never behind a click - even past the inline limit; CSS
        clamps the overflow."""
        big_output = "line %d\n" * 60 % tuple(range(60))
        evt = {"kind": "tool_result", "payload": {
            "tool": "run_command",
            "result": json.dumps({"ok": True, "exit_code": 0,
                                  "stdout": big_output, "stderr": "",
                                  "duration_ms": 5000})}}
        standalone, _, body = result_view(evt)
        attached, _, attached_body = result_view(evt, attached=True)
        self.assertFalse(standalone, "long standalone results collapse")
        self.assertTrue(attached, "attached results are always inline")
        self.assertEqual(attached_body, body)

    def test_folded_pair_attaches_nothing(self):
        calls_by_id, rides_by_id = attach_pairs(
            progress_trio(), folded_quiet_calls(progress_trio()))
        self.assertEqual((calls_by_id, rides_by_id), ({}, {}))

    def test_result_without_visible_call_stands_alone(self):
        rows = progress_trio()[2:]  # only the result is visible
        calls_by_id, rides_by_id = attach_pairs(rows, set())
        self.assertEqual((calls_by_id, rides_by_id), ({}, {}))


class BadgeLineTests(unittest.TestCase):
    """The badge reflects the TASK first: live only while a working
    (or assigned) task streams; every other status names itself."""

    def test_working_is_live(self):
        self.assertEqual(badge_line("working"), ("log-live", "live"))

    def test_assigned_is_live(self):
        self.assertEqual(badge_line("assigned"), ("log-live", "live"))

    def test_retry_only_rides_a_live_task(self):
        self.assertEqual(badge_line("working", "retry"),
                         ("log-live retry", "reconnecting..."))
        self.assertEqual(badge_line("failed", "retry")[1], "failed")

    def test_suspended_shows_as_parked(self):
        cls, label = badge_line("suspended")
        self.assertEqual(label, "parked")
        self.assertIn("still badge badge-suspended", cls)

    def test_review_requested_uses_the_task_page_word(self):
        self.assertEqual(badge_line("review_requested")[1],
                         "review requested")

    def test_terminal_states_name_themselves(self):
        self.assertEqual(badge_line("done")[1], "done")
        self.assertEqual(badge_line("failed")[1], "failed")

    def test_unknown_status_falls_back_to_idle(self):
        self.assertEqual(badge_line("")[1], "idle")


class ToolResultRenderTests(unittest.TestCase):

    def evt(self, payload):
        return {"kind": "tool_result", "payload": payload}

    def test_summary_comes_from_the_envelope(self):
        line = result_summary(self.evt(FIXTURE_RUN_COMMAND_RESULT))
        self.assertEqual(line,
                         "result (exit 0, 5.9 KiB, truncated, 11 ms, turn 52)")

    def test_summary_marks_failed_envelopes(self):
        line = result_summary(self.evt(
            {"tool": "run_command", "result": '{"ok":false,"error":"x"}'}))
        self.assertEqual(line, "result (failed)")

    def test_body_shows_the_envelopes_stdout(self):
        out, err, note, extras, no_output = result_output(
            self.evt(FIXTURE_RUN_COMMAND_RESULT))
        self.assertEqual(out, "find /workspace/task-1 -maxdepth 2")
        self.assertEqual(err, "")
        self.assertFalse(no_output)

    def test_non_json_result_renders_as_output_text(self):
        out, err, note, extras, no_output = result_output(
            self.evt(FIXTURE_NOT_JSON_RESULT))
        self.assertIn("=== Spry ===", out)
        self.assertFalse(no_output,
                         "raw output must never be called (no output)")

    def test_error_envelope_is_not_no_output(self):
        out, err, note, extras, no_output = result_output(
            self.evt(FIXTURE_ERROR_RESULT))
        self.assertEqual((out, err), ("", ""))
        self.assertIn("suspend refused", note)
        self.assertFalse(no_output,
                         "an error member means the result said something")

    def test_genuinely_empty_result_says_no_output(self):
        out, err, note, extras, no_output = result_output(self.evt(FIXTURE_EMPTY_RESULT))
        self.assertEqual((out, err, note, extras), ("", "", "", ""))
        self.assertTrue(no_output)

    def test_object_shaped_result_still_works(self):
        line = result_summary(self.evt(FIXTURE_OBJECT_RESULT))
        self.assertEqual(line, "result (exit 0, 6 B, turn 1)")
        out, err, note, extras, no_output = result_output(self.evt(FIXTURE_OBJECT_RESULT))
        self.assertEqual(out, "hello\n")


    def test_tool_answer_envelope_is_rendered_not_swallowed(self):
        """The live ledger's update_progress/create_task/finish
        results carry {"recorded":true} / {"ok":true,"task_id":4}
        / {"accepted":true} with empty streams: that IS the tool's
        answer and must render, not "(no output)"."""
        for inner in ('{"recorded":true}',
                      '{"ok":true,"task_id":4,"status":"draft"}',
                      '{"accepted":true}',
                      '{"suspended":true}',
                      '{"ok":true,"status":"pending"}'):
            out, err, note, extras, no_output = result_output(self.evt(
                {"tool": "update_progress", "result": inner}))
            self.assertFalse(no_output, inner)
            self.assertEqual((out, err, note), ("", "", ""))
            self.assertTrue(extras, inner)

    def test_list_tasks_answer_renders_as_a_list(self):
        inner = json.dumps({"ok": True, "tasks": [
            {"id": 6, "title": "UX notes", "status": "queued"},
            {"id": 5, "title": "Event log parse", "status": "working"}]})
        out, err, note, extras, no_output = result_output(self.evt(
            {"tool": "list_tasks", "result": inner}))
        self.assertFalse(no_output)
        self.assertIn("UX notes", extras)
        self.assertIn("Event log parse", extras)
        self.assertIn("#6", extras)
        self.assertNotIn('"tasks"', extras,
                         "a task list renders as rows, not a JSON dump")

    def test_replace_in_file_result_counts_the_replacements(self):
        inner = json.dumps({"ok": True, "path": "src/app.css",
                            "replacements": 2, "bytes_written": 2048})
        out, err, note, extras, no_output = result_output(self.evt(
            {"tool": "replace_in_file", "result": inner}))
        self.assertEqual(extras, "2 replacements in src/app.css \u00b7 2.0 KiB")

    def test_read_file_result_renders_the_window(self):
        inner = json.dumps({"ok": True, "path": "src/main.vala",
                            "total_lines": 540, "line_start": 100,
                            "line_end": 160, "bytes": 4200,
                            "truncated": False, "content": "int main() {"})
        out, err, note, extras, no_output = result_output(self.evt(
            {"tool": "read_file", "result": inner}))
        self.assertIn("src/main.vala \u00b7 lines 100-160 of 540 \u00b7 4.1 KiB", extras)
        self.assertIn("[fold] content \u00b7 12 B", extras)

    def test_read_binary_result_renders_the_window(self):
        inner = json.dumps({"ok": True, "path": "model.bin",
                            "representation": "hex", "total_bytes": 1024,
                            "offset": 256, "bytes_read": 512,
                            "truncated": False,
                            "content": "00000100  7f 45 4c 46"})
        out, err, note, extras, no_output = result_output(self.evt(
            {"tool": "read_binary", "result": inner}))
        self.assertIn("model.bin \u00b7 hex \u00b7 512 B of 1.0 KiB \u00b7 at 256 B", extras)
        self.assertIn("[fold] content", extras)

    def test_write_file_result_states_the_write(self):
        out, err, note, extras, no_output = result_output(self.evt(
            {"tool": "write_file",
             "result": '{"ok":true,"path":"notes.md","bytes":24}'}))
        self.assertEqual(extras, "wrote notes.md \u00b7 24 B")

    def test_create_task_result_names_the_task(self):
        out, err, note, extras, no_output = result_output(self.evt(
            {"tool": "create_task",
             "result": '{"ok":true,"task_id":4,"status":"draft"}'}))
        self.assertEqual(extras, "created task #4 \u00b7 draft")

    def test_pure_acknowledgement_is_an_ok_line(self):
        out, err, note, extras, no_output = result_output(self.evt(
            {"tool": "update_progress", "result": '{"ok":true,"recorded":true}'}))
        self.assertEqual(extras, "ok")

    def test_run_command_with_empty_streams_is_still_no_output(self):
        """Only command plumbing present: the run genuinely said
        nothing - keep the honest "(no output)"."""
        out, err, note, extras, no_output = result_output(self.evt(
            {"tool": "run_command",
             "result": '{"ok":true,"exit_code":0,"stdout":"",'
                       '"stderr":"","duration_ms":5,"bytes_out":0,'
                       '"bytes_err":0,"truncated":false}'}))
        self.assertTrue(no_output)
        self.assertEqual(extras, "")



class CommandOutputRenderTests(unittest.TestCase):

    def test_command_is_prefixed_as_a_shell_line(self):
        command, text = command_output_render(
            {"kind": "command_output", "payload": FIXTURE_COMMAND_OUTPUT})
        self.assertEqual(command, "$ meson test -C builddir")
        self.assertEqual(text, "nobody:/task-1$ done\r\n")

    def test_text_only_chunk_keeps_the_old_render(self):
        command, text = command_output_render(
            {"kind": "command_output", "payload": {"text": "hello\r\n"}})
        self.assertEqual(command, "")
        self.assertEqual(text, "hello\r\n")


class SearchHaystackTests(unittest.TestCase):
    """A search for text inside the STRING-encoded args/result must
    still find the event: the haystack appends the raw strings."""

    def test_search_matches_through_the_string_encoding(self):
        payload = FIXTURE_RUN_COMMAND_CALL
        haystack = "tool_call daemon " + json.dumps(payload) + " " \
            + str_text(payload.get("args")) + " " \
            + str_text(payload.get("result"))
        self.assertIn("cd /workspace && ls", haystack)


class YeehawFilterTests(unittest.TestCase):
    """The celebration must ride the comments filter (goal 9 task 4b):
    a human scanning the conversation sees the yeehaw, and the
    comments toggle hides it like any other chatter."""

    def test_yeehaw_maps_to_comments(self):
        self.assertEqual(filter_of("yeehaw"), "comments")

    def test_comment_kinds_unchanged(self):
        for kind in ("comment", "answer", "verdict"):
            self.assertEqual(filter_of(kind), "comments")

    def test_other_kinds_unchanged(self):
        self.assertEqual(filter_of("tool_call"), "commands")
        self.assertEqual(filter_of("tool_result"), "results")
        self.assertEqual(filter_of("command_output"), "results")
        self.assertEqual(filter_of("usage"), "usage")
        self.assertEqual(filter_of("lifecycle"), "transitions")


class YeehawRenderTests(unittest.TestCase):
    """The yeehaw branch: the message in the bold .yeehaw-message span
    (the festive palette), never a JSON payload dump."""

    def test_message_renders_in_the_span(self):
        span, message = yeehaw_render(
            {"kind": "yeehaw", "payload": {"message": "Loved every line"}})
        self.assertEqual(span, "yeehaw-message")
        self.assertEqual(message, "Loved every line")

    def test_missing_message_is_blank_not_garbage(self):
        span, message = yeehaw_render({"kind": "yeehaw", "payload": {}})
        self.assertEqual(message, "")

    def test_non_string_message_is_blank(self):
        span, message = yeehaw_render(
            {"kind": "yeehaw", "payload": {"message": 42}})
        self.assertEqual(message, "")


class RendererSourceTests(unittest.TestCase):
    """The shipped script itself (no JS engine in this container, so
    the render decisions are pinned by the ports above plus these
    shape checks on the served source)."""

    def setUp(self):
        self.body = script_body()

    def test_parse_maybe_json_is_defined_and_used(self):
        self.assertIn("function parseMaybeJson(value) {", self.body)
        self.assertGreaterEqual(self.body.count("parseMaybeJson("), 4)

    def test_no_innerhtml_anywhere(self):
        self.assertNotIn(".innerHTML", self.body)

    def test_command_output_prefixes_the_dollar(self):
        self.assertIn("'$ ' + p.command", self.body)

    def test_tool_call_line_uses_parsed_args(self):
        self.assertIn("args.command", self.body)
        self.assertIn("toolCallBody", self.body)

    def test_result_reads_parsed_envelope(self):
        self.assertIn("resultEnvelope", self.body)
        self.assertIn("bytes_out", self.body)
        self.assertIn("duration_ms", self.body)

    def test_yeehaw_lands_in_the_comments_bucket(self):
        self.assertRegex(
            self.body,
            r"kind === 'comment' \|\| kind === 'answer' \|\| kind === 'verdict'[\s\S]*?"
            r"kind === 'yeehaw'\) return 'comments'")

    def test_yeehaw_renders_the_message_span(self):
        self.assertIn("evt.kind === 'yeehaw'", self.body)
        self.assertIn("'yeehaw-message'", self.body)
        self.assertIn("strText(p.message)", self.body)

    def test_tailored_renderers_present(self):
        self.assertIn("function tailoredCallBody(tool, args) {", self.body)
        self.assertIn("function diffRows(oldText, newText) {", self.body)
        self.assertIn("'event-diff'", self.body)
        self.assertIn("tailoredCallBody(tool, args)", self.body)


if __name__ == "__main__":
    unittest.main(verbosity=2)