| 475 | } |
| 476 | |
| 477 | static bool frontend_enqueue(frontend_state_t *state, char *message, bool content_length_framed) { |
| 478 | size_t length = strlen(message); |
| 479 | bool has_id = false; |
| 480 | int64_t id = 0; |
| 481 | char *id_str = NULL; |
| 482 | cbm_jsonrpc_request_t request = {0}; |
| 483 | if (cbm_jsonrpc_parse(message, &request) == 0) { |
| 484 | has_id = request.has_id; |
| 485 | id = request.id; |
| 486 | if (request.id_str) { |
| 487 | id_str = cbm_strdup(request.id_str); |
| 488 | } |
| 489 | bool identity_copied = !request.id_str || id_str; |
| 490 | cbm_jsonrpc_request_free(&request); |
| 491 | if (!identity_copied) { |
| 492 | return false; |
| 493 | } |
| 494 | } |
| 495 | /* A frame that exceeds the whole byte budget can never be admitted no |
| 496 | * matter how long we wait; only that case is a hard failure. */ |
| 497 | if (length > FRONTEND_QUEUE_BYTES_MAX) { |
| 498 | cbm_mutex_lock(&state->mutex); |
| 499 | if (!state->stopping && !state->failed) { |
| 500 | state->failed = true; |
| 501 | state->stopping = true; |
| 502 | } |
| 503 | cbm_mutex_unlock(&state->mutex); |
| 504 | free(id_str); |
| 505 | return false; |
| 506 | } |
| 507 | /* A full queue is BACKPRESSURE, not failure. This runs on the stdin reader |
| 508 | * thread, so blocking here stops further reads and lets the kernel pipe |
| 509 | * absorb the client's burst — a pipelining client is a client going fast, |
| 510 | * not a client going wrong. The old code failed the whole session at frame |
| 511 | * capacity+1, killing it with every buffered response unwritten: any 7+ |
| 512 | * pipelined requests (e.g. an agent issuing parallel tool calls) died with |
| 513 | * rc=1 and zero output (#1522 follow-up). Waits are bounded only by the |
| 514 | * stop/fail flags, which every teardown path already sets — the same |
| 515 | * condition the polling worker and watchdogs key on. */ |
| 516 | for (;;) { |
| 517 | cbm_mutex_lock(&state->mutex); |
| 518 | bool stopped = state->stopping || state->failed; |
| 519 | if (stopped) { |
| 520 | cbm_mutex_unlock(&state->mutex); |
| 521 | free(id_str); |
| 522 | return false; |
| 523 | } |
| 524 | bool capacity = state->count < FRONTEND_QUEUE_CAPACITY && |
| 525 | state->queued_bytes <= FRONTEND_QUEUE_BYTES_MAX && |
| 526 | length <= FRONTEND_QUEUE_BYTES_MAX - state->queued_bytes; |
| 527 | if (capacity) { |
| 528 | size_t tail = (state->head + state->count) % FRONTEND_QUEUE_CAPACITY; |
| 529 | state->queue[tail] = (frontend_item_t){ |
| 530 | .message = message, |
| 531 | .length = length, |
| 532 | .content_length_framed = content_length_framed, |
| 533 | .has_id = has_id, |
| 534 | .id = id, |
no test coverage detected