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
src/pages/task_log.html two ways:

1. Statically - the <script> body must be ASCII-only (the page
   compiler HTML-escapes script bodies, so any non-ASCII literal is
   re-served as literal entity text like "&middot;" after the
   browser's one decode pass), 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")


def script_body():
    with open(PAGE, encoding="utf-8") as handle:
        source = handle.read()
    head, rest = source.split("<script>", 1)
    body, _ = rest.split("</script>", 1)
    assert head, "no content before the script block"
    return body


# ---------------------------------------------------------------------------
# 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 ("&middot;") 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_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_sits_above_the_feed(self):
        # Search the page markup only - the script's comments also
        # mention [data-log-feed], before the elements do.
        with open(PAGE, encoding="utf-8") as handle:
            markup = handle.read().split("</script>", 1)[1]
        more = markup.index("data-log-more")
        feed = markup.index("data-log-feed")
        self.assertLess(more, feed,
                        "the load-older control must precede the feed markup")

    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_summary(evt):
    """PORT of resultSummary -> the summary line's text."""
    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"])
    # PORT: a non-JSON result string carries no envelope; its own
    # size is the only metric we have.
    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 "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)."""
    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:
                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 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)


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


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"):
        content = str_text(args.get("content"))
        if not path or not content:
            return None
        return [head(tool + " " + path, [fmt_bytes(len(content))]),
                ("fold", "content \u00b7 " + fmt_bytes(len(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_folds_its_content(self):
        blocks = tool_call_render(self.evt(FIXTURE_WRITE_CALL))
        self.assertEqual(blocks[0], ("head", "write_file notes.md  \u00b7 24 B"))
        self.assertEqual(blocks[1], ("fold", "content \u00b7 24 B"))

    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({}), "{}")


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_json(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('"tasks"', extras)
        self.assertIn("UX notes", extras)

    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)