nprasad2077/codebase-memory-mcp
1
0 Fork 0
mirror of https://github.com/DeusData/codebase-memory-mcp.git synced 2026-11-04 10:48:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1784 from Pxx500/fix/mcp-frontend-recovery

Browse Source 
fix(mcp): recover unsent requests after daemon loss
This commit is contained in:
Martin Vogel authored and GitHub committed 2026-11-03 18:08:32 +01:00
2 parents 5830d284c5 b42159e227
commit f031c30fea
9 files changed
+651 -139
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/daemon/application.c
+15 -12
View File
@@ -4026,7 +4026,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) {
@@ -4035,13 +4035,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);
@@ -4072,7 +4075,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(
@@ -4200,7 +4203,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;
}
@@ -4216,7 +4219,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(
@@ -4224,7 +4227,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(
@@ -4237,10 +4240,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(
src/daemon/application.h
+3 -2
View File
@@ -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,
src/daemon/frontend.c
+175 -42
View File
@@ -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
@@ -117,6 +119,8 @@ 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);
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);
@@ -145,6 +149,14 @@ 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);
}
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(
@@ -352,6 +364,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 +378,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);
}
@@ -464,12 +478,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;
@@ -478,12 +488,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++) {
@@ -499,9 +506,116 @@ static frontend_cancellation_route_t frontend_route_cancellation(
}
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;
}
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) {
/* 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) {
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};
/* 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 =
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_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 (;;) {
@@ -522,10 +636,25 @@ 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) {
frontend_response_reset(&response, &response_length);
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 {
@@ -537,19 +666,18 @@ 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);
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;
@@ -632,11 +760,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) {
@@ -646,24 +774,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);
@@ -687,7 +824,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;
@@ -699,18 +836,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;
}
}
@@ -725,7 +858,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
@@ -751,8 +884,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;
@@ -777,9 +910,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++) {
src/daemon/frontend.h
+30 -2
View File
@@ -4,14 +4,31 @@
#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 <stdbool.h>
#include <stdio.h>
typedef struct cbm_daemon_maintenance_monitor cbm_daemon_maintenance_monitor_t;
/* 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;
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. */
@@ -58,6 +75,14 @@ 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);
/* 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.
@@ -66,9 +91,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 */
src/daemon/runtime.c
+10 -4
View File
@@ -3391,13 +3391,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 ||
@@ -3453,6 +3456,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;
@@ -3551,9 +3557,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) {
src/daemon/runtime.h
+3 -2
View File
@@ -423,11 +423,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
src/main.c
+41 -16
View File
@@ -3395,9 +3395,9 @@ 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);
(void)main_version_cohort_close(&client_cohort_lease, &client_cohort_manager);
return EXIT_FAILURE;
}
@@ -3408,33 +3408,46 @@ int main(int argc, char **argv) {
* like the bootstrap before them: a 1 s bound on the reply (a probe budget)
* ended the server whenever the daemon's UI save ran long, which the smoke's
* `--ui=true --port` run saw as a binary without UI assets. */
if (role == CBM_DAEMON_PROCESS_MCP_CLIENT &&
!main_set_client_context(g_daemon_client, NULL, tool_profile, NULL, NULL,
bootstrap_config.startup_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, bootstrap_config.startup_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 &&
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, update_mask, ui_enabled, ui_port,
g_daemon_client, mcp_ui_update_mask, mcp_ui_enabled, mcp_ui_port,
bootstrap_config.startup_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;
}
@@ -3451,13 +3464,25 @@ 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;
}
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);
tests/test_daemon_frontend.c
+354 -52
View File
@@ -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"
@@ -38,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;
@@ -45,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;
@@ -98,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;
}
@@ -127,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,
@@ -158,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,
@@ -169,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;
}
@@ -212,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
@@ -230,6 +234,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,
@@ -239,8 +246,23 @@ 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,
/* 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 {
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 +290,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 +298,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 +362,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 +485,96 @@ 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";
#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;
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 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. */
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);
}
#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);
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 +588,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 +689,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 +784,119 @@ 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_RECOVERY_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_RECOVERY_TEST_STARTUP_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);
#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
? 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 +1157,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);
@@ -1334,9 +1584,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,
@@ -1418,8 +1679,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);
}
@@ -1633,6 +1907,29 @@ 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. */
#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) {
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. */
@@ -1691,6 +1988,11 @@ 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);
#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);
RUN_TEST(daemon_frontend_stdout_backpressure_maintenance_stops_and_cancels_session);
tests/test_daemon_runtime.c
+20 -7
View File
@@ -908,7 +908,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);
@@ -3586,7 +3587,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(
@@ -3643,8 +3644,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;
@@ -3664,7 +3667,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;
@@ -3680,7 +3683,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;
@@ -3699,9 +3702,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);
@@ -3801,6 +3806,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;
@@ -3811,6 +3820,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);
}
@@ -3818,9 +3829,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;
@@ -3838,11 +3849,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);