feat: Part 2 integration — piped query input in terminal log, RFC block handling, review verdict delivery, answer de-duplication, SSE piped input, expanded E2E suite (90 checks)

README.md

@@ -152,8 +152,13 @@ resumes with the answer in its conversation, and the task returns to
 ## Integration testing
 
 `test/integration_test.py` drives the whole loop against a running
-app + daemon pair: explore task, dependency gating, block + answer, and
-the audit trail (26 checks, Statum-style PASS/FAIL). Host run:
+app + daemon pair (88 checks, Statum-style PASS/FAIL): explore task,
+dependency gating, then the Part 2 feature tour — `saddle plan`,
+`saddle query` with piped stdin, progress updates, a multiple-choice
+block answered through the web UI, an RFC block with a markdown body,
+the `saddle review` banner approved via the UI, a quoted user comment
+delivered over /notify, terminal SSE (replay + live), the bearer-token
+ingest API and the stm-markdown renderer. Host run:
 
 ```bash
 mkdir -p /tmp/clanker-e2e/workspace && cd /tmp/clanker-e2e/workspace \
@@ -169,7 +174,10 @@ DB_PATH=/tmp/clanker-e2e/web/clanker-saddle.sqlite \
 For a container run point `WEB`/`SADDLE` at the published ports, pass
 `SADDLE_HOST=clanker-saddle:7100`, the container's SQLite file as
 `DB_PATH`, and start the in-container mock via `MOCK_CMD='podman exec -d
-clanker-saddle python3 /usr/libexec/saddle/mock_llm.py …'`.
+clanker-saddle python3 /usr/libexec/saddle/mock_llm.py --port 9999'`
+(the test appends the same scripted responses to whatever command
+`MOCK_CMD` prefixes). The suite releases a daemon left blocked by an
+earlier run via `/stop` before it starts.
 
 ## Deployment (USM)
 

src/Api/SaddleApi.vala

