feat(bot): Phase 2 — streaming via sendMessageDraft + /stop (Issue #55) (#58)

gHashTag · Antigravity Agent · claude · web-flow · commit 8a77d3252292 · 2026-03-09T21:49:21.000+07:00
Two-thread architecture: main thread polls Telegram, worker thread streams Claude output. NDJSON parsing from --output-format stream-json, sendDraft every 500ms, /stop sends SIGTERM to active process. New: claude_stream.zig (StreamState + runStreaming + stopProcess). Modified: bot_loop.zig (thread dispatch), handlers.zig (help text), telegram_api.zig (shared sendLongMessage). Closes #55 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Antigravity Agent <antigravity@vibee.org> Co-authored-by: Claude <noreply@anthropic.com>
diff --git a/tools/mcp/trinity_mcp/bot/bot_loop.zig b/tools/mcp/trinity_mcp/bot/bot_loop.zig
@@ -1,21 +1,27 @@
 // bot_loop.zig — Main poll → parse → dispatch → repeat loop
+// Phase 2: Two-thread arch — main thread polls, worker thread streams Claude output
 const std = @import("std");
 const telegram_api = @import("telegram_api.zig");
 const json_utils = @import("json_utils.zig");
 const command_parser = @import("command_parser.zig");
 const handlers = @import("handlers.zig");
+const claude_stream = @import("claude_stream.zig");
 
 const BotConfig = telegram_api.BotConfig;
 
+/// Shared state for streaming — module-level, accessible from main + worker threads
+var stream_state = claude_stream.StreamState{};
+
 /// Run the bot loop: poll Telegram, parse commands, dispatch handlers.
-/// Never returns (infinite loop).
+/// Never returns (infinite loop). Main thread stays responsive while
+/// worker thread handles Claude streaming.
 pub fn run(allocator: std.mem.Allocator, config: BotConfig) void {
     var last_update_id: i64 = 0;
 
     // Announce startup
-    telegram_api.sendMessage(allocator, config.bot_token, config.chat_id, "\xf0\x9f\xa4\x96 TRI BOT online! Send /help for commands.");
+    telegram_api.sendMessage(allocator, config.bot_token, config.chat_id, "\xf0\x9f\xa4\x96 TRI BOT v2.0 online! Send /help for commands.");
 
-    std.debug.print("[tri-bot] Started. Polling Telegram...\n", .{});
+    std.debug.print("[tri-bot] Started (Phase 2: streaming). Polling Telegram...\n", .{});
 
     while (true) {
         const body = telegram_api.getUpdates(allocator, config.bot_token, last_update_id + 1) orelse {
@@ -25,48 +31,31 @@ pub fn run(allocator: std.mem.Allocator, config: BotConfig) void {
         };
         defer allocator.free(body);
 
-        // Process each update
-        const Context = struct {
-            allocator: std.mem.Allocator,
-            config: BotConfig,
-            max_id: i64,
-        };
-        var ctx = Context{
-            .allocator = allocator,
-            .config = config,
-            .max_id = last_update_id,
-        };
-
-        // We can't use closures in Zig, so use a global-style dispatch.
-        // Instead, manually iterate updates with a simple loop.
-        processUpdates(allocator, config, body, &ctx.max_id);
-        last_update_id = ctx.max_id;
+        var max_id = last_update_id;
+        processUpdates(allocator, config, body, &max_id);
+        last_update_id = max_id;
     }
 }
 
 fn processUpdates(allocator: std.mem.Allocator, config: BotConfig, body: []const u8, max_id: *i64) void {
-    // Find each "update_id": block manually
     var pos: usize = 0;
     while (pos < body.len) {
         const needle = "\"update_id\":";
         const idx = std.mem.indexOfPos(u8, body, pos, needle) orelse break;
 
-        // Determine block boundary (next update_id or end)
         const next_idx = std.mem.indexOfPos(u8, body, idx + needle.len + 1, needle) orelse body.len;
         const block = body[idx..next_idx];
 
-        // Extract update_id
         const uid = json_utils.extractInt(block, "update_id") orelse {
             pos = idx + needle.len;
             continue;
         };
 
-        // Update max
         if (uid > max_id.*) {
             max_id.* = uid;
         }
 
-        // Extract chat_id
+        // Extract chat_id from nested message.chat.id
         const chat_id_val = blk: {
             const chat_needle = "\"chat\":{\"id\":";
             const ci = std.mem.indexOf(u8, block, chat_needle) orelse break :blk @as(i64, 0);
@@ -76,26 +65,22 @@ fn processUpdates(allocator: std.mem.Allocator, config: BotConfig, body: []const
             break :blk std.fmt.parseInt(i64, block[cs..ce], 10) catch 0;
         };
 
-        // Auth check: only respond to configured chat_id
+        // Auth check
         const expected_chat_id = std.fmt.parseInt(i64, config.chat_id, 10) catch 0;
         if (chat_id_val != expected_chat_id) {
-            std.debug.print("[tri-bot] Ignoring update from chat {d} (expected {d})\n", .{ chat_id_val, expected_chat_id });
+            std.debug.print("[tri-bot] Ignoring update from chat {d}\n", .{chat_id_val});
             pos = next_idx;
             continue;
         }
 
-        // Extract text
         const text = json_utils.extractString(block, "text") orelse {
             pos = next_idx;
             continue;
         };
 
         std.debug.print("[tri-bot] Update {d}: \"{s}\"\n", .{ uid, text });
 
-        // Parse command
         const cmd = command_parser.parse(text);
-
-        // Dispatch
         dispatch(allocator, config, cmd);
 
         pos = next_idx;
@@ -106,15 +91,45 @@ fn dispatch(allocator: std.mem.Allocator, config: BotConfig, cmd: command_parser
     if (std.mem.eql(u8, cmd.name, "help")) {
         handlers.handleHelp(allocator, config);
     } else if (std.mem.eql(u8, cmd.name, "ask")) {
-        handlers.handleAsk(allocator, config, cmd.args);
+        // Phase 2: streaming /ask via worker thread
+        if (cmd.args.len == 0) {
+            telegram_api.sendMessage(allocator, config.bot_token, config.chat_id, "\xe2\x9a\xa0 Usage: /ask <question>");
+            return;
+        }
+        if (stream_state.is_busy.load(.acquire)) {
+            telegram_api.sendMessage(allocator, config.bot_token, config.chat_id, "\xe2\x8f\xb3 Already processing. Send /stop first.");
+            return;
+        }
+        // Dupe args — they point into getUpdates body which will be freed
+        const args_owned = allocator.dupe(u8, cmd.args) catch return;
+        stream_state.is_busy.store(true, .release);
+        _ = std.Thread.spawn(.{}, claude_stream.runStreaming, .{ allocator, config, args_owned, false, &stream_state }) catch {
+            stream_state.is_busy.store(false, .release);
+            allocator.free(args_owned);
+            telegram_api.sendMessage(allocator, config.bot_token, config.chat_id, "\xe2\x9d\x8c Failed to spawn worker thread");
+            return;
+        };
     } else if (std.mem.eql(u8, cmd.name, "continue")) {
-        handlers.handleContinue(allocator, config, cmd.args);
+        // Phase 2: streaming /continue via worker thread
+        if (stream_state.is_busy.load(.acquire)) {
+            telegram_api.sendMessage(allocator, config.bot_token, config.chat_id, "\xe2\x8f\xb3 Already processing. Send /stop first.");
+            return;
+        }
+        const args_owned = allocator.dupe(u8, cmd.args) catch return;
+        stream_state.is_busy.store(true, .release);
+        _ = std.Thread.spawn(.{}, claude_stream.runStreaming, .{ allocator, config, args_owned, true, &stream_state }) catch {
+            stream_state.is_busy.store(false, .release);
+            allocator.free(args_owned);
+            telegram_api.sendMessage(allocator, config.bot_token, config.chat_id, "\xe2\x9d\x8c Failed to spawn worker thread");
+            return;
+        };
     } else if (std.mem.eql(u8, cmd.name, "status")) {
+        // Status stays blocking (short query, no streaming needed)
         handlers.handleStatus(allocator, config);
     } else if (std.mem.eql(u8, cmd.name, "stop")) {
-        telegram_api.sendMessage(allocator, config.bot_token, config.chat_id, "\xe2\x9b\x94 /stop not yet implemented (Phase 2)");
+        // Phase 2: /stop kills active Claude process
+        claude_stream.stopProcess(allocator, config, &stream_state);
     } else if (cmd.name.len > 0) {
-        // Unknown command
         var buf: [256]u8 = undefined;
         telegram_api.sendFmt(allocator, config.bot_token, config.chat_id, &buf, "\xe2\x9d\x93 Unknown command: /{s}. Try /help", .{cmd.name});
     }
diff --git a/tools/mcp/trinity_mcp/bot/claude_stream.zig b/tools/mcp/trinity_mcp/bot/claude_stream.zig
@@ -0,0 +1,153 @@
+// claude_stream.zig — Streaming Claude CLI execution with sendMessageDraft
+// Spawns claude with --output-format stream-json, pipes stdout,
+// reads NDJSON line-by-line, sends drafts every 500ms, final message when done.
+const std = @import("std");
+const telegram_api = @import("telegram_api.zig");
+const json_utils = @import("json_utils.zig");
+
+const BotConfig = telegram_api.BotConfig;
+
+/// Shared state between main thread (polling) and worker thread (streaming).
+/// Atomics for lock-free is_busy check, PID for /stop.
+pub const StreamState = struct {
+    is_busy: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
+    active_pid: std.atomic.Value(i32) = std.atomic.Value(i32).init(0),
+};
+
+/// Worker thread entry point: spawn claude, stream output, send drafts.
+/// Called via std.Thread.spawn() from bot_loop dispatch.
+pub fn runStreaming(
+    allocator: std.mem.Allocator,
+    config: BotConfig,
+    args_owned: []const u8,
+    use_continue: bool,
+    state: *StreamState,
+) void {
+    defer {
+        state.active_pid.store(0, .release);
+        state.is_busy.store(false, .release);
+        allocator.free(args_owned);
+    }
+
+    // Notify user
+    if (use_continue) {
+        telegram_api.sendMessage(allocator, config.bot_token, config.chat_id, "\xf0\x9f\x94\x84 TRI streaming...");
+    } else {
+        telegram_api.sendMessage(allocator, config.bot_token, config.chat_id, "\xf0\x9f\xa7\xa0 TRI streaming...");
+    }
+
+    // Build turns string
+    var turns_buf: [16]u8 = undefined;
+    const turns_str = std.fmt.bufPrint(&turns_buf, "{d}", .{config.max_turns}) catch "10";
+
+    // Build argv dynamically based on args and --continue flag
+    var argv_buf: [12][]const u8 = undefined;
+    var argc: usize = 0;
+
+    argv_buf[argc] = "claude";
+    argc += 1;
+    if (args_owned.len > 0) {
+        argv_buf[argc] = "-p";
+        argc += 1;
+        argv_buf[argc] = args_owned;
+        argc += 1;
+    }
+    if (use_continue) {
+        argv_buf[argc] = "--continue";
+        argc += 1;
+    }
+    argv_buf[argc] = "--output-format";
+    argc += 1;
+    argv_buf[argc] = "stream-json";
+    argc += 1;
+    argv_buf[argc] = "--max-turns";
+    argc += 1;
+    argv_buf[argc] = turns_str;
+    argc += 1;
+
+    // Init child with piped stdout
+    var child = std.process.Child.init(argv_buf[0..argc], allocator);
+    child.stdout_behavior = .Pipe;
+    child.stderr_behavior = .Pipe;
+    child.cwd = config.project_root;
+
+    child.spawn() catch |err| {
+        var err_buf: [256]u8 = undefined;
+        telegram_api.sendFmt(allocator, config.bot_token, config.chat_id, &err_buf, "\xe2\x9d\x8c Spawn error: {s}", .{@errorName(err)});
+        return;
+    };
+
+    // Store PID for /stop
+    state.active_pid.store(@intCast(child.id), .release);
+
+    std.debug.print("[tri-bot] Streaming started (PID {d})\n", .{child.id});
+
+    // Read stdout line-by-line, accumulate text deltas, send drafts
+    var text_buf: std.ArrayList(u8) = .empty;
+    defer text_buf.deinit(allocator);
+
+    var last_draft_ns: i128 = std.time.nanoTimestamp();
+    const draft_interval: i128 = 500_000_000; // 500ms
+
+    const stdout_file = child.stdout.?;
+    var line_buf: [65536]u8 = undefined;
+    var line_len: usize = 0;
+    var read_chunk: [4096]u8 = undefined;
+
+    while (true) {
+        const n = stdout_file.read(&read_chunk) catch break;
+        if (n == 0) break; // EOF
+
+        for (read_chunk[0..n]) |byte| {
+            if (byte == '\n') {
+                // Process complete NDJSON line
+                const line = line_buf[0..line_len];
+                if (std.mem.indexOf(u8, line, "\"text_delta\"") != null) {
+                    if (json_utils.extractString(line, "text")) |text_val| {
+                        text_buf.appendSlice(allocator, text_val) catch {};
+                    }
+                }
+                line_len = 0;
+
+                // Throttle: sendDraft every 500ms
+                const now = std.time.nanoTimestamp();
+                if (now - last_draft_ns >= draft_interval and text_buf.items.len > 0) {
+                    const draft_len = @min(text_buf.items.len, 4000);
+                    telegram_api.sendDraft(allocator, config.bot_token, config.chat_id, text_buf.items[0..draft_len]);
+                    last_draft_ns = now;
+                }
+            } else if (line_len < line_buf.len - 1) {
+                line_buf[line_len] = byte;
+                line_len += 1;
+            }
+        }
+    }
+
+    // Wait for child to finish
+    _ = child.wait() catch {};
+
+    std.debug.print("[tri-bot] Streaming done ({d} bytes)\n", .{text_buf.items.len});
+
+    // Send final message
+    if (text_buf.items.len > 0) {
+        telegram_api.sendLongMessage(allocator, config, text_buf.items);
+    } else {
+        telegram_api.sendMessage(allocator, config.bot_token, config.chat_id, "\xe2\x9a\xa0 Claude returned empty response");
+    }
+}
+
+/// Stop the active Claude process via SIGTERM.
+pub fn stopProcess(allocator: std.mem.Allocator, config: BotConfig, state: *StreamState) void {
+    const pid = state.active_pid.load(.acquire);
+    if (pid > 0) {
+        const posix_pid: std.posix.pid_t = @intCast(pid);
+        std.posix.kill(posix_pid, 15) catch |err| { // 15 = SIGTERM
+            var err_buf: [256]u8 = undefined;
+            telegram_api.sendFmt(allocator, config.bot_token, config.chat_id, &err_buf, "\xe2\x9d\x8c Kill error: {s}", .{@errorName(err)});
+            return;
+        };
+        telegram_api.sendMessage(allocator, config.bot_token, config.chat_id, "\xe2\x9b\x94 Claude process stopped");
+    } else {
+        telegram_api.sendMessage(allocator, config.bot_token, config.chat_id, "\xe2\x9a\xa0 No active Claude process");
+    }
+}
diff --git a/tools/mcp/trinity_mcp/bot/handlers.zig b/tools/mcp/trinity_mcp/bot/handlers.zig
@@ -10,13 +10,13 @@ pub fn handleHelp(allocator: std.mem.Allocator, config: BotConfig) void {
     const help_text =
         "\xf0\x9f\xa4\x96 TRI BOT \xe2\x80\x94 Claude Code Remote Control\n" ++
         "\n" ++
-        "/ask <question> \xe2\x80\x94 Ask Claude anything\n" ++
-        "/continue <question> \xe2\x80\x94 Continue last session\n" ++
+        "/ask <question> \xe2\x80\x94 Ask Claude (streaming)\n" ++
+        "/continue [question] \xe2\x80\x94 Continue session (streaming)\n" ++
         "/status \xe2\x80\x94 Project status\n" ++
         "/stop \xe2\x80\x94 Stop running Claude process\n" ++
         "/help \xe2\x80\x94 This message\n" ++
         "\n" ++
-        "Phase 2: /resume, /model, /board, /pr, /worktree";
+        "Phase 3: /resume, /model, /board, /pr, /worktree";
     telegram_api.sendMessage(allocator, config.bot_token, config.chat_id, help_text);
 }
 
@@ -54,7 +54,7 @@ pub fn handleAsk(allocator: std.mem.Allocator, config: BotConfig, args: []const
     }
 
     // Telegram message limit: 4096 chars. Split if needed.
-    sendLongMessage(allocator, config, result.stdout);
+    telegram_api.sendLongMessage(allocator, config, result.stdout);
 }
 
 /// /continue <question> — Run claude -p "<question>" --continue
@@ -87,7 +87,7 @@ pub fn handleContinue(allocator: std.mem.Allocator, config: BotConfig, args: []c
         return;
     }
 
-    sendLongMessage(allocator, config, result.stdout);
+    telegram_api.sendLongMessage(allocator, config, result.stdout);
 }
 
 /// /status — Run claude -p "Summarize project status in 5 lines"
@@ -117,29 +117,5 @@ pub fn handleStatus(allocator: std.mem.Allocator, config: BotConfig) void {
         return;
     }
 
-    sendLongMessage(allocator, config, result.stdout);
-}
-
-/// Send a potentially long message, splitting at 4096 char boundary
-fn sendLongMessage(allocator: std.mem.Allocator, config: BotConfig, text: []const u8) void {
-    const max_len: usize = 4000; // Leave some margin below 4096
-    var pos: usize = 0;
-
-    while (pos < text.len) {
-        const end = @min(pos + max_len, text.len);
-        // Try to split at a newline
-        var split = end;
-        if (end < text.len) {
-            var j = end;
-            while (j > pos + max_len / 2) : (j -= 1) {
-                if (text[j] == '\n') {
-                    split = j + 1;
-                    break;
-                }
-            }
-        }
-        const chunk = text[pos..split];
-        telegram_api.sendMessage(allocator, config.bot_token, config.chat_id, chunk);
-        pos = split;
-    }
+    telegram_api.sendLongMessage(allocator, config, result.stdout);
 }
diff --git a/tools/mcp/trinity_mcp/bot/telegram_api.zig b/tools/mcp/trinity_mcp/bot/telegram_api.zig
