diff --git a/GraphcodeKit/Sources/Sessions/ZmxSessionLauncher.swift b/GraphcodeKit/Sources/Sessions/ZmxSessionLauncher.swift index 7189ca86..be8ae216 100644 --- a/GraphcodeKit/Sources/Sessions/ZmxSessionLauncher.swift +++ b/GraphcodeKit/Sources/Sessions/ZmxSessionLauncher.swift @@ -33,18 +33,63 @@ public enum ZmxSessionLauncher { /// `.message` edge rides on (docs/02-graph-of-loops.md#inter-loop-messaging-in-practice). /// Returns false when there's no live session to deliver into, so the caller can /// report an undelivered message rather than assume it landed. + /// + /// One write for the whole message, which is why nothing on the delivery path calls + /// this: `sendWrites` is what `send` types, and a message long enough to need framing + /// gets none here (issue #277). static func sendArguments(_ text: String, toNode node: LoopNode) -> [String] { ["send", SurfaceRef(id: node.id, launchesClaudeCode: true).zmxSessionName, flattened(text)] } - /// The most a single `zmx send` may carry. One send is one uninterrupted write into - /// the session's PTY, and a PTY's kernel input queue holds 4 KB: measured on macOS - /// 26, a 3.7 KB message typed as one write arrives intact and a 7.3 KB one vanishes - /// *entirely* — while `zmx send` still exits 0, so every layer above reported the - /// message delivered. Half the queue leaves room for whatever the session's TUI has - /// not yet drained when the write lands. + /// The most a single `zmx send` may carry — a bound on the *write*, which is all it + /// ever was. One send is one uninterrupted write into the session's PTY, and the PTY + /// itself honours this size: measured on macOS 26, 2 KB writes reach a raw-mode reader + /// byte-for-byte even when that reader stalls 300 ms between reads, because the zmx + /// daemon buffers what the kernel will not yet take and drains it on `POLLOUT`. + /// + /// It is **not** what keeps a message intact. That was the theory this constant was + /// introduced with (0.1.29, `d3f7c9a`) and issue #277 disproved it: splitting at this + /// size still lost the head of every multi-chunk message. See + /// `maxUnbracketedSendBytes` for what the real limit turned out to be. static let maxSendChunkBytes = 2048 + /// The most a message may be before it is typed as a **paste** rather than as + /// keystrokes — the fix for issue #277, and the one number here that governs whether a + /// message survives at all. + /// + /// What #277 turned out to be: nothing below the agent's own TUI loses anything. The + /// controls are unambiguous — 2 KB writes reach a raw-mode sink intact, and reach a + /// sink that stalls 300 ms per read intact too, so the kernel, the PTY and the zmx + /// daemon are all innocent. The loss is the **composer's** own input handling, which + /// swallows a large enough burst of keystrokes whole (2634 bytes sent, 590 received; + /// the `[graphcode] :` prefix went with the head, which is why grepping a + /// transcript for `[graphcode]` found only the *undamaged* messages). + /// + /// And it cannot be fixed by choosing a smaller chunk, which is the trap: the TUI's + /// **reads** are not our writes. With a reader stalled 0.5 s, two 512-byte writes + /// arrived as one 1022-byte read — and an agent mid-turn stalls for far longer than + /// `interChunkDelay`. Any chunk size can coalesce into a burst above the threshold, + /// which is why the observed losses (1954, 1096, 2539 bytes) never landed on a chunk + /// boundary. + /// + /// Bracketed paste is the mechanism that does not care: `ESC[200~ … ESC[201~` tells + /// the TUI where the payload begins and ends, so coalescing is harmless and no + /// keystroke heuristic applies. Verified intact at 2600 bytes as one write, and as + /// 512-byte writes inside the brackets. Every backend graphcode launches enables the + /// mode (`DECSET 2004`) at startup. + /// + /// Why a threshold at all, rather than pasting everything: a message this small is + /// *measured* to arrive intact as plain keystrokes and is the overwhelming majority of + /// what loops send. Leaving it on the path it already survives keeps the fix confined + /// to the case that is broken today, where any change can only be an improvement. + static let maxUnbracketedSendBytes = 1024 + + /// The start and end of a bracketed paste (`DECSET 2004`) — what a terminal emits + /// around text a human pastes, and what makes a long message one payload to the TUI + /// instead of a burst of keystrokes it is free to drop. + static let pasteStart = "\u{1B}[200~" + static let pasteEnd = "\u{1B}[201~" + /// The beat between chunks — what gives the session's TUI time to drain one write /// out of the PTY queue before the next lands on top of it. static let interChunkDelay: Duration = .milliseconds(150) @@ -72,6 +117,50 @@ public enum ZmxSessionLauncher { return chunks } + /// The exact sequence of `zmx send` payloads that carries one message, in order. + /// + /// Short messages are unchanged: one write, plain keystrokes, exactly what shipped + /// before issue #277 and exactly what was measured to arrive intact. A message over + /// `maxUnbracketedSendBytes` is wrapped in a bracketed paste and carries a trailer — + /// see those two for why each is there. + /// + /// The brackets ride on the first and last chunk rather than travelling as writes of + /// their own, so that a start marker can never be separated from the payload it opens + /// by a failure between two sends. + static func sendWrites(_ text: String, limit: Int = maxSendChunkBytes) -> [String] { + let flat = flattened(text) + guard flat.utf8.count > maxUnbracketedSendBytes else { + return flat.isEmpty ? [] : [flat] + } + var chunks = messageChunks(flat + deliveryTrailer(for: flat), limit: limit) + chunks[0] = pasteStart + chunks[0] + chunks[chunks.count - 1] += pasteEnd + return chunks + } + + /// What a long message says about itself, so that losing its head stays *reportable*. + /// + /// `delivered` has only ever meant "every `zmx send` exited 0", and issue #277 is the + /// second time that has been true of a message whose text never arrived. Nothing this + /// side can read back proves otherwise — the composer is not legible through `zmx`, + /// and a message to a busy loop sits there unsubmitted for as long as the turn lasts — + /// so the receipt is one the *receiver* can check: a clipped message now arrives + /// carrying the length and the opening it should have had. + /// + /// It rides the tail because the tail is the half that survives. That also answers + /// #277's third suggestion without moving the `[graphcode] :` prefix off the + /// head where it reads naturally: the marker repeats the `[graphcode]` token, so the + /// obvious audit — grepping a transcript for it — stops being blind to exactly the + /// messages it should be finding. + static func deliveryTrailer(for flattenedText: String) -> String { + let opening = String(flattenedText.prefix(deliveryTrailerOpeningCharacters)) + return " [graphcode] end of a \(flattenedText.count)-character message opening " + + "\"\(opening)…\" — if it did not reach you whole, say so and ask for a resend." + } + + /// Enough of the opening to recognise, short enough not to bury the message it guards. + static let deliveryTrailerOpeningCharacters = 48 + /// Wraps a backend command in an **interactive login** shell, so it resolves the same /// way it would if a human typed it into a terminal. /// @@ -349,12 +438,13 @@ public enum ZmxSessionLauncher { } guard ZmxLocator.isInstalled, !text.isEmpty else { return false } guard await sessionExists(node) else { return false } - // Typed in PTY-queue-sized pieces rather than one write — see `maxSendChunkBytes` - // for what a single oversized write silently does. The pieces just accumulate in - // the composer, exactly as the text and the later `\r` already do; nothing is - // submitted until the one Enter below. + // Typed as the writes `sendWrites` frames it into — plain keystrokes when it is + // short, a bracketed paste when it is not, which is what keeps a long message's head + // from being swallowed by the composer (`maxUnbracketedSendBytes`, issue #277). The + // pieces just accumulate in the composer, exactly as the text and the later `\r` + // already do; nothing is submitted until the one Enter below. let sessionName = SurfaceRef(id: node.id, launchesClaudeCode: true).zmxSessionName - for (index, chunk) in messageChunks(text).enumerated() { + for (index, chunk) in sendWrites(text).enumerated() { if index > 0 { try? await Task.sleep(for: interChunkDelay) } guard let session = try? PTYProcessSession( @@ -1116,11 +1206,17 @@ public enum ZmxSessionLauncher { /// Messages flatten the same way and for the same reason (`sendArguments`, /// `messageChunks`): `zmx` terminates what it types with `\r`, so an embedded /// newline would truncate at the first line. + /// + /// `ESC` goes for a related reason. It is never message *text*: typed as a keystroke + /// it is a composer's cancel key, and inside a bracketed paste an `ESC[201~` in the + /// payload would close the paste early and type the remainder as keystrokes — the + /// exact failure `sendWrites` exists to prevent, reintroduced from the inside. static func flattened(_ text: String) -> String { text .replacingOccurrences(of: "\r\n", with: " ") .replacingOccurrences(of: "\r", with: " ") .replacingOccurrences(of: "\n", with: " ") + .replacingOccurrences(of: "\u{1B}", with: "") } /// Where an unattended session should open, mirroring what the app already does for a @@ -1448,11 +1544,11 @@ public enum ZmxSessionLauncher { static func remoteSendInvocation( _ text: String, toNode node: LoopNode, at location: RemoteProjectLocation ) -> [String] { - // Chunked like the local path — the remote host's PTY queue is no bigger than - // ours — but chained into the one ssh round-trip, with the same drain beat - // between writes. + // Framed exactly like the local path — the composer on the far side is the same + // program with the same appetite (issue #277) — but chained into the one ssh + // round-trip, with the same drain beat between writes. let sessionName = SurfaceRef(id: node.id, launchesClaudeCode: true).zmxSessionName - let sends = messageChunks(text) + let sends = sendWrites(text) .map { quotedCommand(["zmx", "send", sessionName, $0]) } .joined(separator: " && sleep 0.15 && ") let submit = quotedCommand(["zmx"] + submitArguments(forNode: node)) diff --git a/graphcode/Tests/MessageDeliveryTests.swift b/graphcode/Tests/MessageDeliveryTests.swift new file mode 100644 index 00000000..943ce9bd --- /dev/null +++ b/graphcode/Tests/MessageDeliveryTests.swift @@ -0,0 +1,223 @@ +import Foundation +import Testing + +@testable import GraphcodeKit + +/// `send(_:to:)` driven end to end against a **real** `zmx` session, because issue #277 +/// is the second silent-loss bug in this path that every unit test of the day passed. +/// +/// The first (0.1.29) split oversized writes; its tests asserted that `messageChunks` +/// split correctly, which it did, and the message still vanished. #277 broke the split +/// path itself: every multi-chunk message arrived with its head gone and `delivered` +/// printed. A test that only checks the splitter would pass on both bugs, so this suite +/// sends the real thing through the real transport and reads back what a session actually +/// received. +/// +/// What the measurements said, since the fix is built on them rather than on a theory. +/// Nothing below the agent's TUI loses anything: 2 KB writes reach a raw-mode reader +/// byte-for-byte, and reach one that stalls 300 ms between reads byte-for-byte too. The +/// loss is the composer swallowing a long burst of keystrokes (2634 bytes sent, 590 +/// received). It cannot be fixed by chunking smaller, because the reader's reads are not +/// our writes — two 512-byte writes arrived as one 1022-byte read when the reader +/// stalled. Bracketed paste is what removes the question. +/// Disabled rather than returned early when `zmx` is missing: every test here needs a +/// real session, and a `guard … else { return }` reports a test that never ran as a +/// passing one — which is the same class of thing this suite exists to catch. +@Suite(.serialized, .enabled(if: ZmxLocator.isInstalled, "zmx is not installed")) +struct MessageDeliveryTests { + private static let zmx = ZmxLocator.binaryURL.path + + private static func quoted(_ value: String) -> String { + RemoteProjectLocation.shellQuoted(value) + } + + @discardableResult + private func run(_ command: String) async -> (succeeded: Bool, output: String) { + guard + let session = try? PTYProcessSession( + executable: "/bin/zsh", arguments: ["-c", command], workingDirectory: nil) + else { return (false, "") } + return await session.waitCollectingOutput() + } + + private func kill(_ name: String) async { + await run("\(Self.quoted(Self.zmx)) kill \(Self.quoted(name)) --force >/dev/null 2>&1") + } + + /// A live session of the shape graphcode launches, running a reader that records what + /// it was typed. `composerLimit` is how big a run of plain keystrokes it will accept + /// before swallowing it — `nil` for a reader that keeps everything. + private func startSink(_ node: LoopNode, composerLimit: Int?) async -> (name: String, out: URL)? { + let directory = FileManager.default.temporaryDirectory + .appendingPathComponent("gc-277-\(UUID().uuidString)") + try? FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true) + let script = directory.appendingPathComponent("sink.py") + let out = directory.appendingPathComponent("received.bin") + guard (try? Self.sinkSource.write(to: script, atomically: true, encoding: .utf8)) != nil + else { return nil } + + let name = SurfaceRef(id: node.id, launchesClaudeCode: true).zmxSessionName + let ready = directory.appendingPathComponent("ready") + await run( + "\(Self.quoted(Self.zmx)) run \(Self.quoted(name)) -d /usr/bin/python3 " + + "\(Self.quoted(script.path)) \(Self.quoted(out.path)) \(composerLimit ?? 0)") + // Waited on, not slept through. `zmx run` returns once the command is typed, and + // until the reader reaches `tty.setraw` its tty is still cooked — where a write + // longer than `MAX_CANON` is clipped by the line discipline and `setraw`'s + // `TCSAFLUSH` throws away whatever did land. A test that raced that would report + // exactly the head loss it exists to detect, from the wrong cause. + for _ in 0..<200 { + if FileManager.default.fileExists(atPath: ready.path) { return (name, out) } + try? await Task.sleep(for: .milliseconds(100)) + } + return nil + } + + private func received(from out: URL) async -> String? { + for _ in 0..<60 { + if let data = try? Data(contentsOf: out) { return String(bytes: data, encoding: .utf8) } + try? await Task.sleep(for: .milliseconds(100)) + } + return nil + } + + private func node() -> LoopNode { + LoopNode(title: "Sink", loopType: .goalBased, goal: GoalSpec(summary: "receive")) + } + + /// What `send` should put in front of the receiver: the message as it will be typed, + /// plus the trailer a long one carries. + private func expected(_ message: String) -> String { + let flat = ZmxSessionLauncher.flattened(message) + guard flat.utf8.count > ZmxSessionLauncher.maxUnbracketedSendBytes else { return flat } + return flat + ZmxSessionLauncher.deliveryTrailer(for: flat) + } + + private func message(bytes: Int, tag: String) -> String { + let filler = (0..<(bytes / 6 + 2)).map { String(format: "<%04d>", $0) }.joined() + return String(("[graphcode] \(tag): " + filler).prefix(bytes)) + } + + @Test + func anOversizedMessageArrivesByteForByte() async throws { + let node = node() + let sink = try #require(await startSink(node, composerLimit: nil)) + defer { Task { await kill(sink.name) } } + + // Comfortably over `maxSendChunkBytes`, so it is carried by several writes — the + // shape that lost its head in the field. + let sent = message(bytes: 6000, tag: "Sender") + #expect(await ZmxSessionLauncher.send(sent, to: node)) + + let arrived = try #require(await received(from: sink.out)) + // Byte for byte, not "contains": the failure being guarded is a message that arrives + // looking complete because only its head is missing. + #expect(arrived == expected(sent)) + #expect(arrived.hasPrefix("[graphcode] Sender: ")) + } + + @Test + func aLongMessageSurvivesAComposerThatSwallowsKeystrokeBursts() async throws { + let node = node() + // The reader models what was measured of the real composer: a run of plain + // keystrokes above this size is swallowed whole, and a bracketed paste is not, + // however large. Before the fix this message travelled as one keystroke run and the + // assertion below found an empty composer — which is #277, reproduced. + let sink = try #require( + await startSink(node, composerLimit: ZmxSessionLauncher.maxUnbracketedSendBytes)) + defer { Task { await kill(sink.name) } } + + let sent = message(bytes: 2600, tag: "Sender") + #expect(await ZmxSessionLauncher.send(sent, to: node)) + + let arrived = try #require(await received(from: sink.out)) + #expect(arrived == expected(sent)) + } + + @Test + func aShortMessageStillTravelsAsPlainKeystrokes() async throws { + let node = node() + // Same swallowing reader. A short message is deliberately *not* pasted — it is + // measured to arrive intact as typing, and leaving it on the path it already + // survives is what keeps this fix confined to the case that is broken. + let sink = try #require( + await startSink(node, composerLimit: ZmxSessionLauncher.maxUnbracketedSendBytes)) + defer { Task { await kill(sink.name) } } + + let sent = message(bytes: 400, tag: "Sender") + #expect(await ZmxSessionLauncher.send(sent, to: node)) + + let arrived = try #require(await received(from: sink.out)) + #expect(arrived == sent) + #expect(!arrived.contains(ZmxSessionLauncher.pasteStart)) + } + + /// A reader standing in for an agent's composer: raw mode, bracketed paste understood, + /// and — when given a limit — the measured habit of swallowing a long run of plain + /// keystrokes rather than keeping it. It writes what the composer holds when Enter + /// arrives, which is the moment `send` says the message was delivered. + private static let sinkSource = """ + import os, sys, tty + + target, limit = sys.argv[1], int(sys.argv[2]) + START, END = b"\\x1b[200~", b"\\x1b[201~" + fd = sys.stdin.fileno() + tty.setraw(fd) + open(os.path.join(os.path.dirname(target), "ready"), "w").close() + + composer, pending, buf = bytearray(), bytearray(), bytearray() + pasting = False + + def commit(): + # The whole of issue #277 in one branch: a burst of typing the composer will not + # take is dropped entirely rather than truncated, so what follows still reads as + # a complete message. + if not limit or len(pending) <= limit: + composer.extend(pending) + del pending[:] + + def holdback(data, marker): + # Never hand on a tail that could be the front of a marker split across reads. + for n in range(len(marker) - 1, 0, -1): + if data.endswith(marker[:n]): + return n + return 0 + + def absorb(data): + global pasting + buf.extend(data) + while True: + if pasting: + cut = buf.find(END) + if cut < 0: + keep = len(buf) - holdback(buf, END) + composer.extend(buf[:keep]) + del buf[:keep] + return + composer.extend(buf[:cut]) + del buf[: cut + len(END)] + pasting = False + continue + cut = buf.find(START) + if cut < 0: + keep = len(buf) - holdback(buf, START) + pending.extend(buf[:keep]) + del buf[:keep] + return + pending.extend(buf[:cut]) + del buf[: cut + len(START)] + commit() + pasting = True + + while True: + chunk = os.read(fd, 65536) + if not chunk: + break + absorb(chunk) + if b"\\r" in pending: + del pending[pending.index(b"\\r"):] + commit() + open(target, "wb").write(bytes(composer)) + break + """ +} diff --git a/graphcode/Tests/ZmxSessionLauncherTests.swift b/graphcode/Tests/ZmxSessionLauncherTests.swift index caf221b5..a0a77fc3 100644 --- a/graphcode/Tests/ZmxSessionLauncherTests.swift +++ b/graphcode/Tests/ZmxSessionLauncherTests.swift @@ -395,10 +395,14 @@ struct ZmxSessionLauncherTests { // MARK: - Message chunking - // Why chunks exist at all: one `zmx send` is one uninterrupted PTY write, and a - // message bigger than the PTY's 4 KB input queue vanished *entirely* — while every - // layer above, `zmx send`'s exit 0 included, reported it delivered. Measured, not - // theorized: 3.7 KB in one write arrived byte-exact, 7.3 KB arrived as nothing. + // Why chunks exist at all: one `zmx send` is one uninterrupted PTY write, and one + // large enough to outrun what the daemon will buffer has nowhere to go — while every + // layer above, `zmx send`'s exit 0 included, reports the message delivered. + // + // Chunking is *not* what keeps a message intact, which is what issue #277 cost two + // orchestration runs to learn: the splitter here was correct and every multi-chunk + // message still arrived headless. `sendWrites` carries that half — these tests cover + // only the cutting, and MessageDeliveryTests covers whether anything arrives. @Test func aShortMessageIsOneChunkUnchanged() { @@ -447,4 +451,88 @@ struct ZmxSessionLauncherTests { let chunks = ZmxSessionLauncher.messageChunks("line one\r\nline two") #expect(chunks == ["line one line two"]) } + + @Test + func anEscapeNeverSurvivesIntoWhatIsTyped() { + // Typed bare it is a composer's cancel key; inside a bracketed paste an `ESC[201~` + // would end the paste early and type the rest as the keystrokes `sendWrites` exists + // to avoid. Neither is ever what the message meant. + #expect(ZmxSessionLauncher.flattened("before\u{1B}[201~after") == "before[201~after") + } + + // MARK: - How a message is framed into writes (issue #277) + + @Test + func aShortMessageIsTypedExactlyAsItAlwaysWas() { + // The path that was never broken, kept bit-for-bit: one plain write, no markers. + #expect(ZmxSessionLauncher.sendWrites("task done") == ["task done"]) + #expect(ZmxSessionLauncher.sendWrites("") == []) + } + + @Test + func aLongMessageIsWrappedInOneBracketedPaste() { + let message = String(repeating: "GC-277-ABCDEFGH ", count: 500) // 8 KB + let writes = ZmxSessionLauncher.sendWrites(message) + + #expect(writes.count > 1) + // Exactly one paste, opened by the first write and closed by the last: a marker on a + // write of its own could be separated from the payload it frames by a failed send. + #expect(writes.first?.hasPrefix(ZmxSessionLauncher.pasteStart) == true) + #expect(writes.last?.hasSuffix(ZmxSessionLauncher.pasteEnd) == true) + #expect(writes.filter { $0.contains(ZmxSessionLauncher.pasteStart) }.count == 1) + #expect(writes.filter { $0.contains(ZmxSessionLauncher.pasteEnd) }.count == 1) + } + + @Test + func theWritesReassembleToTheMessageAndNothingElse() { + let message = String(repeating: "GC-277-ABCDEFGH ", count: 500) + let flat = ZmxSessionLauncher.flattened(message) + let typed = ZmxSessionLauncher.sendWrites(message).joined() + .replacingOccurrences(of: ZmxSessionLauncher.pasteStart, with: "") + .replacingOccurrences(of: ZmxSessionLauncher.pasteEnd, with: "") + + #expect(typed == flat + ZmxSessionLauncher.deliveryTrailer(for: flat)) + } + + @Test + func aLongMessageSaysWhatItsHeadShouldHaveBeen() { + // #277's damage was invisible because the head carried the attribution: a clipped + // message read as a whole one, and grepping for `[graphcode]` found only the + // undamaged. The trailer rides the surviving half so a receiver can tell. + let message = "[graphcode] Reviewer: " + String(repeating: "detail ", count: 400) + let flat = ZmxSessionLauncher.flattened(message) + let trailer = ZmxSessionLauncher.deliveryTrailer(for: flat) + + #expect(trailer.contains("[graphcode]")) + #expect(trailer.contains("\(flat.count)-character")) + #expect(trailer.contains("[graphcode] Reviewer: detail")) + #expect(ZmxSessionLauncher.sendWrites(message).last?.contains(trailer.suffix(20)) == true) + } + + @Test + func noWriteExceedsWhatOnePTYWriteCanCarry() { + let message = String(repeating: "x", count: 10_000) + for write in ZmxSessionLauncher.sendWrites(message) { + // The markers ride on top of a full-sized chunk, so allow for them. + let framing = + ZmxSessionLauncher.pasteStart.utf8.count + ZmxSessionLauncher.pasteEnd.utf8.count + #expect(write.utf8.count <= ZmxSessionLauncher.maxSendChunkBytes + framing) + } + } + + @Test + func theRemoteSendCarriesTheSameFramingAsTheLocalOne() throws { + // The composer on the far side is the same program with the same appetite, and the + // remote path is the one #277 could not test. It chains the same writes into one ssh + // round-trip, so it has to be chaining the *framed* ones. + let node = LoopNode(title: "Remote", loopType: .goalBased, goal: GoalSpec(summary: "work")) + let location = try #require( + RemoteProjectLocation.parse(projectPath: "ssh://someone@box/~/project")) + let message = String(repeating: "GC-277-ABCDEFGH ", count: 500) + let script = ZmxSessionLauncher.remoteSendInvocation(message, toNode: node, at: location) + .joined(separator: " ") + + #expect(script.contains("200~")) + #expect(script.contains("201~")) + } }