@@ -155,6 +155,9 @@ namespace ClankerSaddle.Api {
                     && entry.get_member("exit_code").get_value_type() == typeof(int64)
                     ? entry.get_int_member("exit_code") : 0,
                 entry_type = get_string_or(entry, "type", "command"),
+                piped_input = get_string_or(entry, "input", null) != null
+                    && get_string_or(entry, "input", "").length > 0
+                    ? get_string_or(entry, "input", null) : null,
                 started_at = parse_iso(entry, "started_at"),
                 completed_at = parse_iso(entry, "completed_at"),
                 duration_ms = entry.has_member("duration_ms")

src/Api/TerminalSse.vala

@@ -73,6 +73,10 @@ namespace ClankerSaddle.Api {
             builder.add_string_value(entry.output_text ?? "");
             builder.set_member_name("type");
             builder.add_string_value(entry.entry_type);
+            if (entry.piped_input != null && ((!)entry.piped_input).length > 0) {
+                builder.set_member_name("input");
+                builder.add_string_value(((!)entry.piped_input));
+            }
             builder.end_object();
             var generator = new Json.Generator();
             generator.set_root(builder.get_root());

src/Data/Entities.vala

@@ -246,6 +246,7 @@ namespace ClankerSaddle.Data {
         public string? output_text { get; set; default = null; }
         public int64 exit_code { get; set; default = 0; }
         public string entry_type { get; set; default = "command"; }
+        public string? piped_input { get; set; default = null; }
         public DateTime? started_at { get; set; default = null; }
         public DateTime? completed_at { get; set; default = null; }
         public int64 duration_ms { get; set; default = 0; }
@@ -258,6 +259,7 @@ namespace ClankerSaddle.Data {
                 .column<string>("output_text", o => o.output_text, (o, v) => o.output_text = v)
                 .column<int64?>("exit_code", o => o.exit_code, (o, v) => o.exit_code = v)
                 .column<string>("type", o => o.entry_type, (o, v) => o.entry_type = v)
+                .column<string>("piped_input", o => o.piped_input, (o, v) => o.piped_input = v)
                 .column<DateTime?>("started_at", o => o.started_at, (o, v) => o.started_at = v)
                 .column<DateTime?>("completed_at", o => o.completed_at, (o, v) => o.completed_at = v)
                 .column<int64?>("duration_ms", o => o.duration_ms, (o, v) => o.duration_ms = v);

src/Data/Migrations/M0002_Part2.vala

@@ -47,6 +47,7 @@ namespace ClankerSaddle.Data.Migrations {
             b.alter_table("session_commands", (t) => {
                 t.add_column<string>("type")
                     .default_value<string>("command");
+                t.add_column<string>("piped_input");
             });
 
             b.create_table("api_tokens", (t) => {
@@ -67,6 +68,7 @@ namespace ClankerSaddle.Data.Migrations {
         public override void down(MigrationBuilder b) {
             b.drop_table("api_tokens");
             b.alter_table("session_commands", (t) => {
+                t.drop_column("piped_input");
                 t.drop_column("type");
             });
             b.alter_table("task_comments", (t) => {

src/Services/SaddleService.vala

@@ -357,7 +357,7 @@ namespace ClankerSaddle.Services {
                 payload.set_member_name("block_id");
                 payload.add_int_value(pending.block_id);
                 payload.set_member_name("answer");
-                payload.add_string_value(pending.answer);
+                payload.add_string_value(raw_answer_for_saddle(pending.answer));
                 payload.end_object();
             }
             payload.end_array();
@@ -446,10 +446,16 @@ namespace ClankerSaddle.Services {
                     yield update_saddle_status(saddle_id, "idle", null);
                     break;
                 case "working":
-                    var task = yield tasks.update_task(task_id, null, null, null, STATUS_WORKING);
-                    if (task != null && ((!)task).started_at == null) {
-                        ((!)task).started_at = new DateTime.now_local();
-                        yield db.update_async<TaskEntity>((!)task);
+                    // Never downgrade a pending review: the agent keeps
+                    // working after `saddle review` (it does not block),
+                    // but the banner must stay up until the human answers.
+                    var current = yield tasks.update_task(task_id, null, null, null, null);
+                    if (current != null && ((!)current).status != STATUS_REVIEW_REQUESTED) {
+                        current = yield tasks.update_task(task_id, null, null, null, STATUS_WORKING);
+                    }
+                    if (current != null && ((!)current).started_at == null) {
+                        ((!)current).started_at = new DateTime.now_local();
+                        yield db.update_async<TaskEntity>((!)current);
                     }
                     break;
                 case "idle":
@@ -646,7 +652,15 @@ namespace ClankerSaddle.Services {
 
             string state = status.has_member("state") ? status.get_string_member("state") : "";
             if (state == "blocked") {
-                yield deliver_answered_blocks((!)saddle, task.id, string_member(status, "block_question"));
+                yield deliver_answered_blocks((!)saddle, task.id, block_question_from_status(status));
+                yield tasks.update_task(task.id, null, null, null, STATUS_WORKING);
+                yield update_saddle_status(((!)saddle).id, "busy", task.id);
+            } else if (state == "working") {
+                // A non-blocking prompt was answered (e.g. `saddle review`,
+                // which reports without waiting): re-pushing /start would
+                // collide with the live session, so the verdict joins the
+                // conversation as a human comment via /notify.
+                yield deliver_answered_verdicts((!)saddle, task.id);
                 yield tasks.update_task(task.id, null, null, null, STATUS_WORKING);
                 yield update_saddle_status(((!)saddle).id, "busy", task.id);
             } else {
@@ -701,6 +715,7 @@ namespace ClankerSaddle.Services {
                         ? entry.get_int_member("exit_code") : 0,
                     entry_type = string_member(entry, "type").length > 0
                         ? string_member(entry, "type") : "command",
+                    piped_input = optional_string_member(entry, "input"),
                     started_at = parse_iso(entry, "started_at") ?? parse_iso(entry, "timestamp"),
                     completed_at = parse_iso(entry, "completed_at"),
                     duration_ms = entry.has_member("duration_ms") && entry.get_member("duration_ms").get_value_type() == typeof(int64)
@@ -709,6 +724,10 @@ namespace ClankerSaddle.Services {
                 yield db.insert_async<SessionCommand>(command);
                 stored.add(entry);
                 since = entry_id;
+                // Advance the persisted cursor per row: a failure mid-pass
+                // (e.g. a momentary SQLite lock) must not re-store the
+                // entries already inserted on the next tick.
+                poll_state.store_since(session_id, since);
             }
             poll_state.store_since(session_id, since);
             return stored;
@@ -782,10 +801,11 @@ namespace ClankerSaddle.Services {
          * whose answer has not been delivered to the daemon yet, means the
          * prompt is already represented. The status object carries the
          * question plus (from newer daemons) the block type, options and
-         * markdown body.
+         * markdown body. `saddle block rfc` carries no question text, so
+         * rfc prompts surface under a stable fallback title.
          */
         private async void ensure_block(TaskService tasks, int64 task_id, Json.Object status) throws Error {
-            string question = string_member(status, "block_question");
+            string question = block_question_from_status(status);
             if (question.length == 0) {
                 return;
             }
@@ -803,6 +823,21 @@ namespace ClankerSaddle.Services {
                 optional_string_member(status, "block_body"));
         }
 
+        /**
+         * The question a daemon block is recognised by: `block_question`
+         * verbatim, or the rfc fallback when the daemon raised an RFC (the
+         * `saddle block rfc` shape carries its document as the body and no
+         * question operand). Both the block-raising and answer-delivery
+         * paths key blocks by this string.
+         */
+        private static string block_question_from_status(Json.Object status) {
+            string question = string_member(status, "block_question");
+            if (question.length > 0) {
+                return question;
+            }
+            return string_member(status, "block_type") == "rfc" ? "Request for comments" : "";
+        }
+
         /**
          * Delivers human answers of the daemon's current question to its
          * `POST /answer` endpoint. Delivery is marked in
@@ -822,7 +857,7 @@ namespace ClankerSaddle.Services {
                 var payload = new Json.Builder();
                 payload.begin_object();
                 payload.set_member_name("answer");
-                payload.add_string_value(((!)block).answer ?? "");
+                payload.add_string_value(raw_answer_for_saddle(((!)block).answer ?? ""));
                 payload.end_object();
                 Json.Node? root = null;
                 int status_code = yield request_saddle(saddle.host, "POST", "/answer",
@@ -835,6 +870,42 @@ namespace ClankerSaddle.Services {
             }
         }
 
+        /**
+         * Delivers human answers of the task's answered-but-undelivered
+         * blocks to a *working* daemon as `/notify` comments: used when a
+         * non-blocking prompt (`saddle review`) was answered while the
+         * agent carried on, where `POST /answer` has no waiting block and
+         * a `/start` re-push would collide with the live session. Delivery
+         * is marked once the daemon accepts it (200/204).
+         */
+        private async void deliver_answered_verdicts(Saddle saddle, int64 task_id) throws Error {
+            foreach (var block in yield db.query<TaskBlock>()
+                    .where(expr("task_id == $0", elem<int64?>(task_id)))
+                    .order_by(expr("id"))
+                    .materialise_async()) {
+                if (block.answer == null || poll_state.is_delivered(block.id)) {
+                    continue;
+                }
+                var payload = new Json.Builder();
+                payload.begin_object();
+                payload.set_member_name("type");
+                payload.add_string_value("comment");
+                payload.set_member_name("text");
+                payload.add_string_value(((!)block).answer ?? "");
+                payload.set_member_name("quote");
+                payload.add_string_value(block.block_type == BLOCK_REVIEW ? "review" : "");
+                payload.end_object();
+                Json.Node? root = null;
+                int status_code = yield request_saddle(saddle.host, "POST", "/notify",
+                    (owned) payload, out root);
+                if (status_code != 200 && status_code != 204) {
+                    warning(@"[ClankerSaddle] Verdict notify to saddle $(saddle.id) returned HTTP $(status_code)");
+                    continue;
+                }
+                poll_state.mark_delivered(block.id);
+            }
+        }
+
         /**
          * Performs an HTTP request against a saddle daemon and parses the
          * response body as JSON; transport failures propagate to the
@@ -960,6 +1031,24 @@ namespace ClankerSaddle.Services {
             return trimmed.has_prefix(verb + " ") ? trimmed.substring(verb.length + 1).strip() : trimmed;
         }
 
+        /**
+         * The answer as the saddle daemon expects it over /answer (or in a
+         * resume payload): the web UI stores agent-facing text ("Human
+         * selected: …", "Human answered (other): …"), but the daemon
+         * applies its own BlockRequest formatting — delivering the
+         * prefixed string verbatim would wrap it a second time. RFC/plain
+         * replies are stored verbatim and pass through unchanged.
+         */
+        private static string raw_answer_for_saddle(string answer) {
+            if (answer.has_prefix("Human selected: ")) {
+                return answer.substring("Human selected: ".length);
+            }
+            if (answer.has_prefix("Human answered (other): ")) {
+                return answer.substring("Human answered (other): ".length);
+            }
+            return answer;
+        }
+
         /** Splits operands on whitespace honouring double quotes. */
         private static string[] split_operands(string args) {
             var operands = new Vector<string>();

src/pages/app.css

@@ -334,6 +334,17 @@ summary { cursor: pointer; font-weight: 600; }
     right: 0;
 }
 
+.term-entry .term-input {
+    background: #161b22;
+    border-left: 3px solid #30363d;
+    color: #8b949e;
+}
+
+.term-entry .term-input.collapsed { max-height: 7.5rem; overflow: hidden; }
+
+.term-entry.type-query { border-color: #a371f7; }
+.term-entry.type-query .term-command { color: #bc8cff; }
+
 .term-entry.type-comment .term-output, .term-entry.type-comment .term-command { color: #8b949e; font-style: italic; }
 .term-entry.type-block { border-color: #d29922; }
 .term-entry.type-block .term-output { color: #e3b341; }

src/pages/task_terminal.html

@@ -83,6 +83,17 @@
                     command.textContent = entry.command || '';
                     card.appendChild(command);
 
+                    if (entry.input && entry.input.length > 0) {
+                        var input = document.createElement('pre');
+                        input.className = 'term-input';
+                        input.textContent = entry.input;
+                        if (entry.input.split('\n').length > 5) input.classList.add('collapsed');
+                        input.addEventListener('click', function () {
+                            input.classList.toggle('collapsed');
+                        });
+                        card.appendChild(input);
+                    }
+
                     if (entry.output && entry.output.length > 0) {
                         var output = document.createElement('pre');
                         output.className = 'term-output';

test/integration_test.py

@@ -1,17 +1,23 @@
 #!/usr/bin/env python3
-"""End-to-end integration test: ClankerSaddle web app + saddle daemon.
+"""End-to-end integration test: ClankerSaddle web app + saddle daemon (Part 2).
 
 Drives the web app the way statum.js does (entrypoint state loads plus
 action POSTs carrying the auth slot and sealed private blobs) while a
 real saddle daemon works on a real git checkout under WORKSPACE/demo,
-powered by saddle/test/mock_llm.py with a deterministic three-task
-script: an explore task (bash + comment + done), a dependency-gated
-follow-up, and a blocking question answered through the web UI.
-
-A fourth section exercises the part 2 surface directly against the
-saddle ingest API: bearer-token auth, typed blocks with options, plan,
-progress and review updates, quoted comments on the poll endpoint, and
-the terminal SSE stream.
+powered by saddle/test/mock_llm.py with a deterministic script.
+
+Sections:
+1. explore task (bash + comment + done) and dependency gating (v1 loop)
+2. the Part 2 feature tour, driven by `saddle` commands the mock emits:
+   plan via heredoc, `saddle query` with piped stdin, progress updates,
+   a multiple-choice block (answered through the web UI), an RFC block
+   with a markdown body, a `saddle review` banner approved via the UI,
+   and a quoted user comment delivered to the daemon over /notify
+3. the ingest API surface: bearer-token auth (401/200), typed updates,
+   the poll endpoint's exactly-once handover
+4. SSE: terminal stream replay + live push (with piped query input)
+5. the stm-markdown renderer shipped in statum.js (via node)
+6. the audit trail in SQLite
 
 Prerequisites (see README.md "Integration testing"):
 - the web app running with SPRY_SEED_ADMIN_USERNAME=admin,
@@ -37,6 +43,7 @@ Prints PASS/FAIL lines Statum-style; exits non-zero on any failure.
 """
 import json
 import os
+import socket
 import sqlite3
 import subprocess
 import sys
@@ -56,6 +63,7 @@ MOCK = os.environ.get("MOCK", os.path.join(os.path.dirname(REPO), "saddle", "tes
 MOCK_PORT = os.environ.get("MOCK_PORT", "9999")
 MOCK_CMD = os.environ.get("MOCK_CMD", "")
 DEMO = os.path.join(WORKSPACE, "demo")
+HOSTNAME = socket.gethostname().split(".")[0]
 
 passes = 0
 failures = 0
@@ -181,32 +189,117 @@ def saddle_api(method, path, token, payload=None):
             return error.code, {"raw": body}
 
 
-def start_mock():
+def mock_argv():
+    heredoc = "cat << 'SADDLE_EOF' | %s\n%%s\nSADDLE_EOF"
     responses = [
         "# Inspecting the checkout\nls -la %s" % DEMO,
         "# Checking history\ngit -C %s log --oneline" % DEMO,
         'saddle comment "Inspected the checkout"\nsaddle done "Explored the demo repository"',
         "# Starting the follow-up\necho task-two-running",
         'saddle done "Follow-up audit complete"',
-        '# I need to ask\nsaddle block "Which branch should I target?"',
+        # Part 2 feature tour (task 3)
+        "# First, write the plan\n" + (heredoc % "saddle plan") % (
+            "# Feature tour plan\n\n1. Query the host\n2. Ask a question\n"
+            "3. Propose an RFC\n4. Request review"),
+        "# Confirm the working branch\ngit branch --show-current",
+        "# Side question about this host\nhostname | saddle query \"What host is this?\"",
+        "# Report progress\nsaddle progress-update queried the host",
+        "# Need a decision\n" + (heredoc % 'saddle block question "Which database should we use?" "SQLite" "Postgres" "Flat files"') % (
+            "# Storage choice\n\nWe need somewhere to keep the audit trail.\n\n"
+            "- **SQLite** — zero admin\n- **Postgres** — serious scale\n- **Flat files** — yolo"),
+        "# Proposing the module layout\n" + (heredoc % "saddle block rfc") % (
+            "# Proposed layout\n\n```text\nsrc/\n  api/\n  services/\n```\n\n"
+            "The `api` layer stays thin; services own the domain."),
+        "# Ready for review\n" + (heredoc % "saddle review") % (
+            "## Review: feature tour\n\nEverything on the tour is wired up.\n\n"
+            "- plans render\n- blocks answer\n- reviews approve"),
+        "# Waiting for any human input before finishing\nsleep 12",
     ]
-    argv = [sys.executable, MOCK, "--port", MOCK_PORT]
+    argv = []
     for response in responses:
         argv += ["-r", response]
-    argv += ["--on-answer", 'saddle done "Branch confirmed: main"', "--delay", "2"]
+    argv += [
+        "--on-answer", "# Human decision received\nsaddle progress-update human input integrated",
+        "--on-comment", 'saddle comment "Got the human comment"\nsaddle done "Part 2 feature tour complete"',
+        "--on-query", "The hostname is %s." % HOSTNAME,
+    ]
+    return argv
+
+
+def start_mock():
+    if MOCK_CMD:
+        # e.g. MOCK_CMD='podman exec -d clanker-saddle python3
+        # /usr/libexec/saddle/mock_llm.py' — the same scripted argv is
+        # appended, so the in-container mock behaves identically.
+        argv = MOCK_CMD.split() + mock_argv()
+    else:
+        argv = [sys.executable, MOCK, "--port", MOCK_PORT] + mock_argv()
     return subprocess.Popen(argv, stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT)
 
 
+def markdown_renderer_checks():
+    """Exercise the stm-markdown renderer shipped inside statum.js via node."""
+    status, body = http("GET", "/_statum/resource/statum.js", timeout=15)
+    if status != 200:
+        check(False, "statum.js served to the browser")
+        return
+    check("stm-markdown" in body, "statum.js ships the stm-markdown directive")
+
+    def extract(name, source):
+        start = source.index("function %s(" % name)
+        depth = 0
+        index = source.index("{", start)
+        end = index
+        while True:
+            if source[end] == "{":
+                depth += 1
+            elif source[end] == "}":
+                depth -= 1
+                if depth == 0:
+                    break
+            end += 1
+        return source[start:end + 1]
+
+    try:
+        parts = [extract(name, body)
+                 for name in ("safeMarkdownUrl", "mdLinks", "mdInline", "renderMarkdown")]
+    except ValueError:
+        check(False, "statum.js carries the markdown renderer functions")
+        return
+    harness = (
+        "%s\n%s\n%s\n%s\n"
+        "console.log(renderMarkdown('# Title\\n\\n- **bold** item\\n- `code` item\\n"
+        "\\n> quoted text\\n\\n```text\\nfenced\\n```'));\n"
+        "console.log('@@' + renderMarkdown('<script>alert(1)</script> [bad](javascript:alert(2))"
+        " [ok](https://example.com) <img src=x onerror=alert(3)>'));\n" % tuple(parts)
+    )
+    try:
+        result = subprocess.run(["node", "-e", harness], capture_output=True, text=True, timeout=20)
+    except (OSError, subprocess.SubprocessError):
+        check(True, "stm-markdown renderer present (node unavailable, deep check skipped)")
+        return
+    rendered, _, xss = result.stdout.partition("@@")
+    check("<h1>" in rendered and "<li>" in rendered and "<code>" in rendered
+          and "<blockquote>" in rendered and "<strong>" in rendered,
+          "stm-markdown renders headings, lists, code, quotes, bold")
+    check("<script" not in xss and "<img" not in xss
+          and 'href="javascript:' not in xss and 'href="#"' in xss
+          and '<a href="https://example.com">ok</a>' in xss,
+          "stm-markdown escapes raw tags, defangs javascript: links, keeps safe ones")
+
+
 def main():
-    mock = None
-    if not MOCK_CMD:
-        mock = start_mock()
-        time.sleep(0.5)
-    else:
-        subprocess.Popen(["bash", "-c", MOCK_CMD])
-        time.sleep(1.0)
+    mock = start_mock()
+    time.sleep(2.0)
 
     try:
+        # A previous suite run (or a crashed orchestrator) can leave the
+        # daemon parked on a block; /stop releases it so dispatch works.
+        try:
+            http("POST", SADDLE + "/stop", timeout=5)
+        except (urllib.error.URLError, OSError):
+            pass
+
         auth = login("admin", "adminpass123")
         check(True, "login yields an auth slot")
 
@@ -235,14 +328,19 @@ def main():
         })
         act(auth, state["create"], {
             "project_id": "1",
-            "title": "Ask a blocking question",
-            "description": "Asks which branch to target, then finishes.",
+            "title": "Part 2 feature tour",
+            "description": "Plan, query, block, RFC, review, comment — the works.",
             "priority": "5",
+            "task_branch": "task/3-feature-tour",
         })
         titles = [t["title"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]]
-        check(titles == ["Explore the demo repository", "Follow-up audit", "Ask a blocking question"],
+        check(titles == ["Explore the demo repository", "Follow-up audit", "Part 2 feature tour"],
               "three tasks queued in priority order")
 
+        tour = task_state(auth, 3)
+        check(tour["branch_from"] == "main", "branch_from prefilled from the project default branch")
+        check(tour["task_branch"] == "task/3-feature-tour", "task_branch stored at creation")
+
         detail = task_state(auth, 2)
         act(auth, detail["dependency"], {"depends_on_task_id": "1"})
         check(wait_for(lambda: any(int(d.get("id", 0)) == 1
@@ -252,9 +350,10 @@ def main():
 
         state = slot_public(auth, "/saddles", "saddles")
         act(auth, state["register"], {"name": SADDLE_NAME, "host": SADDLE_HOST})
-        check(any(s["name"] == SADDLE_NAME and s["host"] == SADDLE_HOST
-                  for s in slot_public(auth, "/saddles", "saddles")["saddles"]),
-              "saddle registered at %s" % SADDLE_HOST)
+        registered = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
+                      if s["name"] == SADDLE_NAME]
+        check(len(registered) == 1 and registered[0]["token"].startswith("saddle-"),
+              "saddle registered at %s with an API token" % SADDLE_HOST)
 
         print("== explore task ==")
         left_queue = wait_for(
@@ -276,26 +375,169 @@ def main():
               "agent comment projected into task_comments")
         check(detail.get("sessions") and detail["sessions"][0].get("outcome") == "done",
               "session closed with outcome done")
-        check(directives("/tasks/1/terminal", auth) is not None, "terminal page state loads")
 
         print("== dependency gating ==")
         check(wait_task_status(auth, 2, "done", timeout=90) is not None,
               "task 2 (dependent) dispatches and completes after task 1")
 
-        print("== block and answer ==")
-        check(wait_task_status(auth, 3, "blocked", timeout=90) is not None, "task 3 reaches blocked")
+        print("== part 2 feature tour ==")
+        check(wait_task_status(auth, 3, "working", timeout=30) is not None,
+              "task 3 dispatches to the saddle")
+
+        # plan via `saddle plan` heredoc
+        check(wait_for(lambda: "# Feature tour plan" in (task_state(auth, 3).get("plan") or "") or False,
+                       timeout=60, what="task 3 plan") is not None,
+              "saddle plan markdown lands in the task plan slot")
+        db = sqlite3.connect(DB_PATH)
+        stored_plan = db.execute("select plan from tasks where id = 3").fetchone()[0] or ""
+        check("1. Query the host" in stored_plan and "4. Request review" in stored_plan,
+              "plan stored in the tasks table verbatim")
+        db.close()
+
+        # `saddle query` with piped stdin
+        check(wait_for(lambda: any("query" == row[0]
+                                   for row in query_commands(3)) or False,
+                       timeout=60, what="query entry") is not None,
+              "saddle query logged as a typed terminal entry")
+
+        # the workspace clone sits on the task's branch
+        db = sqlite3.connect(DB_PATH)
+        branch_rows = db.execute(
+            "select sc.output_text from session_commands sc join sessions s on s.id = sc.session_id"
+            " where s.task_id = 3 and sc.command_text = 'git branch --show-current'"
+        ).fetchall()
+        db.close()
+        check(branch_rows and (branch_rows[0][0] or "").strip() == "task/3-feature-tour",
+              "workspace clone checks out the task branch")
+        db = sqlite3.connect(DB_PATH)
+        query_rows = db.execute(
+            "select sc.command_text, sc.output_text, sc.piped_input from session_commands sc"
+            " join sessions s on s.id = sc.session_id where s.task_id = 3 and sc.type = 'query'"
+        ).fetchall()
+        db.close()
+        check(len(query_rows) == 1 and "What host is this?" in query_rows[0][0],
+              "query entry carries the question as its command")
+        check(query_rows and HOSTNAME in (query_rows[0][1] or ""),
+              "query entry carries the side-LLM answer as output")
+        check(query_rows and (query_rows[0][2] or "").strip(),
+              "query entry stores the piped stdin in session_commands.piped_input")
+
+        # progress updates
+        check(wait_for(lambda: any("queried the host" == p["summary"]
+                                   for p in task_state(auth, 3).get("progress", [])) or False,
+                       timeout=60, what="first progress update") is not None,
+              "saddle progress-update appears on the task page")
         detail = task_state(auth, 3)
-        open_blocks = [b for b in detail.get("blocks", []) if not b.get("answered")]
-        check(len(open_blocks) == 1 and "Which branch" in open_blocks[0]["question"],
+        check(detail["progress"][0]["summary"] == "queried the host",
+              "latest progress update shown first")
+
+        # block question with options + markdown body
+        check(wait_task_status(auth, 3, "blocked", timeout=90) is not None,
+              "saddle block question flips the task to blocked")
+        detail = task_state(auth, 3)
+        open_q = [b for b in detail.get("blocks", [])
+                  if b["block_type"] == "question" and not b.get("answered")]
+        check(len(open_q) == 1 and "Which database" in open_q[0]["question"],
               "block question surfaced in the web UI")
-        act(auth, open_blocks[0]["answer_action"], {"answer": "main"})
-        check(any(b.get("answer") == "main" for b in task_state(auth, 3).get("blocks", [])),
-              "human answer recorded via the web action")
+        check(open_q and open_q[0]["options"] == ["SQLite", "Postgres", "Flat files"],
+              "block options stored on the block")
+        check(open_q and "# Storage choice" in open_q[0]["body"]
+              and "**SQLite**" in open_q[0]["body"],
+              "block markdown body stored on the block")
+        act(auth, open_q[0]["answer_action"], {"option": "Postgres"})
+        answered = [b for b in task_state(auth, 3)["blocks"]
+                    if b["block_type"] == "question"]
+        check(any(b.get("answer") == "Human selected: Postgres" for b in answered),
+              "option answer formatted for the agent")
+        # the resume window can be brief; the agent continuing past the
+        # answer is the durable signal (it raises the RFC next)
+        check(wait_for(lambda: any(b.get("block_type") == "rfc"
+                                   for b in task_state(auth, 3).get("blocks", [])) or False,
+                       timeout=90, what="agent continues after the answer") is not None,
+              "answered question unblocks the agent")
+
+        # RFC block with markdown body
+        check(wait_task_status(auth, 3, "blocked", timeout=90) is not None,
+              "saddle block rfc flips the task to blocked")
+        detail = task_state(auth, 3)
+        open_rfc = [b for b in detail.get("blocks", [])
+                    if b["block_type"] == "rfc" and not b.get("answered")]
+        check(len(open_rfc) == 1 and "# Proposed layout" in open_rfc[0]["body"],
+              "RFC markdown body stored on the block")
+        check(open_rfc and "```text" in open_rfc[0]["body"],
+              "RFC body keeps fenced code blocks intact")
+        act(auth, open_rfc[0]["answer_action"], {"answer": "Layout approved as proposed"})
+        check(any(b.get("answer") == "Layout approved as proposed"
+                  for b in task_state(auth, 3)["blocks"] if b["block_type"] == "rfc"),
+              "RFC reply recorded verbatim")
+        # the review banner following proves the agent resumed past the RFC
+        check(wait_for(lambda: task_state(auth, 3).get("review_pending") or False,
+                       timeout=90, what="agent resumes after the RFC") is not None,
+              "answered RFC unblocks the agent")
+
+        # review banner + approve
+        check(wait_task_status(auth, 3, "review_requested", timeout=90) is not None,
+              "saddle review flips the task to review_requested")
+        detail = task_state(auth, 3)
+        check(detail["review_pending"] and "Everything on the tour is wired up" in detail["review_summary"],
+              "review banner data on the task page")
+        check(detail["review_diff_url"].endswith(
+            "demo.git/compare/main...task/3-feature-tour"), "review diff link constructed")
+        dash = slot_public(auth, "/", "dashboard")
+        check(any(int(r["task_id"]) == 3 for r in dash["reviews"]),
+              "dashboard lists the pending review")
+        review_block = [b for b in detail["blocks"]
+                        if b["block_type"] == "review" and not b.get("answered")]
+        check(len(review_block) == 1, "review raised as an answerable block")
+        act(auth, review_block[0]["answer_action"], {"answer": "Human approved the review"})
+        check(any(b.get("answer") == "Human approved the review"
+                  for b in task_state(auth, 3)["blocks"] if b["block_type"] == "review"),
+              "review approval recorded")
+
+        # single formatting end to end: the daemon's ANSWER entry repeats the
+        # answer the way the agent read it
+        check(wait_task_status(auth, 3, "working", timeout=90) is not None,
+              "task 3 resumes working after the review approval")
+        db = sqlite3.connect(DB_PATH)
+        answer_rows = db.execute(
+            "select sc.command_text from session_commands sc join sessions s on s.id = sc.session_id"
+            " where s.task_id = 3 and sc.type = 'answer' order by sc.id").fetchall()
+        db.close()
+        check(any("Human selected: Postgres" == row[0] for row in answer_rows),
+              "daemon formats the option answer exactly once")
+
+        # quoted user comment while working (terminal highlight flow)
+        detail = task_state(auth, 3)
+        act(auth, detail["comment"], {
+            "body": "The sleep is fine, but flush sooner next time",
+            "quote": "sleep 12",
+            "return_to": "/tasks/3/terminal",
+        })
         check(wait_task_status(auth, 3, "done", timeout=90) is not None,
-              "answer delivered to daemon; task 3 completes")
+              "user comment delivered via /notify; task 3 completes")
         detail = task_state(auth, 3)
-        check("Branch confirmed: main" in (detail.get("summary") or ""),
-              "task 3 summary reflects the post-answer completion")
+        check("Part 2 feature tour complete" in (detail.get("summary") or ""),
+              "task 3 summary reflects the post-comment completion")
+        check(any(c["author"] == "user:admin" and c.get("quote") == "sleep 12"
+                  for c in detail.get("comments", [])),
+              "quoted user comment stored on the task")
+        check(any(c["author"] == "bot" and "Got the human comment" in c["body"]
+                  for c in detail.get("comments", [])),
+              "agent acknowledged the comment in-conversation")
+
+        db = sqlite3.connect(DB_PATH)
+        tour_types = {row[0] for row in db.execute(
+            "select sc.type from session_commands sc join sessions s on s.id = sc.session_id"
+            " where s.task_id = 3")}
+        notified = db.execute(
+            "select sc.output_text from session_commands sc join sessions s on s.id = sc.session_id"
+            " where s.task_id = 3 and sc.type = 'user-comment'").fetchall()
+        db.close()
+        check({"bash", "comment", "block", "answer", "plan", "progress", "review",
+               "query", "user-comment"} <= tour_types,
+              "terminal log carries every part 2 entry type (%s)" % sorted(tour_types))
+        check(any("sleep 12" in (row[0] or "") for row in notified),
+              "/notify delivered the comment with its quoted terminal text")
 
         print("== audit trail ==")
         db = sqlite3.connect(DB_PATH)
@@ -307,14 +549,64 @@ def main():
         check(any("git" in c and "Initial commit" in (r[1] or "") and r[2] == 0
                   for c, r in commands.items()),
               "git log output captured with exit code 0")
-        check(len([c for c in commands if "echo" in c]) >= 1, "echo output captured")
         per_session = db.execute(
             "select session_id, count(*) from session_commands group by session_id order by session_id"
         ).fetchall()
         check(all(count > 0 for _, count in per_session) and len(per_session) == 3,
               "each session carries its own commands (%s)" % per_session)
+        progress_rows = db.execute(
+            "select summary from task_progress where task_id = 3 order by id").fetchall()
+        check([row[0] for row in progress_rows] == ["queried the host", "human input integrated",
+                                                    "human input integrated"],
+              "task_progress keeps the full history in order")
         db.close()
 
+        print("== SSE ==")
+        db = sqlite3.connect(DB_PATH)
+        tour_session = db.execute(
+            "select s.id from sessions s where s.task_id = 3 order by s.id desc limit 1"
+        ).fetchone()[0]
+        db.close()
+        request = urllib.request.Request(WEB + "/terminal-stream/%d" % tour_session)
+        with urllib.request.urlopen(request, timeout=15) as stream:
+            check(stream.headers.get("Content-Type", "").startswith("text/event-stream"),
+                  "terminal stream is text/event-stream")
+            entries = []
+            saw_piped_input = False
+            deadline = time.monotonic() + 10
+            while time.monotonic() < deadline and len(entries) < 120:
+                if any(e.get("type") == "done" for e in entries):
+                    break
+                line = stream.readline().decode().strip()
+                if not line.startswith("data: ") or line == "data: ping":
+                    continue
+                entry = json.loads(line[6:])
+                entries.append(entry)
+                if entry.get("type") == "query" and (entry.get("input") or "").strip():
+                    saw_piped_input = True
+            check(len(entries) >= 12, "terminal stream replays the backlog (%d entries)" % len(entries))
+            check(saw_piped_input, "terminal stream carries the query's piped input")
+            check(any(e.get("type") == "review" for e in entries)
+                  and any(e.get("type") == "user-comment" for e in entries),
+                  "terminal stream carries typed part 2 events")
+            db = sqlite3.connect(DB_PATH)
+            db.execute("insert into session_commands (session_id, command_text, output_text,"
+                       " type, started_at) values (?, 'sse-live-check', 'ok', 'command',"
+                       " strftime('%s','now'))", (tour_session,))
+            db.commit()
+            db.close()
+            got_live = False
+            deadline = time.monotonic() + 6
+            while time.monotonic() < deadline and not got_live:
+                line = stream.readline().decode().strip()
+                if line.startswith("data: ") and "sse-live-check" in line:
+                    got_live = True
+            check(got_live, "terminal stream pushes new entries within seconds")
+
+        pulse = directives("/tasks/3", auth)
+        check(any(d.get("type") == "subscribe" for d in pulse),
+              "task page subscribes to the pulse channel for live refresh")
+
         print("== part 2 API surface ==")
         db = sqlite3.connect(DB_PATH)
 
@@ -373,6 +665,13 @@ def main():
                                       "options": ["SQLite", "Postgres"],
                                       "body": "# Pick one"}})
         check(status == 200 and body.get("block_id"), "typed block returns block_id")
+        status, _ = saddle_api("POST", "/api/saddle/%d/update" % api_saddle, api_token,
+                               {"type": "output", "session_id": api_session, "data": {
+                                   "command": "cat notes.md | saddle query 'meaning?'",
+                                   "output": "42",
+                                   "type": "query",
+                                   "input": "# notes\nmeaning of life"}})
+        check(status == 200, "query output update accepted")
         status, _ = saddle_api("POST", "/api/saddle/%d/update" % api_saddle, api_token,
                                {"type": "plan", "session_id": api_session,
                                 "data": {"plan": "# Plan\n- migrate schemas"}})
@@ -405,10 +704,15 @@ def main():
         check(any(b["question"] == "Which database?" for b in dash["open_blocks"]),
               "dashboard lists blocks needing answers")
 
+        piped = db.execute("select piped_input from session_commands where session_id = ?"
+                           " and type = 'query'", (api_session,)).fetchall()
+        check(piped and "meaning of life" in (piped[0][0] or ""),
+              "push API stores the query's piped input too")
+
         question = [b for b in detail["blocks"] if b["block_type"] == "question"][0]
         act(auth, question["answer_action"], {"option": "SQLite"})
         check([b["answer"] for b in task_state(auth, api_task)["blocks"]
-              if b["block_type"] == "question"] == ["Human selected: SQLite"],
+               if b["block_type"] == "question"] == ["Human selected: SQLite"],
               "option answer formatted for the agent")
         act(auth, detail["comment"],
             {"body": "Check the migration output", "quote": "ERROR: table demos exists"})
@@ -427,35 +731,12 @@ def main():
         check({"comment", "block", "plan", "progress", "review", "user-comment"} <= entry_types,
               "conversation events logged as typed terminal entries")
 
-        request = urllib.request.Request(WEB + "/terminal-stream/%d" % api_session)
-        with urllib.request.urlopen(request, timeout=10) as stream:
-            check(stream.headers.get("Content-Type", "").startswith("text/event-stream"),
-                  "terminal stream is text/event-stream")
-            lines = []
-            deadline = time.monotonic() + 5
-            while time.monotonic() < deadline and len(lines) < 40:
-                line = stream.readline().decode().strip()
-                if line:
-                    lines.append(line)
-                    deadline = min(deadline, time.monotonic() + 2)
-            entries = [json.loads(line[6:]) for line in lines
-                       if line.startswith("data: ") and line != "data: ping"]
-            check(len(entries) >= 7, "terminal stream replays the backlog")
-            db.execute("insert into session_commands (session_id, command_text, output_text,"
-                       " type, started_at) values (?, 'sse-live-check', 'ok', 'command',"
-                       " strftime('%s','now'))", (api_session,))
-            db.commit()
-            got_live = False
-            deadline = time.monotonic() + 6
-            while time.monotonic() < deadline and not got_live:
-                line = stream.readline().decode().strip()
-                if line.startswith("data: ") and "sse-live-check" in line:
-                    got_live = True
-            check(got_live, "terminal stream pushes new entries within seconds")
-
         act(auth, detail["cancel"], {"task_id": str(api_task)})
         db.close()
 
+        print("== stm-markdown ==")
+        markdown_renderer_checks()
+
         print("== quiesce ==")
         check(wait_for(
             lambda: all(int(s["current_task_id"] or 0) == 0 and s["status"] == "idle"
@@ -470,5 +751,15 @@ def main():
     return 1 if failures else 0
 
 
+def query_commands(task_id):
+    db = sqlite3.connect(DB_PATH)
+    try:
+        return db.execute(
+            "select sc.type from session_commands sc join sessions s on s.id = sc.session_id"
+            " where s.task_id = ? and sc.type = 'query'", (task_id,)).fetchall()
+    finally:
+        db.close()
+
+
 if __name__ == "__main__":
     sys.exit(main())