jpmschweitzer/clide
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

serve IPC clients with one await-for read loop (T-372)

Browse Source 
The async onData handler never paused its subscription, so pipelined
requests interleaved mid-handler — violating D-72's serial-dispatch
contract — while the shared StringBuffer could re-frame underneath an
in-flight await and the per-chunk utf8.decode corrupted runes split
across reads. One `await for` over a persistent Utf8Decoder +
LineSplitter fixes framing, decoding, and serialization at once.
Tests: two frames pipelined in one write dispatch strictly in order;
a frame split mid-rune across writes decodes intact.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
jpmschweitzer
2026-06-12 01:13:43 +02:00
co-authored by Claude Fable 5
parent bbc6899df2
commit 5d52694889
5 changed files with 86 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.pql/changelog/ticket_history/2026-06.sql
+2
View File
@@ -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;
.pql/changelog/tickets/2026-06.sql
+10
View File
@@ -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<List<int>>().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<List<int>>().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);
CHANGELOG.md
+5
View File
@@ -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,
lib/src/ipc/server.dart
+20 -27
View File
@@ -150,33 +150,26 @@ class IpcServer {
void _onClient(Socket client) {
_clients.add(client);
final buffer = StringBuffer();
late StreamSubscription<List<int>> 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<void> _serveClient(Socket client) async {
try {
await for (final line in client.cast<List<int>>().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<void> _handleLine(Socket client, String line) async {
test/ipc/server_test.dart
+49
View File
@@ -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 = <String>[];
dispatcher.register('slow', (req) async {
order.add('${req.id}:start');
await Future<void>.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<List<int>>().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<void>.delayed(const Duration(milliseconds: 30));
c.add(frame.sublist(cut));
await c.flush();
final line = await c.cast<List<int>>().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