diff --git a/.pql/changelog/ticket_history/2026-06.sql b/.pql/changelog/ticket_history/2026-06.sql index 203e917a..4e007add 100644 --- a/.pql/changelog/ticket_history/2026-06.sql +++ b/.pql/changelog/ticket_history/2026-06.sql @@ -4003,3 +4003,5 @@ INSERT INTO ticket_history (ticket_record_id, field, old_value, new_value, chang INSERT INTO ticket_history (ticket_record_id, field, old_value, new_value, changed_by, changed_at, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FBHC90B72A270CAKA7AP1ZX8', 'status', 'in_progress', 'done', NULL, '2026-06-11 23:08:13', '2026-06-11 23:08:13', '2026-06-11 23:08:13', NULL, 'f83a2fc13f3839070957bea6ac7fdc5f', 2) ON CONFLICT(hash) DO NOTHING; INSERT INTO ticket_history (ticket_record_id, field, old_value, new_value, changed_by, changed_at, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FBHCAFKK334YNJXZJQG4J6AW', 'status', 'backlog', 'in_progress', NULL, '2026-06-11 23:08:40', '2026-06-11 23:08:40', '2026-06-11 23:08:40', NULL, 'd61ecda5f46b3ad0e15ea566bc83a52c', 2) ON CONFLICT(hash) DO NOTHING; INSERT INTO ticket_history (ticket_record_id, field, old_value, new_value, changed_by, changed_at, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FBHCAFKK334YNJXZJQG4J6AW', 'status', 'in_progress', 'done', NULL, '2026-06-11 23:10:56', '2026-06-11 23:10:56', '2026-06-11 23:10:56', NULL, '4b505b885dc39785651e259370fbc97b', 2) ON CONFLICT(hash) DO NOTHING; +INSERT INTO ticket_history (ticket_record_id, field, old_value, new_value, changed_by, changed_at, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FBHC5ZE4EZEGXK8YY8J86CM0', 'status', 'backlog', 'in_progress', NULL, '2026-06-11 23:11:24', '2026-06-11 23:11:24', '2026-06-11 23:11:24', NULL, '4763adf4924a8f8452c393db9ad03868', 2) ON CONFLICT(hash) DO NOTHING; +INSERT INTO ticket_history (ticket_record_id, field, old_value, new_value, changed_by, changed_at, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FBHC5ZE4EZEGXK8YY8J86CM0', 'status', 'in_progress', 'done', NULL, '2026-06-11 23:13:18', '2026-06-11 23:13:18', '2026-06-11 23:13:18', NULL, '5382c848b654d1a93daed15a28d652cf', 2) ON CONFLICT(hash) DO NOTHING; diff --git a/.pql/changelog/tickets/2026-06.sql b/.pql/changelog/tickets/2026-06.sql index 2188e2e2..ee08b871 100644 --- a/.pql/changelog/tickets/2026-06.sql +++ b/.pql/changelog/tickets/2026-06.sql @@ -4119,3 +4119,13 @@ INSERT INTO tickets (record_id, type, parent_record_id, title, description, stat Fix: treat forkSourceId as a one-shot spawn parameter — consume it on first bind (clear it into pane state), so subsequent session-mutating commands operate on the pane''s live session. Acceptance: test that a fork pane after /clear starts an empty session (no fork source passed to the orchestrator on respawn); first bind still forks from the source.', 'done', 'medium', NULL, NULL, NULL, '2026-06-11 21:58:08', '2026-06-11 23:10:56', NULL, '1cd70120bb752815eeaebb58e57b1a1c', 2) ON CONFLICT(record_id) DO UPDATE SET type=excluded.type, parent_record_id=excluded.parent_record_id, title=excluded.title, description=excluded.description, status=excluded.status, priority=excluded.priority, assigned_to=excluded.assigned_to, team=excluded.team, decision_ref=excluded.decision_ref, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at > tickets.updated_at OR (excluded.updated_at = tickets.updated_at AND excluded.hash > tickets.hash); +INSERT INTO tickets (record_id, type, parent_record_id, title, description, status, priority, assigned_to, team, decision_ref, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FBHC5ZE4EZEGXK8YY8J86CM0', 'bug', '06FBHBGHNEQTAEPGNJKN42C1E8', 'IPC server dispatch is not serial; shared StringBuffer framing can drop/double lines', 'lib/src/ipc/server.dart:151-180 — D-72 promises serial dispatch, but the async onData handler never pauses the subscription, so pipelined requests interleave; the shared StringBuffer framing can also drop or double lines when chunks split mid-frame, and per-chunk utf8 decode corrupts multi-byte characters split across chunks. + +Fix in one move: client.cast>().transform(utf8.decoder).transform(const LineSplitter()) consumed with await for — gives correct framing, persistent UTF-8 decoding, and true serialization at once. + +Acceptance: test sending two pipelined requests in a single write (responses arrive in order, both handled); test a request split mid-UTF-8-rune across two socket writes; existing IPC tests stay green. Runs under dart test — keep imports Flutter-free.', 'in_progress', 'medium', NULL, NULL, NULL, '2026-06-11 21:57:31', '2026-06-11 23:11:24', NULL, '03c848a027a4ae292e854398358eb150', 2) ON CONFLICT(record_id) DO UPDATE SET type=excluded.type, parent_record_id=excluded.parent_record_id, title=excluded.title, description=excluded.description, status=excluded.status, priority=excluded.priority, assigned_to=excluded.assigned_to, team=excluded.team, decision_ref=excluded.decision_ref, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at > tickets.updated_at OR (excluded.updated_at = tickets.updated_at AND excluded.hash > tickets.hash); +INSERT INTO tickets (record_id, type, parent_record_id, title, description, status, priority, assigned_to, team, decision_ref, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FBHC5ZE4EZEGXK8YY8J86CM0', 'bug', '06FBHBGHNEQTAEPGNJKN42C1E8', 'IPC server dispatch is not serial; shared StringBuffer framing can drop/double lines', 'lib/src/ipc/server.dart:151-180 — D-72 promises serial dispatch, but the async onData handler never pauses the subscription, so pipelined requests interleave; the shared StringBuffer framing can also drop or double lines when chunks split mid-frame, and per-chunk utf8 decode corrupts multi-byte characters split across chunks. + +Fix in one move: client.cast>().transform(utf8.decoder).transform(const LineSplitter()) consumed with await for — gives correct framing, persistent UTF-8 decoding, and true serialization at once. + +Acceptance: test sending two pipelined requests in a single write (responses arrive in order, both handled); test a request split mid-UTF-8-rune across two socket writes; existing IPC tests stay green. Runs under dart test — keep imports Flutter-free.', 'done', 'medium', NULL, NULL, NULL, '2026-06-11 21:57:31', '2026-06-11 23:13:18', NULL, 'bbc0aa50303eac50666cbcfa140065a1', 2) ON CONFLICT(record_id) DO UPDATE SET type=excluded.type, parent_record_id=excluded.parent_record_id, title=excluded.title, description=excluded.description, status=excluded.status, priority=excluded.priority, assigned_to=excluded.assigned_to, team=excluded.team, decision_ref=excluded.decision_ref, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at > tickets.updated_at OR (excluded.updated_at = tickets.updated_at AND excluded.hash > tickets.hash); diff --git a/CHANGELOG.md b/CHANGELOG.md index 7c6efefd..7ad29b28 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -104,6 +104,11 @@ heading, and (b) bumping `pubspec.yaml` `version:` in the same commit. the original conversation again; the source now seeds only the first bind. (T-375) +- **Pipelined IPC requests are now truly serial and framing-safe.** The + server's read handler could interleave concurrent requests (against + D-72's contract), drop or double frames split across reads, and corrupt + multi-byte characters split across chunks. (T-372) + - **Accepting ExitPlanMode now leaves plan mode in the conversation panel.** Approving Claude's plan (the ExitPlanMode tool) transitioned the underlying session out of plan mode, but clide's tracked permission mode didn't follow, diff --git a/lib/src/ipc/server.dart b/lib/src/ipc/server.dart index fd7742a2..1628e3a0 100644 --- a/lib/src/ipc/server.dart +++ b/lib/src/ipc/server.dart @@ -150,33 +150,26 @@ class IpcServer { void _onClient(Socket client) { _clients.add(client); - final buffer = StringBuffer(); - late StreamSubscription> sub; - sub = client.listen( - (chunk) async { - buffer.write(utf8.decode(chunk, allowMalformed: true)); - var idx = buffer.toString().indexOf('\n'); - while (idx >= 0) { - final raw = buffer.toString().substring(0, idx); - // Trim consumed bytes by rebuilding the buffer with the - // tail — StringBuffer can't slice in place. - final tail = buffer.toString().substring(idx + 1); - buffer.clear(); - buffer.write(tail); - await _handleLine(client, raw); - idx = buffer.toString().indexOf('\n'); - } - }, - onError: (Object e, StackTrace st) { - log.warn('ipc', 'client read error: $e'); - }, - onDone: () { - _clients.remove(client); - _subscribers.remove(client); - sub.cancel(); - }, - cancelOnError: true, - ); + unawaited(_serveClient(client)); + } + + /// One read loop per connection: persistent UTF-8 decode, line framing, + /// and true serial dispatch in a single `await for` (D-72, T-372). The + /// old async onData handler never paused its subscription — pipelined + /// requests interleaved mid-handler, the shared StringBuffer could + /// re-frame while an await was in flight, and per-chunk decode corrupted + /// runes split across reads. + Future _serveClient(Socket client) async { + try { + await for (final line in client.cast>().transform(const Utf8Decoder(allowMalformed: true)).transform(const LineSplitter())) { + await _handleLine(client, line); + } + } catch (e) { + log.warn('ipc', 'client read error: $e'); + } finally { + _clients.remove(client); + _subscribers.remove(client); + } } Future _handleLine(Socket client, String line) async { diff --git a/test/ipc/server_test.dart b/test/ipc/server_test.dart index bc16ca4e..74e451be 100644 --- a/test/ipc/server_test.dart +++ b/test/ipc/server_test.dart @@ -201,6 +201,55 @@ void main() { await c.close(); }); + // T-372: the old async onData never paused its subscription, so + // pipelined requests interleaved mid-handler; per-chunk decode also + // corrupted runes split across socket writes. + test('two requests pipelined in one write are handled serially, in order (T-372/D-72)', () async { + final order = []; + dispatcher.register('slow', (req) async { + order.add('${req.id}:start'); + await Future.delayed(const Duration(milliseconds: 50)); + order.add('${req.id}:end'); + return IpcResponse.ok(id: req.id); + }); + server = IpcServer(dispatcher: dispatcher, workspaceRoot: workRoot, log: _silentLog()); + await server.start(); + final c = await Socket.connect(InternetAddress(server.socketPath, type: InternetAddressType.unix), 0); + // Single write carrying both frames. + c.write('${IpcRequest(id: 'p1', cmd: 'slow').encode()}\n${IpcRequest(id: 'p2', cmd: 'slow').encode()}\n'); + await c.flush(); + final replies = c.cast>().transform(utf8.decoder).transform(const LineSplitter()); + final got = await replies.take(2).toList().timeout(const Duration(seconds: 5)); + await c.close(); + expect((IpcMessage.decode(got[0]) as IpcResponse).id, 'p1'); + expect((IpcMessage.decode(got[1]) as IpcResponse).id, 'p2'); + expect(order, ['p1:start', 'p1:end', 'p2:start', 'p2:end'], reason: 'D-72: dispatch is serial, never interleaved'); + }); + + test('a request split mid-UTF-8-rune across two writes decodes intact (T-372)', () async { + String? gotText; + dispatcher.register('echo', (req) async { + gotText = req.args['text'] as String?; + return IpcResponse.ok(id: req.id, data: {'echo': gotText}); + }); + server = IpcServer(dispatcher: dispatcher, workspaceRoot: workRoot, log: _silentLog()); + await server.start(); + final c = await Socket.connect(InternetAddress(server.socketPath, type: InternetAddressType.unix), 0); + final frame = utf8.encode('${IpcRequest(id: 'u1', cmd: 'echo', args: const {'text': 'héllo — ünïcode'}).encode()}\n'); + // Split inside the multi-byte 'é' (the first non-ASCII rune). + final cut = frame.indexWhere((b) => b > 0x7f) + 1; + c.add(frame.sublist(0, cut)); + await c.flush(); + await Future.delayed(const Duration(milliseconds: 30)); + c.add(frame.sublist(cut)); + await c.flush(); + final line = await c.cast>().transform(utf8.decoder).transform(const LineSplitter()).first.timeout(const Duration(seconds: 2)); + await c.close(); + final reply = IpcMessage.decode(line) as IpcResponse; + expect(reply.ok, isTrue); + expect(gotText, 'héllo — ünïcode', reason: 'persistent decoder must join the split rune'); + }); + test('socketPath returns the resolved path before start (no bind)', () async { server = IpcServer(dispatcher: dispatcher, workspaceRoot: workRoot, log: _silentLog()); // Before start, the getter falls back to workspaceSocketPath; it