From 358d6e76ab749a7808ce02089b1b5b8d6431fc82 Mon Sep 17 00:00:00 2001 From: Pxx500 Date: Thu, 20 Aug 2026 16:29:19 +0200 Subject: [PATCH 1/5] fix(mcp): recover unsent requests after daemon loss Signed-off-by: Pxx500 --- src/daemon/application.c | 27 +-- src/daemon/application.h | 5 +- src/daemon/frontend.c | 183 +++++++++++++++----- src/daemon/frontend.h | 22 ++- src/daemon/runtime.c | 14 +- src/daemon/runtime.h | 5 +- src/main.c | 60 +++++-- tests/test_daemon_frontend.c | 322 +++++++++++++++++++++++++++++++---- tests/test_daemon_runtime.c | 27 ++- 9 files changed, 547 insertions(+), 118 deletions(-) diff --git a/src/daemon/application.c b/src/daemon/application.c index 69ff52ec7f..58fb6bb311 100644 --- a/src/daemon/application.c +++ b/src/daemon/application.c @@ -3116,7 +3116,7 @@ cbm_daemon_runtime_application_callbacks_t cbm_daemon_application_runtime_callba static cbm_daemon_runtime_application_status_t application_client_exchange_tagged( cbm_daemon_runtime_client_t *client, cbm_daemon_runtime_application_token_t request_token, uint8_t *request, uint32_t request_length, uint8_t **response_out, - uint32_t *response_length_out, uint32_t timeout_ms) { + uint32_t *response_length_out, bool *request_sent_out, uint32_t timeout_ms) { uint8_t *response = NULL; uint32_t response_length = 0; if (response_out) { @@ -3125,13 +3125,16 @@ static cbm_daemon_runtime_application_status_t application_client_exchange_tagge if (response_length_out) { *response_length_out = 0; } + if (request_sent_out) { + *request_sent_out = false; + } cbm_daemon_runtime_application_status_t status = request_token == CBM_DAEMON_RUNTIME_APPLICATION_TOKEN_INVALID ? cbm_daemon_runtime_client_application_request(client, request, request_length, &response, &response_length, timeout_ms) - : cbm_daemon_runtime_client_application_request_tagged(client, request_token, request, - request_length, &response, - &response_length, timeout_ms); + : cbm_daemon_runtime_client_application_request_tagged( + client, request_token, request, request_length, &response, &response_length, + request_sent_out, timeout_ms); free(request); if (status != CBM_DAEMON_RUNTIME_APPLICATION_OK) { free(response); @@ -3162,7 +3165,7 @@ static cbm_daemon_runtime_application_status_t application_client_exchange( uint8_t **response_out, uint32_t *response_length_out, uint32_t timeout_ms) { return application_client_exchange_tagged(client, CBM_DAEMON_RUNTIME_APPLICATION_TOKEN_INVALID, request, request_length, response_out, - response_length_out, timeout_ms); + response_length_out, NULL, timeout_ms); } cbm_daemon_runtime_application_status_t cbm_daemon_application_client_set_context( @@ -3290,7 +3293,7 @@ cbm_daemon_runtime_application_status_t cbm_daemon_application_client_ui_readine static cbm_daemon_runtime_application_status_t application_client_text_request_tagged( cbm_daemon_runtime_client_t *client, cbm_daemon_runtime_application_token_t request_token, cbm_daemon_application_request_kind_t kind, const char *text, uint8_t **response_out, - uint32_t *response_length_out, uint32_t timeout_ms) { + uint32_t *response_length_out, bool *request_sent_out, uint32_t timeout_ms) { if (!client || !text || !text[0]) { return CBM_DAEMON_RUNTIME_APPLICATION_REJECTED; } @@ -3306,7 +3309,7 @@ static cbm_daemon_runtime_application_status_t application_client_text_request_t memcpy(request + 1, text, text_length); return application_client_exchange_tagged(client, request_token, request, (uint32_t)text_length + 1U, response_out, - response_length_out, timeout_ms); + response_length_out, request_sent_out, timeout_ms); } static cbm_daemon_runtime_application_status_t application_client_text_request( @@ -3314,7 +3317,7 @@ static cbm_daemon_runtime_application_status_t application_client_text_request( const char *text, uint8_t **response_out, uint32_t *response_length_out, uint32_t timeout_ms) { return application_client_text_request_tagged( client, CBM_DAEMON_RUNTIME_APPLICATION_TOKEN_INVALID, kind, text, response_out, - response_length_out, timeout_ms); + response_length_out, NULL, timeout_ms); } cbm_daemon_runtime_application_status_t cbm_daemon_application_client_mcp( @@ -3327,10 +3330,10 @@ cbm_daemon_runtime_application_status_t cbm_daemon_application_client_mcp( cbm_daemon_runtime_application_status_t cbm_daemon_application_client_mcp_tagged( cbm_daemon_runtime_client_t *client, cbm_daemon_runtime_application_token_t request_token, const char *message, uint8_t **response_out, uint32_t *response_length_out, - uint32_t timeout_ms) { - return application_client_text_request_tagged(client, request_token, - CBM_DAEMON_APPLICATION_REQUEST_MCP, message, - response_out, response_length_out, timeout_ms); + bool *request_sent_out, uint32_t timeout_ms) { + return application_client_text_request_tagged( + client, request_token, CBM_DAEMON_APPLICATION_REQUEST_MCP, message, response_out, + response_length_out, request_sent_out, timeout_ms); } cbm_daemon_runtime_application_status_t cbm_daemon_application_client_tool( diff --git a/src/daemon/application.h b/src/daemon/application.h index 933186aa73..b071ea03cc 100644 --- a/src/daemon/application.h +++ b/src/daemon/application.h @@ -149,11 +149,12 @@ cbm_daemon_runtime_application_status_t cbm_daemon_application_client_mcp( cbm_daemon_runtime_client_t *client, const char *message, uint8_t **response_out, uint32_t *response_length_out, uint32_t timeout_ms); -/* Cancellable frontend variant using a token reserved on the runtime client. */ +/* Cancellable frontend variant using a token reserved on the runtime client. + * request_sent_out preserves the runtime's application-frame delivery fact. */ cbm_daemon_runtime_application_status_t cbm_daemon_application_client_mcp_tagged( cbm_daemon_runtime_client_t *client, cbm_daemon_runtime_application_token_t request_token, const char *message, uint8_t **response_out, uint32_t *response_length_out, - uint32_t timeout_ms); + bool *request_sent_out, uint32_t timeout_ms); cbm_daemon_runtime_application_status_t cbm_daemon_application_client_tool( cbm_daemon_runtime_client_t *client, const char *tool_name, const char *args_json, diff --git a/src/daemon/frontend.c b/src/daemon/frontend.c index 2dd9869011..92b0326867 100644 --- a/src/daemon/frontend.c +++ b/src/daemon/frontend.c @@ -76,6 +76,7 @@ typedef struct { cbm_mutex_t mutex; cbm_daemon_runtime_client_t *client; cbm_version_cohort_manager_t *cohort_manager; + const cbm_daemon_frontend_session_config_t *session; FILE *out; frontend_item_t queue[FRONTEND_QUEUE_CAPACITY]; size_t head; @@ -89,6 +90,7 @@ typedef struct { int64_t active_id; const char *active_id_str; cbm_daemon_runtime_application_token_t active_request_token; + bool active_cancelled; bool failed; /* Monotonic count of fully processed queue items (responses written or * cancellations acknowledged). The EOF drain below watches it to tell a @@ -352,6 +354,7 @@ static bool frontend_pop_begin(frontend_state_t *state, frontend_item_t *item) { state->active_id_str = item->id_str; item->request_token = request_token; state->active_request_token = request_token; + state->active_cancelled = false; popped = true; } cbm_mutex_unlock(&state->mutex); @@ -365,6 +368,7 @@ static void frontend_end_request(frontend_state_t *state, bool failed) { state->active_id = 0; state->active_id_str = NULL; state->active_request_token = CBM_DAEMON_RUNTIME_APPLICATION_TOKEN_INVALID; + state->active_cancelled = false; state->failed = state->failed || failed; cbm_mutex_unlock(&state->mutex); } @@ -446,12 +450,8 @@ typedef enum { * request routes its exact runtime token without closing the authenticated * session. A queued request is marked and receives a cancellation error * without ever reaching the daemon. Stale/invalid targets are ignored. */ -static frontend_cancellation_route_t frontend_route_cancellation( - frontend_state_t *state, const char *message, - cbm_daemon_runtime_application_token_t *request_token_out) { - if (request_token_out) { - *request_token_out = CBM_DAEMON_RUNTIME_APPLICATION_TOKEN_INVALID; - } +static frontend_cancellation_route_t frontend_route_cancellation(frontend_state_t *state, + const char *message) { cbm_jsonrpc_request_t request = {0}; if (!frontend_parse_cancellation(message, &request)) { return FRONTEND_CANCELLATION_NONE; @@ -460,12 +460,9 @@ static frontend_cancellation_route_t frontend_route_cancellation( frontend_cancellation_route_t route = FRONTEND_CANCELLATION_STALE; cbm_mutex_lock(&state->mutex); if (state->in_request && state->active_has_id && - state->active_request_token != CBM_DAEMON_RUNTIME_APPLICATION_TOKEN_INVALID && cbm_mcp_cancel_request_matches(request.params_raw, state->active_id, state->active_id_str)) { - if (request_token_out) { - *request_token_out = state->active_request_token; - } + state->active_cancelled = true; route = FRONTEND_CANCELLATION_ACTIVE; } else { for (size_t offset = 0; offset < state->count; offset++) { @@ -484,6 +481,89 @@ static frontend_cancellation_route_t frontend_route_cancellation( return route; } +static cbm_daemon_runtime_cancel_result_t frontend_cancel_active(frontend_state_t *state) { + cbm_mutex_lock(&state->mutex); + cbm_daemon_runtime_cancel_result_t result = CBM_DAEMON_RUNTIME_CANCEL_STALE; + if (state->in_request && state->active_cancelled && state->client && + state->active_request_token != CBM_DAEMON_RUNTIME_APPLICATION_TOKEN_INVALID) { + result = cbm_daemon_runtime_client_application_cancel(state->client, + state->active_request_token); + } else if (state->in_request && state->active_cancelled) { + /* A reconnect has no daemon request to cancel yet. The worker observes + * active_cancelled before retrying the frame. */ + result = CBM_DAEMON_RUNTIME_CANCEL_ACCEPTED; + } + cbm_mutex_unlock(&state->mutex); + return result; +} + +static bool frontend_configure_client(cbm_daemon_runtime_client_t *client, + const cbm_daemon_frontend_session_config_t *session) { + uint32_t timeout_ms = session->bootstrap.connect_timeout_ms; + if (cbm_daemon_application_client_set_context( + client, session->session_root, session->allowed_root, session->tool_profile, NULL, NULL, + timeout_ms) != CBM_DAEMON_RUNTIME_APPLICATION_OK) { + return false; + } + return session->ui_update_mask == 0 || + cbm_daemon_application_client_set_ui_config( + client, session->ui_update_mask, session->ui_enabled, session->ui_port, + timeout_ms) == CBM_DAEMON_RUNTIME_APPLICATION_OK; +} + +static bool frontend_recover_client(frontend_state_t *state, frontend_item_t *item, + bool *cancelled_out) { + if (cancelled_out) { + *cancelled_out = false; + } + + cbm_mutex_lock(&state->mutex); + if (state->stopping || !state->client) { + cbm_mutex_unlock(&state->mutex); + return false; + } + cbm_daemon_runtime_client_t *previous = state->client; + state->client = NULL; + state->active_request_token = CBM_DAEMON_RUNTIME_APPLICATION_TOKEN_INVALID; + cbm_mutex_unlock(&state->mutex); + if (previous) { + (void)cbm_daemon_runtime_client_close(previous, FRONTEND_CLOSE_TIMEOUT_MS); + } + + cbm_daemon_bootstrap_result_t bootstrap = {0}; + cbm_daemon_bootstrap_status_t bootstrap_status = + cbm_daemon_bootstrap_execute(&state->session->bootstrap, &bootstrap); + cbm_daemon_runtime_client_t *replacement = + bootstrap_status == CBM_DAEMON_BOOTSTRAP_CONNECTED ? bootstrap.client : NULL; + bool configured = replacement && frontend_configure_client(replacement, state->session); + cbm_daemon_runtime_application_token_t replacement_token = + CBM_DAEMON_RUNTIME_APPLICATION_TOKEN_INVALID; + + cbm_mutex_lock(&state->mutex); + bool cancelled = state->active_cancelled; + bool usable = configured && !state->stopping && !state->failed; + if (usable && !cancelled) { + usable = + cbm_daemon_runtime_client_application_token_reserve(replacement, &replacement_token); + } + if (usable) { + state->client = replacement; + if (!cancelled) { + state->active_request_token = replacement_token; + item->request_token = replacement_token; + } + if (cancelled_out) { + *cancelled_out = cancelled; + } + } + cbm_mutex_unlock(&state->mutex); + + if (!usable && replacement) { + (void)cbm_daemon_runtime_client_close(replacement, FRONTEND_CLOSE_TIMEOUT_MS); + } + return usable; +} + static void *frontend_worker(void *opaque) { frontend_state_t *state = opaque; for (;;) { @@ -504,10 +584,27 @@ static void *frontend_worker(void *opaque) { if (item.cancelled) { failed = !frontend_write_cancelled_response(state->out, &item); } else { + bool request_sent = false; cbm_daemon_runtime_application_status_t status = - cbm_daemon_application_client_mcp_tagged(state->client, item.request_token, - item.message, &response, &response_length, - FRONTEND_REQUEST_TIMEOUT_MS); + cbm_daemon_application_client_mcp_tagged( + state->client, item.request_token, item.message, &response, &response_length, + &request_sent, FRONTEND_REQUEST_TIMEOUT_MS); + if (status == CBM_DAEMON_RUNTIME_APPLICATION_TRANSPORT_ERROR && !request_sent) { + free(response); + response = NULL; + response_length = 0; + bool cancelled = false; + if (frontend_recover_client(state, &item, &cancelled)) { + if (cancelled) { + status = CBM_DAEMON_RUNTIME_APPLICATION_CANCELLED; + } else { + bool retry_sent = false; + status = cbm_daemon_application_client_mcp_tagged( + state->client, item.request_token, item.message, &response, + &response_length, &retry_sent, FRONTEND_REQUEST_TIMEOUT_MS); + } + } + } if (status == CBM_DAEMON_RUNTIME_APPLICATION_CANCELLED) { failed = !frontend_write_cancelled_response(state->out, &item); } else { @@ -526,12 +623,11 @@ static void *frontend_worker(void *opaque) { atomic_fetch_add_explicit(&state->completed_items, 1, memory_order_release); if (failed) { if (!expected_stop) { - /* A failed daemon transport cannot wake a thread blocked in - * stdio portably. This frontend owns no state: immediate - * process exit closes the kernel IPC handle, which cancels - * daemon session ownership without a detached reader or an - * unsafe cross-thread fclose. Logging or flushing here could - * itself block on agent-owned stdout/stderr. */ + /* Recovery was unsafe or the one safe retry failed. Immediate + * process exit wakes an agent blocked on stdio and closes the + * kernel IPC handle without a detached reader or cross-thread + * fclose. Logging or flushing here could itself block on + * agent-owned stdout/stderr. */ _Exit(EXIT_FAILURE); } break; @@ -614,11 +710,11 @@ static bool frontend_enqueue(frontend_state_t *state, char *message, bool conten static bool frontend_stop_begin(frontend_state_t *state) { cbm_mutex_lock(&state->mutex); state->stopping = true; + cbm_daemon_runtime_client_t *client = state->client; + bool stop_begun = !client || cbm_daemon_runtime_client_close_begin(client); cbm_mutex_unlock(&state->mutex); - /* Retain the client allocation until the worker is joined. This covers the - * boundary where the worker has claimed an item but has not yet entered the - * runtime exchange: a late call observes closing instead of freed memory. */ - return cbm_daemon_runtime_client_close_begin(state->client); + /* NULL means recovery owns cleanup. Otherwise retain the client until join. */ + return stop_begun; } static bool frontend_cancel_for_maintenance(void *opaque) { @@ -628,24 +724,33 @@ static bool frontend_cancel_for_maintenance(void *opaque) { cbm_mutex_lock(&state->mutex); state->stopping = true; if (state->in_request) { + state->active_cancelled = true; request_token = state->active_request_token; } - cbm_mutex_unlock(&state->mutex); if (request_token == CBM_DAEMON_RUNTIME_APPLICATION_TOKEN_INVALID) { + cbm_mutex_unlock(&state->mutex); return false; } - return cbm_daemon_runtime_client_application_cancel(state->client, request_token) == - CBM_DAEMON_RUNTIME_CANCEL_ACCEPTED; + bool cancelled = state->client && + cbm_daemon_runtime_client_application_cancel(state->client, request_token) == + CBM_DAEMON_RUNTIME_CANCEL_ACCEPTED; + cbm_mutex_unlock(&state->mutex); + return cancelled; } int cbm_daemon_frontend_mcp_run(cbm_daemon_runtime_client_t *client, - cbm_version_cohort_manager_t *cohort_manager, FILE *in, FILE *out) { - if (!client || !cohort_manager || !in || !out) { + cbm_version_cohort_manager_t *cohort_manager, + const cbm_daemon_frontend_session_config_t *session, FILE *in, + FILE *out) { + if (!client || !cohort_manager || !session || !session->bootstrap.endpoint || + !session->bootstrap.identity || !session->session_root || !session->session_root[0] || + !in || !out) { return -1; } frontend_state_t state = { .client = client, .cohort_manager = cohort_manager, + .session = session, .out = out, }; cbm_mutex_init(&state.mutex); @@ -669,7 +774,7 @@ int cbm_daemon_frontend_mcp_run(cbm_daemon_runtime_client_t *client, } int result = 0; - bool close_begun = false; + bool stop_begun = false; bool clean_eof = false; for (;;) { char *message = NULL; @@ -681,18 +786,14 @@ int cbm_daemon_frontend_mcp_run(cbm_daemon_runtime_client_t *client, free(message); break; } - cbm_daemon_runtime_application_token_t cancel_token = - CBM_DAEMON_RUNTIME_APPLICATION_TOKEN_INVALID; - frontend_cancellation_route_t cancellation = - frontend_route_cancellation(&state, message, &cancel_token); + frontend_cancellation_route_t cancellation = frontend_route_cancellation(&state, message); if (cancellation != FRONTEND_CANCELLATION_NONE) { free(message); if (cancellation == FRONTEND_CANCELLATION_ACTIVE) { - cbm_daemon_runtime_cancel_result_t cancelled = - cbm_daemon_runtime_client_application_cancel(state.client, cancel_token); + cbm_daemon_runtime_cancel_result_t cancelled = frontend_cancel_active(&state); if (cancelled == CBM_DAEMON_RUNTIME_CANCEL_ERROR) { result = -1; - close_begun = frontend_stop_begin(&state); + stop_begun = frontend_stop_begin(&state); break; } } @@ -707,7 +808,7 @@ int cbm_daemon_frontend_mcp_run(cbm_daemon_runtime_client_t *client, } } - if (clean_eof && !close_begun) { + if (clean_eof && !stop_begun) { frontend_input_closed(&state); /* EOF ends INPUT, not accepted work: as long as queued items keep * completing, every already-enqueued request still receives its @@ -733,8 +834,8 @@ int cbm_daemon_frontend_mcp_run(cbm_daemon_runtime_client_t *client, } } } - if (!close_begun) { - close_begun = frontend_stop_begin(&state); + if (!stop_begun) { + stop_begun = frontend_stop_begin(&state); } frontend_join_watchdog_t watchdog; cbm_thread_t watchdog_thread; @@ -759,9 +860,9 @@ int cbm_daemon_frontend_mcp_run(cbm_daemon_runtime_client_t *client, if (!cbm_daemon_maintenance_monitor_stop(&maintenance_monitor)) { _Exit(EXIT_FAILURE); } - if (close_begun) { + if (stop_begun && state.client) { (void)cbm_daemon_runtime_client_close_finish(state.client, FRONTEND_CLOSE_TIMEOUT_MS); - } else { + } else if (!stop_begun) { result = -1; } for (size_t i = 0; i < FRONTEND_QUEUE_CAPACITY; i++) { diff --git a/src/daemon/frontend.h b/src/daemon/frontend.h index f23395774a..10255687fd 100644 --- a/src/daemon/frontend.h +++ b/src/daemon/frontend.h @@ -4,14 +4,29 @@ #ifndef CBM_DAEMON_FRONTEND_H #define CBM_DAEMON_FRONTEND_H +#include "daemon/bootstrap.h" #include "daemon/runtime.h" #include "daemon/version_cohort.h" +#include "mcp/mcp.h" #include #include typedef struct cbm_daemon_maintenance_monitor cbm_daemon_maintenance_monitor_t; +/* Everything the stdio frontend must replay when its authenticated daemon + * connection is replaced. Pointer fields are borrowed for the complete + * frontend call. */ +typedef struct { + cbm_daemon_bootstrap_config_t bootstrap; + const char *session_root; + const char *allowed_root; + cbm_mcp_tool_profile_t tool_profile; + uint8_t ui_update_mask; + bool ui_enabled; + int ui_port; +} cbm_daemon_frontend_session_config_t; + /* Called once when install/update/uninstall requests an active local command * to stop cooperatively. Returning false does not authorize the command to * outlive the bounded grace period. */ @@ -60,9 +75,12 @@ uint64_t cbm_daemon_frontend_test_worker_idle_cycles(void); * either thread is blocked in stdio, requests cooperative cancellation for the * exact active request, and then bounds process exit. Kernel IPC close cancels * only this session's daemon work. EOF/parse failure closes the authenticated - * session. An unexpected daemon transport failure likewise terminates the + * session. A transport failure before the application frame is sent replaces + * the daemon client and retries once. Sent or repeated failures terminate the * process so an agent waiting with stdin still open observes server EOF. */ int cbm_daemon_frontend_mcp_run(cbm_daemon_runtime_client_t *client, - cbm_version_cohort_manager_t *cohort_manager, FILE *in, FILE *out); + cbm_version_cohort_manager_t *cohort_manager, + const cbm_daemon_frontend_session_config_t *session, FILE *in, + FILE *out); #endif /* CBM_DAEMON_FRONTEND_H */ diff --git a/src/daemon/runtime.c b/src/daemon/runtime.c index 7db0f0225a..c9eaeda499 100644 --- a/src/daemon/runtime.c +++ b/src/daemon/runtime.c @@ -3019,13 +3019,16 @@ cbm_daemon_runtime_cancel_result_t cbm_daemon_runtime_client_application_cancel( cbm_daemon_runtime_application_status_t cbm_daemon_runtime_client_application_request_tagged( cbm_daemon_runtime_client_t *client, cbm_daemon_runtime_application_token_t request_token, const void *request, uint32_t request_length, uint8_t **response_out, - uint32_t *response_length_out, uint32_t timeout_ms) { + uint32_t *response_length_out, bool *request_sent_out, uint32_t timeout_ms) { if (response_out) { *response_out = NULL; } if (response_length_out) { *response_length_out = 0; } + if (request_sent_out) { + *request_sent_out = false; + } if (!client || !response_out || !response_length_out || request_token == CBM_DAEMON_RUNTIME_APPLICATION_TOKEN_INVALID || timeout_ms == CBM_DAEMON_IPC_WAIT_FOREVER || @@ -3081,6 +3084,9 @@ cbm_daemon_runtime_application_status_t cbm_daemon_runtime_client_application_re request_can_send && cbm_daemon_ipc_send_frame(connection, CBM_DAEMON_FRAME_REQUEST, CBM_DAEMON_RUNTIME_OP_APPLICATION_REQUEST, wire, (uint32_t)wire_length); + if (request_sent_out) { + *request_sent_out = sent; + } bool send_cancel = false; cbm_mutex_lock(&client->state_mutex); client->application_request_sent = sent; @@ -3179,9 +3185,9 @@ cbm_daemon_runtime_application_status_t cbm_daemon_runtime_client_application_re if (!cbm_daemon_runtime_client_application_token_reserve(client, &request_token)) { return CBM_DAEMON_RUNTIME_APPLICATION_TRANSPORT_ERROR; } - return cbm_daemon_runtime_client_application_request_tagged(client, request_token, request, - request_length, response_out, - response_length_out, timeout_ms); + return cbm_daemon_runtime_client_application_request_tagged( + client, request_token, request, request_length, response_out, response_length_out, NULL, + timeout_ms); } bool cbm_daemon_runtime_client_close_begin(cbm_daemon_runtime_client_t *client) { diff --git a/src/daemon/runtime.h b/src/daemon/runtime.h index 4d53da9a4a..15521ac83d 100644 --- a/src/daemon/runtime.h +++ b/src/daemon/runtime.h @@ -399,11 +399,12 @@ cbm_daemon_runtime_application_status_t cbm_daemon_runtime_client_application_re /* Execute using the sole outstanding token returned by token_reserve(). This * is the cancellable frontend path; the legacy helper above reserves - * internally. */ + * internally. request_sent_out reports whether the application frame crossed + * the local transport, including failures while awaiting its response. */ cbm_daemon_runtime_application_status_t cbm_daemon_runtime_client_application_request_tagged( cbm_daemon_runtime_client_t *client, cbm_daemon_runtime_application_token_t request_token, const void *request, uint32_t request_length, uint8_t **response_out, - uint32_t *response_length_out, uint32_t timeout_ms); + uint32_t *response_length_out, bool *request_sent_out, uint32_t timeout_ms); /* Begin a two-phase close without freeing the client. This atomically rejects * future exchanges and interrupts an exchange already in flight. Exactly one diff --git a/src/main.c b/src/main.c index 8a575c44f2..98db00561a 100644 --- a/src/main.c +++ b/src/main.c @@ -3073,38 +3073,54 @@ int main(int argc, char **argv) { cbm_daemon_ipc_endpoint_free(endpoint); if (bootstrap_status != CBM_DAEMON_BOOTSTRAP_CONNECTED || !bootstrap_result.client) { main_report_client_bootstrap_failure(role, &bootstrap_result); + cbm_daemon_ipc_endpoint_free(endpoint); (void)main_version_cohort_close(&client_cohort_lease, &client_cohort_manager); return EXIT_FAILURE; } g_daemon_client = bootstrap_result.client; - if (role == CBM_DAEMON_PROCESS_MCP_CLIENT && - !main_set_client_context(g_daemon_client, NULL, tool_profile, NULL, NULL, - MAIN_CONNECT_TIMEOUT_MS)) { - (void)fprintf(stderr, "codebase-memory-mcp: daemon session context was rejected\n"); - (void)cbm_daemon_runtime_client_close(g_daemon_client, MAIN_CLOSE_TIMEOUT_MS); - g_daemon_client = NULL; - (void)main_version_cohort_close(&client_cohort_lease, &client_cohort_manager); - return EXIT_FAILURE; + char mcp_session_root[MAIN_PATH_CAP] = {0}; + char mcp_allowed_root[MAIN_PATH_CAP] = {0}; + const char *mcp_allowed_root_ptr = NULL; + if (role == CBM_DAEMON_PROCESS_MCP_CLIENT) { + bool context_resolved = + main_session_context(NULL, mcp_session_root, mcp_allowed_root, &mcp_allowed_root_ptr); + cbm_daemon_runtime_application_status_t context_status = + context_resolved + ? cbm_daemon_application_client_set_context(g_daemon_client, mcp_session_root, + mcp_allowed_root_ptr, tool_profile, + NULL, NULL, MAIN_CONNECT_TIMEOUT_MS) + : CBM_DAEMON_RUNTIME_APPLICATION_REJECTED; + if (context_status != CBM_DAEMON_RUNTIME_APPLICATION_OK) { + (void)fprintf(stderr, "codebase-memory-mcp: daemon session context was rejected\n"); + (void)cbm_daemon_runtime_client_close(g_daemon_client, MAIN_CLOSE_TIMEOUT_MS); + g_daemon_client = NULL; + cbm_daemon_ipc_endpoint_free(endpoint); + (void)main_version_cohort_close(&client_cohort_lease, &client_cohort_manager); + return EXIT_FAILURE; + } } /* Persist UI mutations only after the exact-build HELLO succeeds. A * conflicting binary must be observationally read-only: applying its * flags before bootstrap could reconfigure the already-running daemon * even though that client was then rejected. */ + bool mcp_ui_enabled = false; + int mcp_ui_port = 0; + uint8_t mcp_ui_update_mask = 0; if (role == CBM_DAEMON_PROCESS_MCP_CLIENT && cbm_mcp_tool_profile_allows_http(tool_profile)) { - bool ui_enabled = false; - int ui_port = 0; bool explicitly_enabled = false; - uint8_t update_mask = - parse_ui_flags(argc, argv, &ui_enabled, &ui_port, &explicitly_enabled); - if (update_mask != 0 && cbm_daemon_application_client_set_ui_config( - g_daemon_client, update_mask, ui_enabled, ui_port, - MAIN_CONNECT_TIMEOUT_MS) != CBM_DAEMON_RUNTIME_APPLICATION_OK) { + mcp_ui_update_mask = + parse_ui_flags(argc, argv, &mcp_ui_enabled, &mcp_ui_port, &explicitly_enabled); + if (mcp_ui_update_mask != 0 && + cbm_daemon_application_client_set_ui_config( + g_daemon_client, mcp_ui_update_mask, mcp_ui_enabled, mcp_ui_port, + MAIN_CONNECT_TIMEOUT_MS) != CBM_DAEMON_RUNTIME_APPLICATION_OK) { (void)fprintf(stderr, "codebase-memory-mcp: daemon UI configuration update failed\n"); (void)cbm_daemon_runtime_client_close(g_daemon_client, MAIN_CLOSE_TIMEOUT_MS); g_daemon_client = NULL; + cbm_daemon_ipc_endpoint_free(endpoint); (void)main_version_cohort_close(&client_cohort_lease, &client_cohort_manager); return EXIT_FAILURE; } @@ -3119,14 +3135,26 @@ int main(int argc, char **argv) { (void)fprintf(stderr, "codebase-memory-mcp: parent-death watchdog could not start\n"); (void)cbm_daemon_runtime_client_close(g_daemon_client, MAIN_CLOSE_TIMEOUT_MS); g_daemon_client = NULL; + cbm_daemon_ipc_endpoint_free(endpoint); (void)main_version_cohort_close(&client_cohort_lease, &client_cohort_manager); return EXIT_FAILURE; } #endif setup_signal_handlers(); - int result = cbm_daemon_frontend_mcp_run(g_daemon_client, client_cohort_manager, stdin, stdout); + cbm_daemon_frontend_session_config_t frontend_session = { + .bootstrap = bootstrap_config, + .session_root = mcp_session_root, + .allowed_root = mcp_allowed_root_ptr, + .tool_profile = tool_profile, + .ui_update_mask = mcp_ui_update_mask, + .ui_enabled = mcp_ui_enabled, + .ui_port = mcp_ui_port, + }; + int result = cbm_daemon_frontend_mcp_run(g_daemon_client, client_cohort_manager, + &frontend_session, stdin, stdout); g_daemon_client = NULL; /* frontend consumed the handle */ + cbm_daemon_ipc_endpoint_free(endpoint); bool client_cohort_cleanup = main_version_cohort_close(&client_cohort_lease, &client_cohort_manager); atomic_store(&g_shutdown, 1); diff --git a/tests/test_daemon_frontend.c b/tests/test_daemon_frontend.c index ed0e868e90..7df0e3e9ff 100644 --- a/tests/test_daemon_frontend.c +++ b/tests/test_daemon_frontend.c @@ -4,6 +4,7 @@ #include "test_framework.h" #include "test_helpers.h" +#include "daemon/application.h" #include "daemon/frontend.h" #include "daemon/ipc.h" #include "daemon/service.h" @@ -239,8 +240,18 @@ enum { * regression black-box: it must remain valid if the exact capacity changes * while still proving that overload cannot hide an already-pending EOF. */ FRONTEND_EOF_TEST_OVERFLOW_MESSAGES = 32, + /* Hold replacement context past the production frontend's 15 second + * no-progress EOF + drain so shutdown deterministically overlaps recovery. */ + FRONTEND_RECOVERY_EOF_RELEASE_MS = 16000, }; +typedef enum { + FRONTEND_RECOVERY_RETRY = 0, + FRONTEND_RECOVERY_CANCEL = 1, + FRONTEND_RECOVERY_CLEAN_EOF = 2, +} frontend_recovery_mode_t; + typedef struct { atomic_int requests; atomic_int session_cancels; @@ -268,6 +279,7 @@ typedef struct { typedef struct { int fd; bool overflow; + frontend_recovery_mode_t recovery_mode; frontend_eof_application_context_t *application; atomic_bool finished; atomic_bool succeeded; @@ -275,7 +287,9 @@ typedef struct { typedef struct { char conflict_log[FRONTEND_TEST_PATH_CAP]; + char build_fingerprint[CBM_DAEMON_BUILD_FINGERPRINT_SIZE]; cbm_daemon_ipc_endpoint_t *endpoint; + cbm_daemon_build_identity_t identity; cbm_version_cohort_manager_t *manager; cbm_daemon_runtime_service_t *service; cbm_daemon_runtime_client_t *client; @@ -337,16 +351,22 @@ static cbm_daemon_runtime_application_status_t frontend_eof_application_request( } /* Released: fall through and answer normally like every later item. */ } - if (request_length == 0) { + if (request_length == 0 || request[0] == CBM_DAEMON_APPLICATION_REQUEST_SET_CONTEXT) { return CBM_DAEMON_RUNTIME_APPLICATION_OK; } - uint8_t *response = malloc(request_length); + const uint8_t *response_source = request; + uint32_t response_length = request_length; + if (request[0] == CBM_DAEMON_APPLICATION_REQUEST_MCP) { + response_source++; + response_length--; + } + uint8_t *response = malloc(response_length); if (!response) { return CBM_DAEMON_RUNTIME_APPLICATION_HANDLER_ERROR; } - memcpy(response, request, request_length); + memcpy(response, response_source, response_length); *response_out = response; - *response_length_out = request_length; + *response_length_out = response_length; if (context->request_observed_fd >= 0) { const char marker = 'Q'; (void)write(context->request_observed_fd, &marker, 1); @@ -454,6 +474,86 @@ static void *frontend_eof_writer(void *opaque) { return NULL; } +static const char frontend_recovery_request[] = + "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"tools/list\",\"params\":{}}\n"; +static const char frontend_recovery_cancel[] = + "{\"jsonrpc\":\"2.0\",\"method\":\"notifications/cancelled\"," + "\"params\":{\"requestId\":1}}\n"; +static const char frontend_recovery_followup[] = + "{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"tools/list\",\"params\":{}}\n"; + +static void *frontend_recovery_writer(void *opaque) { + frontend_eof_writer_t *writer = opaque; + bool ok = frontend_eof_write_all(writer->fd, frontend_recovery_request, + sizeof(frontend_recovery_request) - 1U); + uint64_t deadline = cbm_now_ms() + FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS; + if (writer->recovery_mode != FRONTEND_RECOVERY_RETRY) { + while (ok && + !atomic_load_explicit(&writer->application->first_request_started, + memory_order_acquire) && + cbm_now_ms() < deadline) { + cbm_usleep(1000); + } + ok = ok && atomic_load_explicit(&writer->application->first_request_started, + memory_order_acquire); + } + if (writer->recovery_mode == FRONTEND_RECOVERY_CANCEL) { + ok = ok && frontend_eof_write_all(writer->fd, frontend_recovery_cancel, + sizeof(frontend_recovery_cancel) - 1U); + /* Keep context blocked until the reader routes the cancellation. */ + cbm_usleep(100000); + atomic_store_explicit(&writer->application->release_first_request, true, + memory_order_release); + ok = ok && frontend_eof_write_all(writer->fd, frontend_recovery_followup, + sizeof(frontend_recovery_followup) - 1U); + } else if (writer->recovery_mode == FRONTEND_RECOVERY_CLEAN_EOF) { + ok = close(writer->fd) == 0 && ok; + writer->fd = -1; + cbm_usleep(FRONTEND_RECOVERY_EOF_RELEASE_MS * 1000U); + atomic_store_explicit(&writer->application->release_first_request, true, + memory_order_release); + } + int expected_requests = writer->recovery_mode == FRONTEND_RECOVERY_CLEAN_EOF ? 1 : 2; + while (ok && + atomic_load_explicit(&writer->application->requests, memory_order_acquire) < + expected_requests && + cbm_now_ms() < deadline) { + cbm_usleep(1000); + } + ok = ok && atomic_load_explicit(&writer->application->requests, memory_order_acquire) == + expected_requests; + if (writer->fd >= 0) { + ok = close(writer->fd) == 0 && ok; + writer->fd = -1; + } + atomic_store_explicit(&writer->succeeded, ok, memory_order_release); + atomic_store_explicit(&writer->finished, true, memory_order_release); + return NULL; +} + +static cbm_daemon_runtime_service_t *frontend_eof_service_start(frontend_eof_fixture_t *fixture) { + cbm_daemon_runtime_application_callbacks_t callbacks = { + .context = &fixture->application, + .session_open = frontend_eof_application_session_open, + .request = frontend_eof_application_request, + .request_cancel = frontend_eof_application_request_cancel, + .session_cancel = frontend_eof_application_session_cancel, + .session_close = frontend_eof_application_session_close, + }; + cbm_daemon_runtime_service_config_t config = { + .endpoint = fixture->endpoint, + .identity = fixture->identity, + .conflict_log_path = fixture->conflict_log, + .conflict_log_cap_bytes = 64U * 1024U, + .max_clients = 4, + .lease_timeout_ms = 5000, + .request_timeout_ms = FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS, + .shutdown_timeout_ms = FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS, + .application = callbacks, + }; + return cbm_daemon_runtime_service_start(&config); +} + static bool frontend_eof_fixture_start(frontend_eof_fixture_t *fixture, const char *parent) { memset(fixture, 0, sizeof(*fixture)); atomic_init(&fixture->application.requests, 0); @@ -467,53 +567,42 @@ static bool frontend_eof_fixture_start(frontend_eof_fixture_t *fixture, const ch fixture->application.request_observed_fd = -1; fixture->application.session_cancel_fd = -1; char key[CBM_DAEMON_KEY_SIZE]; - char build[CBM_DAEMON_BUILD_FINGERPRINT_SIZE]; int log_written = snprintf(fixture->conflict_log, sizeof(fixture->conflict_log), "%s/conflicts.ndjson", parent); if (log_written <= 0 || log_written >= (int)sizeof(fixture->conflict_log) || !cbm_daemon_rendezvous_key(key) || - !cbm_daemon_runtime_process_build_fingerprint((uint64_t)getpid(), build)) { + !cbm_daemon_runtime_process_build_fingerprint((uint64_t)getpid(), + fixture->build_fingerprint)) { return false; } fixture->endpoint = cbm_daemon_ipc_endpoint_new(key, parent); fixture->manager = fixture->endpoint ? cbm_version_cohort_manager_new(fixture->endpoint) : NULL; - cbm_daemon_build_identity_t identity = { + fixture->identity = (cbm_daemon_build_identity_t){ .semantic_version = "2.4.0", - .build_fingerprint = build, + .build_fingerprint = fixture->build_fingerprint, .cache_fingerprint = FRONTEND_TEST_CACHE, .protocol_abi = 3, .store_abi = 11, .feature_abi = 7, }; - cbm_daemon_runtime_application_callbacks_t callbacks = { - .context = &fixture->application, - .session_open = frontend_eof_application_session_open, - .request = frontend_eof_application_request, - .request_cancel = frontend_eof_application_request_cancel, - .session_cancel = frontend_eof_application_session_cancel, - .session_close = frontend_eof_application_session_close, - }; - cbm_daemon_runtime_service_config_t config = { - .endpoint = fixture->endpoint, - .identity = identity, - .conflict_log_path = fixture->conflict_log, - .conflict_log_cap_bytes = 64U * 1024U, - .max_clients = 4, - .lease_timeout_ms = 5000, - .request_timeout_ms = FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS, - .shutdown_timeout_ms = FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS, - .application = callbacks, - }; - fixture->service = fixture->manager ? cbm_daemon_runtime_service_start(&config) : NULL; + fixture->service = fixture->manager ? frontend_eof_service_start(fixture) : NULL; cbm_daemon_runtime_connect_result_t connect_result = {0}; fixture->client = fixture->service - ? cbm_daemon_runtime_client_connect(fixture->endpoint, &identity, + ? cbm_daemon_runtime_client_connect(fixture->endpoint, &fixture->identity, FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS, &connect_result) : NULL; return fixture->client && connect_result.status == CBM_DAEMON_RUNTIME_CONNECT_ACCEPTED; } +static bool frontend_eof_fixture_restart_service(frontend_eof_fixture_t *fixture) { + bool stopped = fixture->service && cbm_daemon_runtime_service_stop( + fixture->service, FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS); + bool freed = stopped && cbm_daemon_runtime_service_free(fixture->service); + fixture->service = freed ? frontend_eof_service_start(fixture) : NULL; + return stopped && freed && fixture->service; +} + static bool frontend_eof_fixture_finish(frontend_eof_fixture_t *fixture) { bool ok = true; if (fixture->client) { @@ -579,7 +668,21 @@ static int frontend_eof_child_run(const char *parent, bool overflow) { cbm_daemon_runtime_client_t *frontend_client = fixture.client; fixture.client = NULL; /* cbm_daemon_frontend_mcp_run consumes it. */ - int result = cbm_daemon_frontend_mcp_run(frontend_client, fixture.manager, input, output); + cbm_daemon_frontend_session_config_t session = { + .bootstrap = + { + .role = CBM_DAEMON_PROCESS_MCP_CLIENT, + .endpoint = fixture.endpoint, + .identity = &fixture.identity, + .executable_path = "unused", + .connect_timeout_ms = FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS, + .startup_timeout_ms = FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS, + }, + .session_root = parent, + .tool_profile = CBM_MCP_TOOL_PROFILE_ALL, + }; + int result = + cbm_daemon_frontend_mcp_run(frontend_client, fixture.manager, &session, input, output); bool joined = cbm_thread_join(&writer_thread) == 0; bool writer_ok = atomic_load_explicit(&writer.finished, memory_order_acquire) && atomic_load_explicit(&writer.succeeded, memory_order_acquire); @@ -660,6 +763,111 @@ static bool frontend_eof_run_isolated(const char *tag, bool overflow) { return child_ok && cleaned; } +static int frontend_recovery_child_run(const char *parent, frontend_recovery_mode_t recovery_mode) { + (void)alarm(FRONTEND_EOF_TEST_CATASTROPHIC_TIMEOUT_S); + frontend_eof_fixture_t fixture; + if (!frontend_eof_fixture_start(&fixture, parent)) { + (void)frontend_eof_fixture_finish(&fixture); + return 73; + } + atomic_store_explicit(&fixture.application.block_first_request, + recovery_mode != FRONTEND_RECOVERY_RETRY, memory_order_release); + cbm_daemon_runtime_client_t *stale_client = fixture.client; + fixture.client = NULL; + bool replacement_ready = frontend_eof_fixture_restart_service(&fixture); + int input_pipe[2] = {-1, -1}; + bool pipe_ready = replacement_ready && pipe(input_pipe) == 0; + FILE *input = pipe_ready ? fdopen(input_pipe[0], "rb") : NULL; + FILE *output = input ? tmpfile() : NULL; + frontend_eof_writer_t writer = { + .fd = input_pipe[1], + .recovery_mode = recovery_mode, + .application = &fixture.application, + }; + atomic_init(&writer.finished, false); + atomic_init(&writer.succeeded, false); + cbm_thread_t writer_thread; + bool writer_started = + output && cbm_thread_create(&writer_thread, 0, frontend_recovery_writer, &writer) == 0; + if (!writer_started && input_pipe[1] >= 0) { + (void)close(input_pipe[1]); + input_pipe[1] = -1; + } + bool request_ready = writer_started; + cbm_daemon_frontend_session_config_t session = { + .bootstrap = + { + .role = CBM_DAEMON_PROCESS_MCP_CLIENT, + .endpoint = fixture.endpoint, + .identity = &fixture.identity, + .executable_path = "unused", + .connect_timeout_ms = FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS, + .startup_timeout_ms = FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS, + }, + .session_root = parent, + .tool_profile = CBM_MCP_TOOL_PROFILE_ALL, + }; + int result = request_ready ? cbm_daemon_frontend_mcp_run(stale_client, fixture.manager, + &session, input, output) + : -1; + stale_client = NULL; /* the frontend consumes the stale or replacement handle */ + bool writer_joined = writer_started && cbm_thread_join(&writer_thread) == 0; + bool writer_ok = writer_joined && + atomic_load_explicit(&writer.finished, memory_order_acquire) && + atomic_load_explicit(&writer.succeeded, memory_order_acquire); + char response[512]; + memset(response, 0, sizeof(response)); + bool response_ready = result == 0 && fflush(output) == 0 && fseek(output, 0, SEEK_SET) == 0; + size_t response_length = response_ready ? fread(response, 1, sizeof(response) - 1U, output) : 0; + bool response_exact = false; + if (recovery_mode == FRONTEND_RECOVERY_CANCEL) { + response_exact = strstr(response, "\"id\":1") && strstr(response, "\"code\":-32800") && + strstr(response, frontend_recovery_followup); + } else if (recovery_mode == FRONTEND_RECOVERY_CLEAN_EOF) { + response_exact = response_length == 0; + } else { + response_exact = response_length == sizeof(frontend_recovery_request) - 1U && + memcmp(response, frontend_recovery_request, + sizeof(frontend_recovery_request) - 1U) == 0; + } + int requests = atomic_load_explicit(&fixture.application.requests, memory_order_acquire); + int expected_requests = recovery_mode == FRONTEND_RECOVERY_CLEAN_EOF ? 1 : 2; + bool input_closed = input && fclose(input) == 0; + bool output_closed = output && fclose(output) == 0; + bool fixture_closed = frontend_eof_fixture_finish(&fixture); + return replacement_ready && request_ready && result == 0 && response_exact && + requests == expected_requests && writer_ok && input_closed && output_closed && + fixture_closed + ? 0 + : 74; +} + +static bool frontend_recovery_run_isolated(frontend_recovery_mode_t recovery_mode) { + char parent[FRONTEND_TEST_PATH_CAP]; + int written = snprintf(parent, sizeof(parent), "%s/cbm-frontend-recovery-XXXXXX", cbm_tmpdir()); + if (written <= 0 || written >= (int)sizeof(parent) || !cbm_mkdtemp(parent)) { + return false; + } + pid_t child = fork(); + if (child == 0) { + (void)signal(SIGALRM, SIG_DFL); + _exit(frontend_recovery_child_run(parent, recovery_mode)); + } + int status = 0; + pid_t waited; + do { + waited = child > 0 ? waitpid(child, &status, 0) : -1; + } while (waited < 0 && errno == EINTR); + bool cleaned = th_rmtree(parent) == 0; + bool child_ok = child > 0 && waited == child && WIFEXITED(status) && WEXITSTATUS(status) == 0; + if (!child_ok && child > 0 && waited == child) { + (void)fprintf(stderr, "frontend recovery fixture failed: status=0x%x exit=%d signal=%d\n", + status, WIFEXITED(status) ? WEXITSTATUS(status) : -1, + WIFSIGNALED(status) ? WTERMSIG(status) : 0); + } + return child_ok && cleaned; +} + static void frontend_test_release_lease(cbm_version_cohort_lease_t **lease) { while (lease && *lease && cbm_version_cohort_lease_release(lease) != CBM_PRIVATE_FILE_LOCK_OK) { cbm_usleep(1000); @@ -920,7 +1128,20 @@ static int frontend_backpressure_frontend_run(const char *parent, int input_fd, return 81; } - int result = cbm_daemon_frontend_mcp_run(client, manager, input, output); + cbm_daemon_frontend_session_config_t session = { + .bootstrap = + { + .role = CBM_DAEMON_PROCESS_MCP_CLIENT, + .endpoint = endpoint, + .identity = &identity, + .executable_path = "unused", + .connect_timeout_ms = FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS, + .startup_timeout_ms = FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS, + }, + .session_root = parent, + .tool_profile = CBM_MCP_TOOL_PROFILE_ALL, + }; + int result = cbm_daemon_frontend_mcp_run(client, manager, &session, input, output); bool joined = cbm_thread_join(&writer_thread) == 0; bool writer_ok = atomic_load_explicit(&writer.finished, memory_order_acquire) && atomic_load_explicit(&writer.succeeded, memory_order_acquire); @@ -1397,8 +1618,21 @@ TEST(daemon_frontend_maintenance_exits_while_stdio_reader_is_blocked) { if (!announced) { _exit(70); } + cbm_daemon_frontend_session_config_t session = { + .bootstrap = + { + .role = CBM_DAEMON_PROCESS_MCP_CLIENT, + .endpoint = endpoint, + .identity = &identity, + .executable_path = "unused", + .connect_timeout_ms = FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS, + .startup_timeout_ms = FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS, + }, + .session_root = fixture.parent, + .tool_profile = CBM_MCP_TOOL_PROFILE_ALL, + }; int result = cbm_daemon_frontend_mcp_run((cbm_daemon_runtime_client_t *)(uintptr_t)1, - manager, input, output); + manager, &session, input, output); (void)result; _exit(71); } @@ -1612,6 +1846,27 @@ TEST(daemon_frontend_over_capacity_input_backpressures_without_loss) { PASS(); } +/* A request rejected before its application frame reaches the daemon is safe + * to replay. The frontend must replace its stale authenticated client, restore + * the session, reserve a token from the new connection, and return the one + * response without closing Codex's stdio transport. */ +TEST(daemon_frontend_recovers_one_unsent_request_after_daemon_replacement) { + ASSERT_TRUE(frontend_recovery_run_isolated(FRONTEND_RECOVERY_RETRY)); + PASS(); +} + +/* Context-replay cancellation must not strand a token or break the next request. */ +TEST(daemon_frontend_recovery_cancellation_preserves_next_request) { + ASSERT_TRUE(frontend_recovery_run_isolated(FRONTEND_RECOVERY_CANCEL)); + PASS(); +} + +/* Clean EOF during context replay must preserve worker-owned client cleanup. */ +TEST(daemon_frontend_clean_eof_during_recovery_preserves_client_ownership) { + ASSERT_TRUE(frontend_recovery_run_isolated(FRONTEND_RECOVERY_CLEAN_EOF)); + PASS(); +} + /* Clean EOF is the normal MCP-session ownership boundary. Accepted work gets a * bounded drain opportunity; work still active at the deadline is cancelled * with that exact session, while the frontend itself reports a clean close. */ @@ -1669,6 +1924,9 @@ SUITE(daemon_frontend) { RUN_TEST(daemon_local_participant_monitor_cancels_then_bounds_active_operation); RUN_TEST(daemon_local_participant_monitor_allows_supervisor_containment_window); RUN_TEST(daemon_frontend_over_capacity_input_backpressures_without_loss); + RUN_TEST(daemon_frontend_recovers_one_unsent_request_after_daemon_replacement); + RUN_TEST(daemon_frontend_recovery_cancellation_preserves_next_request); + RUN_TEST(daemon_frontend_clean_eof_during_recovery_preserves_client_ownership); RUN_TEST(daemon_frontend_eof_drain_timeout_cancels_and_returns_success); RUN_TEST(daemon_frontend_stdout_backpressure_eof_fail_stops_and_cancels_session); RUN_TEST(daemon_frontend_stdout_backpressure_maintenance_stops_and_cancels_session); diff --git a/tests/test_daemon_runtime.c b/tests/test_daemon_runtime.c index 991506a74b..b79d6c463b 100644 --- a/tests/test_daemon_runtime.c +++ b/tests/test_daemon_runtime.c @@ -907,7 +907,8 @@ static void *runtime_application_client_request_thread(void *opaque) { call->status = call->tagged ? cbm_daemon_runtime_client_application_request_tagged( call->client, call->request_token, call->request, call->request_length, - &call->response, &call->response_length, RUNTIME_TEST_TIMEOUT_MS) + &call->response, &call->response_length, NULL, + RUNTIME_TEST_TIMEOUT_MS) : cbm_daemon_runtime_client_application_request( call->client, call->request, call->request_length, &call->response, &call->response_length, RUNTIME_TEST_TIMEOUT_MS); @@ -3161,7 +3162,7 @@ TEST(daemon_runtime_presend_request_cancel_is_sticky_and_nonterminal) { cancel = cbm_daemon_runtime_client_application_cancel(client, request_token); cancelled_status = cbm_daemon_runtime_client_application_request_tagged( client, request_token, blocking_request, (uint32_t)sizeof(blocking_request), - &cancelled_response, &cancelled_response_length, RUNTIME_TEST_TIMEOUT_MS); + &cancelled_response, &cancelled_response_length, NULL, RUNTIME_TEST_TIMEOUT_MS); } if (cancelled_status == CBM_DAEMON_RUNTIME_APPLICATION_CANCELLED) { next_status = cbm_daemon_runtime_client_application_request( @@ -3218,8 +3219,10 @@ TEST(daemon_runtime_allows_only_one_unstarted_application_token) { bool first_reserved = false; bool duplicate_reservation_rejected = false; bool first_exact = false; + bool first_sent = false; bool second_reserved = false; bool second_exact = false; + bool second_sent = false; bool closed = false; bool exited = false; @@ -3239,7 +3242,7 @@ TEST(daemon_runtime_allows_only_one_unstarted_application_token) { cbm_daemon_runtime_application_status_t status = cbm_daemon_runtime_client_application_request_tagged( client, first_token, first_request, (uint32_t)sizeof(first_request), &response, - &response_length, RUNTIME_TEST_TIMEOUT_MS); + &response_length, &first_sent, RUNTIME_TEST_TIMEOUT_MS); first_exact = status == CBM_DAEMON_RUNTIME_APPLICATION_OK && response_length == sizeof(first_request) && response && memcmp(response, first_request, sizeof(first_request)) == 0; @@ -3255,7 +3258,7 @@ TEST(daemon_runtime_allows_only_one_unstarted_application_token) { cbm_daemon_runtime_application_status_t status = cbm_daemon_runtime_client_application_request_tagged( client, second_token, second_request, (uint32_t)sizeof(second_request), &response, - &response_length, RUNTIME_TEST_TIMEOUT_MS); + &response_length, &second_sent, RUNTIME_TEST_TIMEOUT_MS); second_exact = status == CBM_DAEMON_RUNTIME_APPLICATION_OK && response_length == sizeof(second_request) && response && memcmp(response, second_request, sizeof(second_request)) == 0; @@ -3274,9 +3277,11 @@ TEST(daemon_runtime_allows_only_one_unstarted_application_token) { ASSERT_TRUE(first_reserved); ASSERT_TRUE(duplicate_reservation_rejected); ASSERT_TRUE(first_exact); + ASSERT_TRUE(first_sent); ASSERT_TRUE(second_reserved); ASSERT_EQ(second_token, first_token + 1U); ASSERT_TRUE(second_exact); + ASSERT_TRUE(second_sent); ASSERT_TRUE(closed); ASSERT_TRUE(exited); ASSERT_EQ(atomic_load(&context.requests), 2); @@ -3376,6 +3381,10 @@ TEST(daemon_runtime_close_begin_retains_storage_and_rejects_late_exchange) { uint8_t *response = NULL; uint32_t response_length = 0; cbm_daemon_runtime_application_status_t status = CBM_DAEMON_RUNTIME_APPLICATION_OK; + cbm_daemon_runtime_application_token_t request_token = + CBM_DAEMON_RUNTIME_APPLICATION_TOKEN_INVALID; + bool token_reserved = false; + bool request_sent = true; bool close_begun = false; bool duplicate_begin_rejected = false; bool close_acknowledged = false; @@ -3386,6 +3395,8 @@ TEST(daemon_runtime_close_begin_retains_storage_and_rejects_late_exchange) { RUNTIME_TEST_TIMEOUT_MS, &result); } if (client) { + token_reserved = + cbm_daemon_runtime_client_application_token_reserve(client, &request_token); close_begun = cbm_daemon_runtime_client_close_begin(client); duplicate_begin_rejected = close_begun && !cbm_daemon_runtime_client_close_begin(client); } @@ -3393,9 +3404,9 @@ TEST(daemon_runtime_close_begin_retains_storage_and_rejects_late_exchange) { /* Deterministically models the frontend boundary where close begins * after a worker claims an item but before it enters the runtime API. * The retained allocation must reject the call without touching IPC. */ - status = cbm_daemon_runtime_client_application_request( - client, request, (uint32_t)sizeof(request), &response, &response_length, - RUNTIME_TEST_TIMEOUT_MS); + status = cbm_daemon_runtime_client_application_request_tagged( + client, request_token, request, (uint32_t)sizeof(request), &response, &response_length, + &request_sent, RUNTIME_TEST_TIMEOUT_MS); close_acknowledged = cbm_daemon_runtime_client_close_finish(client, RUNTIME_TEST_TIMEOUT_MS); client = NULL; @@ -3413,11 +3424,13 @@ TEST(daemon_runtime_close_begin_retains_storage_and_rejects_late_exchange) { runtime_test_fixture_finish(&fixture); ASSERT_TRUE(started); + ASSERT_TRUE(token_reserved); ASSERT_TRUE(close_begun); ASSERT_TRUE(duplicate_begin_rejected); ASSERT_EQ(status, CBM_DAEMON_RUNTIME_APPLICATION_TRANSPORT_ERROR); ASSERT_NULL(response); ASSERT_EQ(response_length, 0); + ASSERT_FALSE(request_sent); ASSERT_TRUE(close_acknowledged); ASSERT_TRUE(exited); ASSERT_EQ(atomic_load(&context.opened), 1); From b9da6dca03126ef0c8cacccb3321dd5d69ccf34d Mon Sep 17 00:00:00 2001 From: Pxx500 Date: Fri, 21 Aug 2026 18:44:15 +0200 Subject: [PATCH 2/5] test(frontend): widen recovery hang guard for MSan Signed-off-by: Pxx500 --- src/main.c | 1 - tests/test_daemon_frontend.c | 5 ++++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/src/main.c b/src/main.c index 98db00561a..5547d79b15 100644 --- a/src/main.c +++ b/src/main.c @@ -3070,7 +3070,6 @@ int main(int argc, char **argv) { cbm_daemon_bootstrap_result_t bootstrap_result; cbm_daemon_bootstrap_status_t bootstrap_status = main_client_bootstrap_with_upgrade(&bootstrap_config, &bootstrap_result, true); - cbm_daemon_ipc_endpoint_free(endpoint); if (bootstrap_status != CBM_DAEMON_BOOTSTRAP_CONNECTED || !bootstrap_result.client) { main_report_client_bootstrap_failure(role, &bootstrap_result); cbm_daemon_ipc_endpoint_free(endpoint); diff --git a/tests/test_daemon_frontend.c b/tests/test_daemon_frontend.c index 7df0e3e9ff..96889ad330 100644 --- a/tests/test_daemon_frontend.c +++ b/tests/test_daemon_frontend.c @@ -231,6 +231,9 @@ enum { * dev machine (see the index-supervisor calibration). */ FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS = 30000, FRONTEND_EOF_TEST_CATASTROPHIC_TIMEOUT_S = 90, + /* Recovery can spend most of the generic guard restarting an instrumented + * daemon before the clean-EOF case enters its deliberate 15s drain. */ + FRONTEND_RECOVERY_TEST_CATASTROPHIC_TIMEOUT_S = 120, FRONTEND_BACKPRESSURE_MESSAGE_BYTES = 2 * 1024 * 1024, FRONTEND_BACKPRESSURE_FRONTEND_TIMEOUT_S = 90, FRONTEND_BACKPRESSURE_DAEMON_TIMEOUT_S = 90, @@ -764,7 +767,7 @@ static bool frontend_eof_run_isolated(const char *tag, bool overflow) { } static int frontend_recovery_child_run(const char *parent, frontend_recovery_mode_t recovery_mode) { - (void)alarm(FRONTEND_EOF_TEST_CATASTROPHIC_TIMEOUT_S); + (void)alarm(FRONTEND_RECOVERY_TEST_CATASTROPHIC_TIMEOUT_S); frontend_eof_fixture_t fixture; if (!frontend_eof_fixture_start(&fixture, parent)) { (void)frontend_eof_fixture_finish(&fixture); From 85482e3c599d454d76ccef5e862a5dc6c38a898e Mon Sep 17 00:00:00 2001 From: Pxx500 Date: Mon, 7 Sep 2026 17:30:13 +0200 Subject: [PATCH 3/5] test(frontend): address recovery review feedback Signed-off-by: Pxx500 --- src/daemon/frontend.c | 16 ++++++++ src/daemon/frontend.h | 12 ++++-- tests/test_daemon_frontend.c | 72 +++++++++++++++++++++++++----------- 3 files changed, 75 insertions(+), 25 deletions(-) diff --git a/src/daemon/frontend.c b/src/daemon/frontend.c index 92b0326867..d65284bd7a 100644 --- a/src/daemon/frontend.c +++ b/src/daemon/frontend.c @@ -119,6 +119,7 @@ static atomic_bool frontend_test_monitor_is_waiting = ATOMIC_VAR_INIT(false); static atomic_uint_fast64_t frontend_test_monitor_observation_count = ATOMIC_VAR_INIT(0); static atomic_uint_fast64_t frontend_test_worker_observation_count = ATOMIC_VAR_INIT(0); static atomic_uint_fast64_t frontend_test_worker_idle_count = ATOMIC_VAR_INIT(0); +static atomic_uint_fast64_t frontend_test_routed_cancellation_count = ATOMIC_VAR_INIT(0); void cbm_daemon_frontend_test_observer_reset(bool hold_monitor) { atomic_store_explicit(&frontend_test_monitor_observation_count, 0, memory_order_release); @@ -147,6 +148,10 @@ uint64_t cbm_daemon_frontend_test_worker_observations(void) { uint64_t cbm_daemon_frontend_test_worker_idle_cycles(void) { return atomic_load_explicit(&frontend_test_worker_idle_count, memory_order_acquire); } + +uint64_t cbm_daemon_frontend_test_routed_cancellations(void) { + return atomic_load_explicit(&frontend_test_routed_cancellation_count, memory_order_acquire); +} #endif static cbm_version_cohort_maintenance_presence_t frontend_observe_maintenance( @@ -478,6 +483,12 @@ static frontend_cancellation_route_t frontend_route_cancellation(frontend_state_ } cbm_mutex_unlock(&state->mutex); cbm_jsonrpc_request_free(&request); +#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS + if (route == FRONTEND_CANCELLATION_ACTIVE || route == FRONTEND_CANCELLATION_QUEUED) { + atomic_fetch_add_explicit(&frontend_test_routed_cancellation_count, 1, + memory_order_release); + } +#endif return route; } @@ -531,6 +542,11 @@ static bool frontend_recover_client(frontend_state_t *state, frontend_item_t *it } cbm_daemon_bootstrap_result_t bootstrap = {0}; + /* Bootstrap can report launch failures on stderr from this worker. + * Keep that diagnostic + * to explain failed recovery. It never writes MCP + * stdout, but blocked stderr can delay + * exit. Do not add more logging here. */ cbm_daemon_bootstrap_status_t bootstrap_status = cbm_daemon_bootstrap_execute(&state->session->bootstrap, &bootstrap); cbm_daemon_runtime_client_t *replacement = diff --git a/src/daemon/frontend.h b/src/daemon/frontend.h index 10255687fd..9ee193be89 100644 --- a/src/daemon/frontend.h +++ b/src/daemon/frontend.h @@ -14,9 +14,11 @@ typedef struct cbm_daemon_maintenance_monitor cbm_daemon_maintenance_monitor_t; -/* Everything the stdio frontend must replay when its authenticated daemon - * connection is replaced. Pointer fields are borrowed for the complete - * frontend call. */ +/* Connection settings replayed when the authenticated daemon is replaced. + * Background activation + * from initialize is not restored here. + * Pointer fields are borrowed for the complete frontend + * call. */ typedef struct { cbm_daemon_bootstrap_config_t bootstrap; const char *session_root; @@ -67,6 +69,10 @@ bool cbm_daemon_frontend_test_monitor_waiting(void); uint64_t cbm_daemon_frontend_test_monitor_observations(void); uint64_t cbm_daemon_frontend_test_worker_observations(void); uint64_t cbm_daemon_frontend_test_worker_idle_cycles(void); +/* Counts matching cancellations marked under the frontend mutex. + * Tests snapshot this monotonic + * count before sending a notification. */ +uint64_t cbm_daemon_frontend_test_routed_cancellations(void); #endif /* Takes ownership of client and borrows cohort_manager for the complete call. diff --git a/tests/test_daemon_frontend.c b/tests/test_daemon_frontend.c index 96889ad330..acdaf05127 100644 --- a/tests/test_daemon_frontend.c +++ b/tests/test_daemon_frontend.c @@ -39,6 +39,8 @@ typedef struct { typedef struct { frontend_maintenance_fixture_t maintenance; char conflict_log[FRONTEND_TEST_PATH_CAP]; + char build_fingerprint[CBM_DAEMON_BUILD_FINGERPRINT_SIZE]; + cbm_daemon_build_identity_t identity; cbm_daemon_runtime_service_t *service; cbm_daemon_runtime_client_t *client; } frontend_idle_fixture_t; @@ -46,6 +48,7 @@ typedef struct { typedef struct { cbm_daemon_runtime_client_t *client; cbm_version_cohort_manager_t *manager; + const cbm_daemon_frontend_session_config_t *session; FILE *input; FILE *output; int result; @@ -99,22 +102,22 @@ static cbm_daemon_runtime_application_status_t frontend_idle_request( return CBM_DAEMON_RUNTIME_APPLICATION_OK; } -static void frontend_idle_request_cancel( - void *context, cbm_daemon_runtime_application_session_t *session, - cbm_daemon_runtime_application_token_t request_token) { +static void frontend_idle_request_cancel(void *context, + cbm_daemon_runtime_application_session_t *session, + cbm_daemon_runtime_application_token_t request_token) { (void)context; (void)session; (void)request_token; } -static void frontend_idle_session_cancel( - void *context, cbm_daemon_runtime_application_session_t *session) { +static void frontend_idle_session_cancel(void *context, + cbm_daemon_runtime_application_session_t *session) { (void)context; (void)session; } -static void frontend_idle_session_close( - void *context, cbm_daemon_runtime_application_session_t *session) { +static void frontend_idle_session_close(void *context, + cbm_daemon_runtime_application_session_t *session) { (void)context; (void)session; } @@ -128,22 +131,21 @@ static uint64_t frontend_test_process_id(void) { } static bool frontend_idle_fixture_start(frontend_idle_fixture_t *fixture) { - static const char cache[] = - "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd"; + static const char cache[] = "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd"; memset(fixture, 0, sizeof(*fixture)); if (!frontend_maintenance_fixture_start(&fixture->maintenance, "idle-observer")) { return false; } - char build[CBM_DAEMON_BUILD_FINGERPRINT_SIZE]; int log_written = snprintf(fixture->conflict_log, sizeof(fixture->conflict_log), "%s/conflicts.ndjson", fixture->maintenance.parent); if (log_written <= 0 || log_written >= (int)sizeof(fixture->conflict_log) || - !cbm_daemon_runtime_process_build_fingerprint(frontend_test_process_id(), build)) { + !cbm_daemon_runtime_process_build_fingerprint(frontend_test_process_id(), + fixture->build_fingerprint)) { return false; } - cbm_daemon_build_identity_t identity = { + fixture->identity = (cbm_daemon_build_identity_t){ .semantic_version = "2.4.0", - .build_fingerprint = build, + .build_fingerprint = fixture->build_fingerprint, .cache_fingerprint = cache, .protocol_abi = 3, .store_abi = 11, @@ -159,7 +161,7 @@ static bool frontend_idle_fixture_start(frontend_idle_fixture_t *fixture) { }; cbm_daemon_runtime_service_config_t config = { .endpoint = fixture->maintenance.endpoint, - .identity = identity, + .identity = fixture->identity, .conflict_log_path = fixture->conflict_log, .conflict_log_cap_bytes = 64U * 1024U, .max_clients = 2, @@ -170,10 +172,11 @@ static bool frontend_idle_fixture_start(frontend_idle_fixture_t *fixture) { }; fixture->service = cbm_daemon_runtime_service_start(&config); cbm_daemon_runtime_connect_result_t connect_result = {0}; - fixture->client = fixture->service - ? cbm_daemon_runtime_client_connect(fixture->maintenance.endpoint, &identity, - 30000, &connect_result) - : NULL; + fixture->client = + fixture->service + ? cbm_daemon_runtime_client_connect(fixture->maintenance.endpoint, &fixture->identity, + 30000, &connect_result) + : NULL; return fixture->client && connect_result.status == CBM_DAEMON_RUNTIME_CONNECT_ACCEPTED; } @@ -213,8 +216,8 @@ static int frontend_test_close_fd(int fd) { static void *frontend_idle_run(void *opaque) { frontend_idle_run_t *run = opaque; - run->result = - cbm_daemon_frontend_mcp_run(run->client, run->manager, run->input, run->output); + run->result = cbm_daemon_frontend_mcp_run(run->client, run->manager, run->session, run->input, + run->output); return NULL; } #endif @@ -479,11 +482,13 @@ static void *frontend_eof_writer(void *opaque) { static const char frontend_recovery_request[] = "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"tools/list\",\"params\":{}}\n"; +#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS static const char frontend_recovery_cancel[] = "{\"jsonrpc\":\"2.0\",\"method\":\"notifications/cancelled\"," "\"params\":{\"requestId\":1}}\n"; static const char frontend_recovery_followup[] = "{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"tools/list\",\"params\":{}}\n"; +#endif static void *frontend_recovery_writer(void *opaque) { frontend_eof_writer_t *writer = opaque; @@ -500,16 +505,24 @@ static void *frontend_recovery_writer(void *opaque) { ok = ok && atomic_load_explicit(&writer->application->first_request_started, memory_order_acquire); } +#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS if (writer->recovery_mode == FRONTEND_RECOVERY_CANCEL) { + uint64_t cancellations = cbm_daemon_frontend_test_routed_cancellations(); ok = ok && frontend_eof_write_all(writer->fd, frontend_recovery_cancel, sizeof(frontend_recovery_cancel) - 1U); /* Keep context blocked until the reader routes the cancellation. */ - cbm_usleep(100000); + while (ok && cbm_daemon_frontend_test_routed_cancellations() == cancellations && + cbm_now_ms() < deadline) { + cbm_usleep(1000); + } + ok = ok && cbm_daemon_frontend_test_routed_cancellations() > cancellations; atomic_store_explicit(&writer->application->release_first_request, true, memory_order_release); ok = ok && frontend_eof_write_all(writer->fd, frontend_recovery_followup, sizeof(frontend_recovery_followup) - 1U); - } else if (writer->recovery_mode == FRONTEND_RECOVERY_CLEAN_EOF) { + } +#endif + if (writer->recovery_mode == FRONTEND_RECOVERY_CLEAN_EOF) { ok = close(writer->fd) == 0 && ok; writer->fd = -1; cbm_usleep(FRONTEND_RECOVERY_EOF_RELEASE_MS * 1000U); @@ -1537,9 +1550,20 @@ TEST(daemon_frontend_idle_uses_one_maintenance_observer) { bool pipe_ready = fixture_ready && cbm_pipe(input_pipe) == 0; FILE *input = pipe_ready ? frontend_test_fdopen_read(input_pipe[0]) : NULL; FILE *output = input ? tmpfile() : NULL; + cbm_daemon_frontend_session_config_t session = { + .bootstrap = {.role = CBM_DAEMON_PROCESS_MCP_CLIENT, + .endpoint = fixture.maintenance.endpoint, + .identity = &fixture.identity, + .executable_path = "unused", + .connect_timeout_ms = 30000, + .startup_timeout_ms = 30000}, + .session_root = fixture.maintenance.parent, + .tool_profile = CBM_MCP_TOOL_PROFILE_ALL, + }; frontend_idle_run_t run = { .client = fixture.client, .manager = fixture.maintenance.manager, + .session = &session, .input = input, .output = output, .result = -1, @@ -1859,10 +1883,12 @@ TEST(daemon_frontend_recovers_one_unsent_request_after_daemon_replacement) { } /* Context-replay cancellation must not strand a token or break the next request. */ +#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS TEST(daemon_frontend_recovery_cancellation_preserves_next_request) { ASSERT_TRUE(frontend_recovery_run_isolated(FRONTEND_RECOVERY_CANCEL)); PASS(); } +#endif /* Clean EOF during context replay must preserve worker-owned client cleanup. */ TEST(daemon_frontend_clean_eof_during_recovery_preserves_client_ownership) { @@ -1928,7 +1954,9 @@ SUITE(daemon_frontend) { RUN_TEST(daemon_local_participant_monitor_allows_supervisor_containment_window); RUN_TEST(daemon_frontend_over_capacity_input_backpressures_without_loss); RUN_TEST(daemon_frontend_recovers_one_unsent_request_after_daemon_replacement); +#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS RUN_TEST(daemon_frontend_recovery_cancellation_preserves_next_request); +#endif RUN_TEST(daemon_frontend_clean_eof_during_recovery_preserves_client_ownership); RUN_TEST(daemon_frontend_eof_drain_timeout_cancels_and_returns_success); RUN_TEST(daemon_frontend_stdout_backpressure_eof_fail_stops_and_cancels_session); From 71ac6ecc15d3ab3ebcd31f0e467071a506006ca2 Mon Sep 17 00:00:00 2001 From: Pxx500 Date: Sat, 26 Sep 2026 23:01:11 +0200 Subject: [PATCH 4/5] fix(mcp): reuse response cleanup in frontend worker Signed-off-by: Pxx500 --- src/daemon/frontend.c | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/src/daemon/frontend.c b/src/daemon/frontend.c index d65284bd7a..a4d4f88467 100644 --- a/src/daemon/frontend.c +++ b/src/daemon/frontend.c @@ -580,6 +580,12 @@ static bool frontend_recover_client(frontend_state_t *state, frontend_item_t *it return usable; } +static void frontend_response_reset(uint8_t **response, uint32_t *response_length) { + free(*response); + *response = NULL; + *response_length = 0; +} + static void *frontend_worker(void *opaque) { frontend_state_t *state = opaque; for (;;) { @@ -606,9 +612,7 @@ static void *frontend_worker(void *opaque) { state->client, item.request_token, item.message, &response, &response_length, &request_sent, FRONTEND_REQUEST_TIMEOUT_MS); if (status == CBM_DAEMON_RUNTIME_APPLICATION_TRANSPORT_ERROR && !request_sent) { - free(response); - response = NULL; - response_length = 0; + frontend_response_reset(&response, &response_length); bool cancelled = false; if (frontend_recover_client(state, &item, &cancelled)) { if (cancelled) { @@ -632,7 +636,7 @@ static void *frontend_worker(void *opaque) { item.content_length_framed); } } - free(response); + frontend_response_reset(&response, &response_length); bool expected_stop = failed && frontend_should_stop(state); frontend_end_request(state, failed && !expected_stop); frontend_item_free(&item); From bf30a1aaf01d01f750fdfa695baec34ed977dd53 Mon Sep 17 00:00:00 2001 From: Martin Vogel Date: Sun, 1 Nov 2026 23:48:31 +0100 Subject: [PATCH 5/5] fix(daemon): replay the session within the startup budget The recovery path replays the session context and UI settings on the replacement daemon. It waited on the bootstrap connect budget, which only bounds a probe (1 s in production), while main.c gives the first set_context and set_ui_config the startup budget since eb294298e. A replacement daemon whose UI save runs longer than a probe would end an otherwise good recovery. frontend_configure_client now waits on startup_timeout_ms. A test seam records the budget the replay used, and the recovery fixture gives the bootstrap distinct connect and startup budgets and exits 75 when the replay waited on the wrong one. With the old budget exactly the three recovery tests fail with exit 75; with the fix daemon_frontend is 17/17. Signed-off-by: Martin Vogel --- src/daemon/frontend.c | 14 +++++++++++++- src/daemon/frontend.h | 4 ++++ tests/test_daemon_frontend.c | 15 ++++++++++++++- 3 files changed, 31 insertions(+), 2 deletions(-) diff --git a/src/daemon/frontend.c b/src/daemon/frontend.c index 123ec00e47..e76c920b6c 100644 --- a/src/daemon/frontend.c +++ b/src/daemon/frontend.c @@ -120,6 +120,7 @@ static atomic_uint_fast64_t frontend_test_monitor_observation_count = ATOMIC_VAR static atomic_uint_fast64_t frontend_test_worker_observation_count = ATOMIC_VAR_INIT(0); static atomic_uint_fast64_t frontend_test_worker_idle_count = ATOMIC_VAR_INIT(0); static atomic_uint_fast64_t frontend_test_routed_cancellation_count = ATOMIC_VAR_INIT(0); +static atomic_uint_fast32_t frontend_test_replay_budget_ms = ATOMIC_VAR_INIT(0); void cbm_daemon_frontend_test_observer_reset(bool hold_monitor) { atomic_store_explicit(&frontend_test_monitor_observation_count, 0, memory_order_release); @@ -152,6 +153,10 @@ uint64_t cbm_daemon_frontend_test_worker_idle_cycles(void) { uint64_t cbm_daemon_frontend_test_routed_cancellations(void) { return atomic_load_explicit(&frontend_test_routed_cancellation_count, memory_order_acquire); } + +uint32_t cbm_daemon_frontend_test_replay_timeout_ms(void) { + return (uint32_t)atomic_load_explicit(&frontend_test_replay_budget_ms, memory_order_acquire); +} #endif static cbm_version_cohort_maintenance_presence_t frontend_observe_maintenance( @@ -528,7 +533,14 @@ static cbm_daemon_runtime_cancel_result_t frontend_cancel_active(frontend_state_ static bool frontend_configure_client(cbm_daemon_runtime_client_t *client, const cbm_daemon_frontend_session_config_t *session) { - uint32_t timeout_ms = session->bootstrap.connect_timeout_ms; + /* The replay restores what startup set, so it waits within the startup + * budget like the first set_context and set_ui_config in main.c. The + * connect budget only bounds a probe (1 s in production): a replacement + * daemon whose UI save runs longer would end an otherwise good recovery. */ + uint32_t timeout_ms = session->bootstrap.startup_timeout_ms; +#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS + atomic_store_explicit(&frontend_test_replay_budget_ms, timeout_ms, memory_order_release); +#endif if (cbm_daemon_application_client_set_context( client, session->session_root, session->allowed_root, session->tool_profile, NULL, NULL, timeout_ms) != CBM_DAEMON_RUNTIME_APPLICATION_OK) { diff --git a/src/daemon/frontend.h b/src/daemon/frontend.h index 8f59ec28b5..8ee33d2765 100644 --- a/src/daemon/frontend.h +++ b/src/daemon/frontend.h @@ -79,6 +79,10 @@ uint64_t cbm_daemon_frontend_test_worker_idle_cycles(void); * Tests snapshot this monotonic * count before sending a notification. */ uint64_t cbm_daemon_frontend_test_routed_cancellations(void); +/* Budget in ms that the most recent session replay gave its context and UI + * settings requests. Recovery tests use distinct connect and startup budgets + * and compare. */ +uint32_t cbm_daemon_frontend_test_replay_timeout_ms(void); #endif /* Takes ownership of client and borrows cohort_manager for the complete call. diff --git a/tests/test_daemon_frontend.c b/tests/test_daemon_frontend.c index e4ed5e135b..aa49bf4c72 100644 --- a/tests/test_daemon_frontend.c +++ b/tests/test_daemon_frontend.c @@ -250,6 +250,11 @@ enum { * no-progress EOF drain so shutdown deterministically overlaps recovery. */ FRONTEND_RECOVERY_EOF_RELEASE_MS = 16000, + /* Startup budget of the recovery bootstrap. It differs from the connect + * budget (FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS) so the recovery tests can + * tell which one the session replay waited on, and it is the larger one, + * as in production (1 s connect probe, 30 s startup). */ + FRONTEND_RECOVERY_TEST_STARTUP_TIMEOUT_MS = 45000, }; typedef enum { @@ -818,7 +823,7 @@ static int frontend_recovery_child_run(const char *parent, frontend_recovery_mod .identity = &fixture.identity, .executable_path = "unused", .connect_timeout_ms = FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS, - .startup_timeout_ms = FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS, + .startup_timeout_ms = FRONTEND_RECOVERY_TEST_STARTUP_TIMEOUT_MS, }, .session_root = parent, .tool_profile = CBM_MCP_TOOL_PROFILE_ALL, @@ -851,6 +856,14 @@ static int frontend_recovery_child_run(const char *parent, frontend_recovery_mod bool input_closed = input && fclose(input) == 0; bool output_closed = output && fclose(output) == 0; bool fixture_closed = frontend_eof_fixture_finish(&fixture); +#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS + /* The replay restores what startup set within the startup budget. On the + * connect budget (a probe bound) a replacement daemon that answers slower + * than a probe would end an otherwise good recovery. Every mode replays. */ + if (cbm_daemon_frontend_test_replay_timeout_ms() != FRONTEND_RECOVERY_TEST_STARTUP_TIMEOUT_MS) { + return 75; + } +#endif return replacement_ready && request_ready && result == 0 && response_exact && requests == expected_requests && writer_ok && input_closed && output_closed && fixture_closed