github-featured/deusdata--codebase-memory-mcp
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity
Files
fix/security-strings-json
deusdata--codebase-memory-mcp/tests/test_daemon_application.c
T
Martin Vogel 8018561cfe fix(release): externalize runtime assets and harden VT verification 
Signed-off-by: Martin Vogel <martin.vogel.tech@gmail.com>
2026-08-08 17:35:05 +02:00

5136 lines
224 KiB
C
Raw Permalink Blame History

/*
* test_daemon_application.c — RED contract for daemon-owned frontend state.
*/
#include "test_framework.h"
#include "test_helpers.h"
#include "cli/cli.h"
#include "daemon/application.h"
#include "daemon/application_internal.h"
#include "daemon/ipc.h"
#include "daemon/project_lock.h"
#include "foundation/compat.h"
#include "foundation/compat_fs.h"
#include "foundation/compat_thread.h"
#include "foundation/platform.h"
#include "foundation/sha256.h"
#include "foundation/subprocess.h"
#include "mcp/mcp.h"
#include "mcp/mcp_internal.h"
#include "pipeline/pipeline.h"
#include "store/store.h"
#include "ui/config.h"
#include "watcher/watcher.h"
#include <stdint.h>
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
/* Observation-wait budget. This is a HANG DETECTOR, not a latency
* assertion: every waiter returns the moment its condition holds, so green
* runs never pay it. It must absorb ThreadSanitizer's 5-20x slowdown on a
* loaded 3-core CI runner — at 2000 ms the cancel-delivery wait lost the
* tail of that distribution once in seven otherwise-green TSan rounds. */
enum { APP_TEST_TIMEOUT_MS = 10000, APP_TEST_PATH_CAP = 1024 };
typedef struct {
char runtime_parent[APP_TEST_PATH_CAP];
cbm_daemon_ipc_endpoint_t *endpoint;
cbm_project_lock_manager_t *external_locks;
cbm_project_lock_manager_t *daemon_locks;
cbm_project_lock_manager_t *worker_locks;
cbm_project_lock_manager_t *probe_locks;
} app_project_lock_fixture_t;
static bool app_project_lock_fixture_cleanup(app_project_lock_fixture_t *fixture) {
if (!fixture) {
return false;
}
bool clean = true;
if (fixture->daemon_locks &&
cbm_project_lock_manager_free(&fixture->daemon_locks) != CBM_PRIVATE_FILE_LOCK_OK) {
clean = false;
}
if (fixture->external_locks &&
cbm_project_lock_manager_free(&fixture->external_locks) != CBM_PRIVATE_FILE_LOCK_OK) {
clean = false;
}
if (fixture->worker_locks &&
cbm_project_lock_manager_free(&fixture->worker_locks) != CBM_PRIVATE_FILE_LOCK_OK) {
clean = false;
}
if (fixture->probe_locks &&
cbm_project_lock_manager_free(&fixture->probe_locks) != CBM_PRIVATE_FILE_LOCK_OK) {
clean = false;
}
if (!fixture->daemon_locks && !fixture->external_locks && !fixture->worker_locks &&
!fixture->probe_locks) {
cbm_daemon_ipc_endpoint_free(fixture->endpoint);
fixture->endpoint = NULL;
if (fixture->runtime_parent[0] && th_rmtree(fixture->runtime_parent) != 0) {
clean = false;
}
} else {
clean = false;
}
return clean;
}
static bool app_project_lock_fixture_init(app_project_lock_fixture_t *fixture) {
if (!fixture) {
return false;
}
memset(fixture, 0, sizeof(*fixture));
if (!th_secure_runtime_parent_new(fixture->runtime_parent, sizeof(fixture->runtime_parent),
"app-project-lock")) {
return false;
}
fixture->endpoint = cbm_daemon_ipc_endpoint_new("fedcba9876543210", fixture->runtime_parent);
if (fixture->endpoint) {
fixture->external_locks = cbm_project_lock_manager_new(fixture->endpoint);
fixture->daemon_locks = cbm_project_lock_manager_new(fixture->endpoint);
fixture->worker_locks = cbm_project_lock_manager_new(fixture->endpoint);
fixture->probe_locks = cbm_project_lock_manager_new(fixture->endpoint);
}
if (fixture->endpoint && fixture->external_locks && fixture->daemon_locks &&
fixture->worker_locks && fixture->probe_locks) {
return true;
}
(void)app_project_lock_fixture_cleanup(fixture);
return false;
}
static bool app_project_lock_release(cbm_project_lock_lease_t **lease) {
if (!lease) {
return false;
}
uint64_t deadline = cbm_now_ms() + APP_TEST_TIMEOUT_MS;
while (*lease && cbm_now_ms() < deadline) {
(void)cbm_project_lock_lease_release(lease);
if (*lease) {
cbm_usleep(1000);
}
}
return *lease == NULL;
}
/* Windows may briefly report an overlapping byte-range lock as BUSY while the
* just-completed worker's handle close becomes visible to another manager.
* Accept only that bounded handoff; IO/UNSAFE failures remain immediate. */
static cbm_private_file_lock_status_t app_project_lock_try_acquire_until_released(
cbm_project_lock_manager_t *manager, const char *project,
cbm_project_lock_lease_t **lease_out) {
if (!lease_out) {
return CBM_PRIVATE_FILE_LOCK_UNSAFE;
}
*lease_out = NULL;
uint64_t deadline = cbm_now_ms() + APP_TEST_TIMEOUT_MS;
cbm_private_file_lock_status_t status = CBM_PRIVATE_FILE_LOCK_BUSY;
do {
status = cbm_project_lock_try_acquire(manager, project, lease_out);
if (status != CBM_PRIVATE_FILE_LOCK_BUSY) {
return status;
}
cbm_usleep(1000);
} while (cbm_now_ms() < deadline);
return status;
}
static cbm_daemon_runtime_application_session_t *app_test_open(
const cbm_daemon_runtime_application_callbacks_t *callbacks, cbm_daemon_client_id_t client_id) {
return callbacks->session_open(callbacks->context, client_id, 1234);
}
static atomic_uint_fast64_t g_app_test_request_token = ATOMIC_VAR_INIT(1);
static cbm_daemon_runtime_application_status_t app_test_request_tagged(
const cbm_daemon_runtime_application_callbacks_t *callbacks,
cbm_daemon_runtime_application_session_t *session,
cbm_daemon_runtime_application_token_t request_token, const uint8_t *request,
uint32_t request_length, uint8_t **response_out, uint32_t *response_length_out) {
return callbacks->request(callbacks->context, session, request_token, request, request_length,
response_out, response_length_out);
}
static cbm_daemon_runtime_application_status_t app_test_request(
const cbm_daemon_runtime_application_callbacks_t *callbacks,
cbm_daemon_runtime_application_session_t *session, const uint8_t *request,
uint32_t request_length, uint8_t **response_out, uint32_t *response_length_out) {
cbm_daemon_runtime_application_token_t request_token =
atomic_fetch_add_explicit(&g_app_test_request_token, 1, memory_order_relaxed);
return app_test_request_tagged(callbacks, session, request_token, request, request_length,
response_out, response_length_out);
}
static bool app_test_context_request_options(const char *root, const char *allowed,
cbm_mcp_tool_profile_t tool_profile,
const char *hook_event, const char *hook_dialect,
uint8_t **request_out, uint32_t *length_out) {
size_t root_length = strlen(root);
size_t allowed_length = allowed ? strlen(allowed) : 0;
size_t event_length = hook_event ? strlen(hook_event) : 0;
size_t dialect_length = hook_dialect ? strlen(hook_dialect) : 0;
size_t total = 19U + root_length + allowed_length + event_length + dialect_length;
if (root_length == 0 || root_length > UINT32_MAX || allowed_length > UINT32_MAX ||
event_length > UINT32_MAX || dialect_length > UINT32_MAX || total > UINT32_MAX) {
return false;
}
uint8_t *request = calloc(1, total);
if (!request) {
return false;
}
request[0] = CBM_DAEMON_APPLICATION_REQUEST_SET_CONTEXT;
request[1] = (uint8_t)(root_length >> 24);
request[2] = (uint8_t)(root_length >> 16);
request[3] = (uint8_t)(root_length >> 8);
request[4] = (uint8_t)root_length;
request[5] = allowed ? 1U : 0U;
request[6] = (uint8_t)(allowed_length >> 24);
request[7] = (uint8_t)(allowed_length >> 16);
request[8] = (uint8_t)(allowed_length >> 8);
request[9] = (uint8_t)allowed_length;
request[10] = (uint8_t)tool_profile;
request[11] = (uint8_t)(event_length >> 24);
request[12] = (uint8_t)(event_length >> 16);
request[13] = (uint8_t)(event_length >> 8);
request[14] = (uint8_t)event_length;
request[15] = (uint8_t)(dialect_length >> 24);
request[16] = (uint8_t)(dialect_length >> 16);
request[17] = (uint8_t)(dialect_length >> 8);
request[18] = (uint8_t)dialect_length;
memcpy(request + 19, root, root_length);
if (allowed_length > 0) {
memcpy(request + 19 + root_length, allowed, allowed_length);
}
if (event_length > 0) {
memcpy(request + 19 + root_length + allowed_length, hook_event, event_length);
}
if (dialect_length > 0) {
memcpy(request + 19 + root_length + allowed_length + event_length, hook_dialect,
dialect_length);
}
*request_out = request;
*length_out = (uint32_t)total;
return true;
}
static bool app_test_context_request(const char *root, const char *allowed, uint8_t **request_out,
uint32_t *length_out) {
return app_test_context_request_options(root, allowed, CBM_MCP_TOOL_PROFILE_ALL, NULL, NULL,
request_out, length_out);
}
static bool app_test_text_request(cbm_daemon_application_request_kind_t kind, const char *text,
uint8_t **request_out, uint32_t *length_out) {
size_t text_length = strlen(text);
if (text_length == 0 || text_length >= UINT32_MAX) {
return false;
}
uint8_t *request = malloc(text_length + 1U);
if (!request) {
return false;
}
request[0] = (uint8_t)kind;
memcpy(request + 1, text, text_length);
*request_out = request;
*length_out = (uint32_t)text_length + 1U;
return true;
}
static bool app_test_tool_request(const char *tool, const char *args, uint8_t **request_out,
uint32_t *length_out) {
size_t tool_length = strlen(tool);
size_t args_length = strlen(args);
size_t total = 5U + tool_length + args_length;
if (tool_length == 0 || args_length == 0 || total > UINT32_MAX) {
return false;
}
uint8_t *request = malloc(total);
if (!request) {
return false;
}
request[0] = CBM_DAEMON_APPLICATION_REQUEST_TOOL;
request[1] = (uint8_t)(tool_length >> 24);
request[2] = (uint8_t)(tool_length >> 16);
request[3] = (uint8_t)(tool_length >> 8);
request[4] = (uint8_t)tool_length;
memcpy(request + 5, tool, tool_length);
memcpy(request + 5 + tool_length, args, args_length);
*request_out = request;
*length_out = (uint32_t)total;
return true;
}
static cbm_daemon_runtime_application_status_t app_test_ui_config_request(
const cbm_daemon_runtime_application_callbacks_t *callbacks,
cbm_daemon_runtime_application_session_t *session, uint8_t mask, uint8_t enabled, uint32_t port,
uint32_t request_length) {
uint8_t request[7] = {
CBM_DAEMON_APPLICATION_REQUEST_SET_UI_CONFIG,
mask,
enabled,
(uint8_t)(port >> 24),
(uint8_t)(port >> 16),
(uint8_t)(port >> 8),
(uint8_t)port,
};
uint8_t *response = NULL;
uint32_t response_length = 0;
cbm_daemon_runtime_application_status_t status =
app_test_request(callbacks, session, request, request_length, &response, &response_length);
if (response || response_length != 0) {
free(response);
return CBM_DAEMON_RUNTIME_APPLICATION_HANDLER_ERROR;
}
return status;
}
TEST(daemon_application_new_session_does_not_retain_initial_store) {
cbm_daemon_application_t *application = cbm_daemon_application_new(NULL);
cbm_daemon_runtime_application_callbacks_t callbacks =
cbm_daemon_application_runtime_callbacks(application);
cbm_daemon_runtime_application_session_t *session = app_test_open(&callbacks, 300);
bool retains_store = session && cbm_daemon_application_session_retains_store_for_test(session);
if (session) {
callbacks.session_close(callbacks.context, session);
}
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
ASSERT_NOT_NULL(application);
ASSERT_NOT_NULL(session);
ASSERT_FALSE(retains_store);
ASSERT_TRUE(stopped);
PASS();
}
TEST(daemon_application_request_cancel_is_scoped_to_exact_token) {
char root[APP_TEST_PATH_CAP];
(void)snprintf(root, sizeof(root), "%s/cbm-app-request-cancel-XXXXXX", cbm_tmpdir());
bool root_ok = cbm_mkdtemp(root) != NULL;
cbm_daemon_application_t *application = cbm_daemon_application_new(NULL);
cbm_daemon_runtime_application_callbacks_t callbacks =
cbm_daemon_application_runtime_callbacks(application);
cbm_daemon_runtime_application_session_t *session = app_test_open(&callbacks, 301);
uint8_t *request = NULL;
uint32_t request_length = 0;
bool request_created =
root_ok && app_test_context_request(root, root, &request, &request_length);
cbm_daemon_runtime_application_token_t cancelled_token = UINT64_C(7001);
cbm_daemon_runtime_application_token_t next_token = UINT64_C(7002);
uint8_t *cancelled_response = NULL;
uint32_t cancelled_response_length = 0;
cbm_daemon_runtime_application_status_t cancelled_status =
CBM_DAEMON_RUNTIME_APPLICATION_TRANSPORT_ERROR;
uint8_t *next_response = NULL;
uint32_t next_response_length = 0;
cbm_daemon_runtime_application_status_t next_status =
CBM_DAEMON_RUNTIME_APPLICATION_TRANSPORT_ERROR;
if (session && request_created) {
callbacks.request_cancel(callbacks.context, session, cancelled_token);
cancelled_status =
app_test_request_tagged(&callbacks, session, cancelled_token, request, request_length,
&cancelled_response, &cancelled_response_length);
/* A late duplicate for the completed token must not poison the next
* unique request on this still-live session. */
callbacks.request_cancel(callbacks.context, session, cancelled_token);
next_status =
app_test_request_tagged(&callbacks, session, next_token, request, request_length,
&next_response, &next_response_length);
}
free(request);
free(cancelled_response);
free(next_response);
if (session) {
callbacks.session_close(callbacks.context, session);
}
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
if (root_ok) {
(void)cbm_rmdir(root);
}
ASSERT_TRUE(root_ok);
ASSERT_NOT_NULL(application);
ASSERT_NOT_NULL(session);
ASSERT_TRUE(request_created);
ASSERT_EQ(cancelled_status, CBM_DAEMON_RUNTIME_APPLICATION_CANCELLED);
ASSERT_EQ(cancelled_response_length, 0);
ASSERT_EQ(next_status, CBM_DAEMON_RUNTIME_APPLICATION_OK);
ASSERT_EQ(next_response_length, 0);
ASSERT_TRUE(stopped);
PASS();
}
TEST(daemon_application_ui_config_updates_are_masked_and_serialized) {
char cache[APP_TEST_PATH_CAP];
(void)snprintf(cache, sizeof(cache), "%s/cbm-app-ui-config-XXXXXX", cbm_tmpdir());
bool cache_ok = cbm_mkdtemp(cache) != NULL;
char *old_cache = getenv("CBM_CACHE_DIR") ? strdup(getenv("CBM_CACHE_DIR")) : NULL;
bool env_ok = cache_ok && cbm_setenv("CBM_CACHE_DIR", cache, 1) == 0;
cbm_ui_config_t initial = {.ui_enabled = false, .ui_port = 9749};
bool initial_saved = env_ok && cbm_ui_config_save(&initial);
cbm_daemon_application_t *application = cbm_daemon_application_new(NULL);
cbm_daemon_runtime_application_callbacks_t callbacks =
cbm_daemon_application_runtime_callbacks(application);
cbm_daemon_runtime_application_session_t *enabled_session = app_test_open(&callbacks, 301);
cbm_daemon_runtime_application_session_t *port_session = app_test_open(&callbacks, 302);
uint8_t *context = NULL;
uint32_t context_length = 0;
uint8_t *context_response = NULL;
uint32_t context_response_length = 0;
bool context_encoded =
cache_ok && app_test_context_request(cache, cache, &context, &context_length);
bool contexts_set =
context_encoded && enabled_session && port_session &&
app_test_request(&callbacks, enabled_session, context, context_length, &context_response,
&context_response_length) == CBM_DAEMON_RUNTIME_APPLICATION_OK;
free(context_response);
context_response = NULL;
context_response_length = 0;
contexts_set =
contexts_set &&
app_test_request(&callbacks, port_session, context, context_length, &context_response,
&context_response_length) == CBM_DAEMON_RUNTIME_APPLICATION_OK;
free(context_response);
/* Each request deliberately carries a stale value for the unmasked field.
* Whole-struct replacement would lose one update; masked daemon-side
* read-modify-write must preserve both regardless of request order. */
cbm_daemon_runtime_application_status_t enabled_status = app_test_ui_config_request(
&callbacks, enabled_session, CBM_DAEMON_APPLICATION_UI_CONFIG_ENABLED, 1, 0, 7);
cbm_daemon_runtime_application_status_t port_status = app_test_ui_config_request(
&callbacks, port_session, CBM_DAEMON_APPLICATION_UI_CONFIG_PORT, 0, 18432, 7);
cbm_ui_config_t updated = {0};
cbm_ui_config_load(&updated);
if (enabled_session) {
callbacks.session_close(callbacks.context, enabled_session);
}
if (port_session) {
callbacks.session_close(callbacks.context, port_session);
}
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
free(context);
if (old_cache) {
(void)cbm_setenv("CBM_CACHE_DIR", old_cache, 1);
} else {
(void)cbm_unsetenv("CBM_CACHE_DIR");
}
free(old_cache);
(void)th_rmtree(cache);
ASSERT_TRUE(cache_ok);
ASSERT_TRUE(env_ok);
ASSERT_TRUE(initial_saved);
ASSERT_NOT_NULL(application);
ASSERT_NOT_NULL(enabled_session);
ASSERT_NOT_NULL(port_session);
ASSERT_TRUE(contexts_set);
ASSERT_EQ(enabled_status, CBM_DAEMON_RUNTIME_APPLICATION_OK);
ASSERT_EQ(port_status, CBM_DAEMON_RUNTIME_APPLICATION_OK);
ASSERT_TRUE(updated.ui_enabled);
ASSERT_EQ(updated.ui_port, 18432);
ASSERT_TRUE(stopped);
PASS();
}
TEST(daemon_application_ui_config_rejects_noncanonical_frames) {
char cache[APP_TEST_PATH_CAP];
(void)snprintf(cache, sizeof(cache), "%s/cbm-app-ui-frame-XXXXXX", cbm_tmpdir());
bool cache_ok = cbm_mkdtemp(cache) != NULL;
char *old_cache = getenv("CBM_CACHE_DIR") ? strdup(getenv("CBM_CACHE_DIR")) : NULL;
bool env_ok = cache_ok && cbm_setenv("CBM_CACHE_DIR", cache, 1) == 0;
cbm_ui_config_t initial = {.ui_enabled = false, .ui_port = 9749};
bool initial_saved = env_ok && cbm_ui_config_save(&initial);
cbm_daemon_application_t *application = cbm_daemon_application_new(NULL);
cbm_daemon_runtime_application_callbacks_t callbacks =
cbm_daemon_application_runtime_callbacks(application);
cbm_daemon_runtime_application_session_t *session = app_test_open(&callbacks, 303);
uint8_t *context = NULL;
uint32_t context_length = 0;
uint8_t *context_response = NULL;
uint32_t context_response_length = 0;
bool context_encoded =
cache_ok && app_test_context_request(cache, cache, &context, &context_length);
bool context_set =
context_encoded && session &&
app_test_request(&callbacks, session, context, context_length, &context_response,
&context_response_length) == CBM_DAEMON_RUNTIME_APPLICATION_OK;
free(context_response);
cbm_daemon_runtime_application_status_t short_frame = app_test_ui_config_request(
&callbacks, session, CBM_DAEMON_APPLICATION_UI_CONFIG_ENABLED, 1, 0, 6);
cbm_daemon_runtime_application_status_t empty_mask =
app_test_ui_config_request(&callbacks, session, 0, 0, 0, 7);
cbm_daemon_runtime_application_status_t unknown_mask =
app_test_ui_config_request(&callbacks, session, 0x80, 0, 0, 7);
cbm_daemon_runtime_application_status_t invalid_boolean = app_test_ui_config_request(
&callbacks, session, CBM_DAEMON_APPLICATION_UI_CONFIG_ENABLED, 2, 0, 7);
cbm_daemon_runtime_application_status_t invalid_port = app_test_ui_config_request(
&callbacks, session, CBM_DAEMON_APPLICATION_UI_CONFIG_PORT, 0, 65536, 7);
cbm_daemon_runtime_application_status_t nonzero_unused = app_test_ui_config_request(
&callbacks, session, CBM_DAEMON_APPLICATION_UI_CONFIG_ENABLED, 1, 9749, 7);
cbm_ui_config_t unchanged = {0};
cbm_ui_config_load(&unchanged);
if (session) {
callbacks.session_close(callbacks.context, session);
}
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
free(context);
if (old_cache) {
(void)cbm_setenv("CBM_CACHE_DIR", old_cache, 1);
} else {
(void)cbm_unsetenv("CBM_CACHE_DIR");
}
free(old_cache);
(void)th_rmtree(cache);
ASSERT_TRUE(cache_ok);
ASSERT_TRUE(env_ok);
ASSERT_TRUE(initial_saved);
ASSERT_NOT_NULL(application);
ASSERT_NOT_NULL(session);
ASSERT_TRUE(context_set);
ASSERT_EQ(short_frame, CBM_DAEMON_RUNTIME_APPLICATION_REJECTED);
ASSERT_EQ(empty_mask, CBM_DAEMON_RUNTIME_APPLICATION_REJECTED);
ASSERT_EQ(unknown_mask, CBM_DAEMON_RUNTIME_APPLICATION_REJECTED);
ASSERT_EQ(invalid_boolean, CBM_DAEMON_RUNTIME_APPLICATION_REJECTED);
ASSERT_EQ(invalid_port, CBM_DAEMON_RUNTIME_APPLICATION_REJECTED);
ASSERT_EQ(nonzero_unused, CBM_DAEMON_RUNTIME_APPLICATION_REJECTED);
ASSERT_FALSE(unchanged.ui_enabled);
ASSERT_EQ(unchanged.ui_port, 9749);
ASSERT_TRUE(stopped);
PASS();
}
TEST(daemon_application_ui_readiness_proof_is_generation_bound_before_context) {
uint8_t first_secret[CBM_SHA256_DIGEST_LEN];
uint8_t second_secret[CBM_SHA256_DIGEST_LEN];
uint8_t request[1U + CBM_SHA256_DIGEST_LEN];
for (size_t index = 0; index < CBM_SHA256_DIGEST_LEN; index++) {
first_secret[index] = (uint8_t)index;
second_secret[index] = (uint8_t)(index + 1U);
request[index + 1U] = (uint8_t)(index + 32U);
}
request[0] = CBM_DAEMON_APPLICATION_REQUEST_UI_READINESS_PROOF;
static const uint8_t expected_first[CBM_SHA256_DIGEST_LEN] = {
0x62, 0x21, 0x5d, 0xe7, 0xbd, 0xdc, 0xea, 0x7e, 0x2c, 0x40, 0x47,
0xff, 0x6b, 0xb9, 0x4f, 0x8d, 0x18, 0x26, 0x2f, 0xc8, 0xb3, 0xf3,
0x64, 0x81, 0x34, 0xbb, 0x7d, 0x44, 0x15, 0x8f, 0xf8, 0x4d,
};
static const uint8_t expected_second[CBM_SHA256_DIGEST_LEN] = {
0xe3, 0xb0, 0x69, 0x53, 0x9c, 0x19, 0xc2, 0x2d, 0x3f, 0x6d, 0x34,
0x60, 0x67, 0x86, 0x15, 0x78, 0xbe, 0xa9, 0xc4, 0xb9, 0xe4, 0x61,
0xcb, 0xf9, 0xb8, 0x06, 0xad, 0x1b, 0xcd, 0x64, 0xa4, 0x81,
};
cbm_daemon_application_config_t first_config = {
.ui_readiness_secret = first_secret,
.ui_readiness_secret_length = sizeof(first_secret),
};
cbm_daemon_application_config_t second_config = {
.ui_readiness_secret = second_secret,
.ui_readiness_secret_length = sizeof(second_secret),
};
cbm_daemon_application_t *first = cbm_daemon_application_new(&first_config);
cbm_daemon_application_t *second = cbm_daemon_application_new(&second_config);
cbm_daemon_application_t *without_secret = cbm_daemon_application_new(NULL);
cbm_daemon_application_config_t malformed_config = {
.ui_readiness_secret = first_secret,
.ui_readiness_secret_length = sizeof(first_secret) - 1U,
};
cbm_daemon_application_t *malformed = cbm_daemon_application_new(&malformed_config);
cbm_daemon_runtime_application_callbacks_t first_callbacks =
cbm_daemon_application_runtime_callbacks(first);
cbm_daemon_runtime_application_callbacks_t second_callbacks =
cbm_daemon_application_runtime_callbacks(second);
cbm_daemon_runtime_application_callbacks_t no_secret_callbacks =
cbm_daemon_application_runtime_callbacks(without_secret);
cbm_daemon_runtime_application_session_t *first_session =
first ? app_test_open(&first_callbacks, 304) : NULL;
cbm_daemon_runtime_application_session_t *second_session =
second ? app_test_open(&second_callbacks, 305) : NULL;
cbm_daemon_runtime_application_session_t *no_secret_session =
without_secret ? app_test_open(&no_secret_callbacks, 306) : NULL;
uint8_t *first_response = NULL;
uint32_t first_length = 0;
cbm_daemon_runtime_application_status_t first_status =
first_session ? app_test_request(&first_callbacks, first_session, request, sizeof(request),
&first_response, &first_length)
: CBM_DAEMON_RUNTIME_APPLICATION_TRANSPORT_ERROR;
uint8_t *second_response = NULL;
uint32_t second_length = 0;
cbm_daemon_runtime_application_status_t second_status =
second_session ? app_test_request(&second_callbacks, second_session, request,
sizeof(request), &second_response, &second_length)
: CBM_DAEMON_RUNTIME_APPLICATION_TRANSPORT_ERROR;
uint8_t *malformed_response = NULL;
uint32_t malformed_length = 0;
cbm_daemon_runtime_application_status_t short_status =
first_session
? app_test_request(&first_callbacks, first_session, request, sizeof(request) - 1U,
&malformed_response, &malformed_length)
: CBM_DAEMON_RUNTIME_APPLICATION_TRANSPORT_ERROR;
free(malformed_response);
malformed_response = NULL;
malformed_length = 0;
uint8_t long_request[2U + CBM_SHA256_DIGEST_LEN];
memcpy(long_request, request, sizeof(request));
long_request[sizeof(long_request) - 1U] = 0;
cbm_daemon_runtime_application_status_t long_status =
first_session
? app_test_request(&first_callbacks, first_session, long_request, sizeof(long_request),
&malformed_response, &malformed_length)
: CBM_DAEMON_RUNTIME_APPLICATION_TRANSPORT_ERROR;
free(malformed_response);
malformed_response = NULL;
malformed_length = 0;
cbm_daemon_runtime_application_status_t no_secret_status =
no_secret_session
? app_test_request(&no_secret_callbacks, no_secret_session, request, sizeof(request),
&malformed_response, &malformed_length)
: CBM_DAEMON_RUNTIME_APPLICATION_TRANSPORT_ERROR;
free(malformed_response);
if (first_session) {
first_callbacks.session_close(first_callbacks.context, first_session);
}
if (second_session) {
second_callbacks.session_close(second_callbacks.context, second_session);
}
if (no_secret_session) {
no_secret_callbacks.session_close(no_secret_callbacks.context, no_secret_session);
}
bool first_stopped = first && cbm_daemon_application_shutdown(first, APP_TEST_TIMEOUT_MS);
bool second_stopped = second && cbm_daemon_application_shutdown(second, APP_TEST_TIMEOUT_MS);
bool no_secret_stopped =
without_secret && cbm_daemon_application_shutdown(without_secret, APP_TEST_TIMEOUT_MS);
bool malformed_rejected = malformed == NULL;
bool first_exact = first_response && first_length == sizeof(expected_first) &&
memcmp(first_response, expected_first, sizeof(expected_first)) == 0;
bool second_exact = second_response && second_length == sizeof(expected_second) &&
memcmp(second_response, expected_second, sizeof(expected_second)) == 0;
bool generations_differ = first_response && second_response &&
memcmp(first_response, second_response, CBM_SHA256_DIGEST_LEN) != 0;
cbm_daemon_application_free(first);
cbm_daemon_application_free(second);
cbm_daemon_application_free(without_secret);
cbm_daemon_application_free(malformed);
free(first_response);
free(second_response);
ASSERT_NOT_NULL(first);
ASSERT_NOT_NULL(second);
ASSERT_NOT_NULL(without_secret);
ASSERT_TRUE(malformed_rejected);
ASSERT_NOT_NULL(first_session);
ASSERT_NOT_NULL(second_session);
ASSERT_NOT_NULL(no_secret_session);
ASSERT_EQ(first_status, CBM_DAEMON_RUNTIME_APPLICATION_OK);
ASSERT_TRUE(first_exact);
ASSERT_EQ(second_status, CBM_DAEMON_RUNTIME_APPLICATION_OK);
ASSERT_TRUE(second_exact);
ASSERT_TRUE(generations_differ);
ASSERT_EQ(short_status, CBM_DAEMON_RUNTIME_APPLICATION_REJECTED);
ASSERT_EQ(long_status, CBM_DAEMON_RUNTIME_APPLICATION_REJECTED);
ASSERT_EQ(no_secret_status, CBM_DAEMON_RUNTIME_APPLICATION_REJECTED);
ASSERT_TRUE(first_stopped);
ASSERT_TRUE(second_stopped);
ASSERT_TRUE(no_secret_stopped);
PASS();
}
TEST(daemon_application_requires_immutable_explicit_context) {
cbm_daemon_application_t *application = cbm_daemon_application_new(NULL);
cbm_daemon_runtime_application_callbacks_t callbacks =
cbm_daemon_application_runtime_callbacks(application);
cbm_daemon_runtime_application_session_t *session = app_test_open(&callbacks, 1);
char root[APP_TEST_PATH_CAP];
snprintf(root, sizeof(root), "%s/cbm-app-context-XXXXXX", cbm_tmpdir());
bool root_ok = cbm_mkdtemp(root) != NULL;
uint8_t *mcp = NULL;
uint32_t mcp_length = 0;
uint8_t *context = NULL;
uint32_t context_length = 0;
uint8_t *response = NULL;
uint32_t response_length = 0;
bool mcp_ok = app_test_text_request(CBM_DAEMON_APPLICATION_REQUEST_MCP,
"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"ping\"}", &mcp,
&mcp_length);
cbm_daemon_runtime_application_status_t before =
app_test_request(&callbacks, session, mcp, mcp_length, &response, &response_length);
free(response);
response = NULL;
bool context_ok = root_ok && app_test_context_request(root, root, &context, &context_length);
cbm_daemon_runtime_application_status_t first =
context_ok ? app_test_request(&callbacks, session, context, context_length, &response,
&response_length)
: CBM_DAEMON_RUNTIME_APPLICATION_TRANSPORT_ERROR;
free(response);
response = NULL;
cbm_daemon_runtime_application_status_t repeated =
context_ok ? app_test_request(&callbacks, session, context, context_length, &response,
&response_length)
: CBM_DAEMON_RUNTIME_APPLICATION_TRANSPORT_ERROR;
free(response);
response = NULL;
cbm_daemon_runtime_application_status_t after =
app_test_request(&callbacks, session, mcp, mcp_length, &response, &response_length);
bool ping_response =
response && response_length > 0 && strstr((const char *)response, "\"result\":{}") != NULL;
callbacks.session_close(callbacks.context, session);
(void)cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
free(mcp);
free(context);
free(response);
(void)cbm_rmdir(root);
ASSERT_TRUE(application != NULL);
ASSERT_TRUE(session != NULL);
ASSERT_TRUE(mcp_ok);
ASSERT_EQ(before, CBM_DAEMON_RUNTIME_APPLICATION_REJECTED);
ASSERT_EQ(first, CBM_DAEMON_RUNTIME_APPLICATION_OK);
ASSERT_EQ(repeated, CBM_DAEMON_RUNTIME_APPLICATION_REJECTED);
ASSERT_EQ(after, CBM_DAEMON_RUNTIME_APPLICATION_OK);
ASSERT_TRUE(ping_response);
PASS();
}
TEST(daemon_application_mcp_notification_has_no_response) {
cbm_daemon_application_t *application = cbm_daemon_application_new(NULL);
cbm_daemon_runtime_application_callbacks_t callbacks =
cbm_daemon_application_runtime_callbacks(application);
cbm_daemon_runtime_application_session_t *session = app_test_open(&callbacks, 2);
char root[APP_TEST_PATH_CAP];
snprintf(root, sizeof(root), "%s/cbm-app-notify-XXXXXX", cbm_tmpdir());
bool root_ok = cbm_mkdtemp(root) != NULL;
uint8_t *context = NULL;
uint32_t context_length = 0;
uint8_t *notification = NULL;
uint32_t notification_length = 0;
uint8_t *response = (uint8_t *)(uintptr_t)1;
uint32_t response_length = UINT32_MAX;
bool encoded =
root_ok && app_test_context_request(root, NULL, &context, &context_length) &&
app_test_text_request(CBM_DAEMON_APPLICATION_REQUEST_MCP,
"{\"jsonrpc\":\"2.0\",\"method\":\"notifications/initialized\"}",
&notification, &notification_length);
cbm_daemon_runtime_application_status_t context_status =
encoded ? app_test_request(&callbacks, session, context, context_length, &response,
&response_length)
: CBM_DAEMON_RUNTIME_APPLICATION_TRANSPORT_ERROR;
free(response);
response = (uint8_t *)(uintptr_t)1;
response_length = UINT32_MAX;
cbm_daemon_runtime_application_status_t notification_status =
encoded ? app_test_request(&callbacks, session, notification, notification_length,
&response, &response_length)
: CBM_DAEMON_RUNTIME_APPLICATION_TRANSPORT_ERROR;
callbacks.session_close(callbacks.context, session);
cbm_daemon_application_free(application);
free(context);
free(notification);
(void)cbm_rmdir(root);
ASSERT_TRUE(encoded);
ASSERT_EQ(context_status, CBM_DAEMON_RUNTIME_APPLICATION_OK);
ASSERT_EQ(notification_status, CBM_DAEMON_RUNTIME_APPLICATION_OK);
ASSERT_TRUE(response == NULL);
ASSERT_EQ(response_length, 0);
PASS();
}
static int app_test_index_noop(const char *project_name, const char *root_path, void *context) {
(void)project_name;
(void)root_path;
(void)context;
return 0;
}
static bool app_test_create_empty_file(const char *path) {
FILE *file = path ? cbm_fopen(path, "wb") : NULL;
return file && fclose(file) == 0;
}
static int app_test_git(const char *root, const char *operation, const char *argument_one,
const char *argument_two) {
const char *git = "git";
#ifdef _WIN32
char git_executable[APP_TEST_PATH_CAP];
const char *resolved = cbm_find_cli("git", cbm_get_home_dir());
int resolved_length =
resolved ? snprintf(git_executable, sizeof(git_executable), "%s", resolved) : -1;
if (resolved_length <= 0 || (size_t)resolved_length >= sizeof(git_executable)) {
return -1;
}
git = git_executable;
#endif
const char *argv[] = {git, "-C", root, operation, argument_one, argument_two, NULL};
cbm_proc_opts_t options = {0};
cbm_proc_result_t result = {0};
options.bin = git;
options.argv = argv;
options.quiet_timeout_ms = 10000;
return cbm_subprocess_run(&options, &result) == 0 && result.outcome == CBM_PROC_CLEAN ? 0 : -1;
}
/* Rebase guard: restricted MCP profiles are a property of one authenticated
* daemon session. Before profile propagation, the daemon silently created a
* full-surface MCP server, subscribed that session to the shared watcher, and
* still accepted raw UI mutations. */
TEST(daemon_application_restricted_profile_owns_no_background_surfaces) {
const char *old_cache = getenv("CBM_CACHE_DIR");
bool had_cache = old_cache != NULL;
char *saved_cache = old_cache ? cbm_strdup(old_cache) : NULL;
char root[APP_TEST_PATH_CAP];
char cache[APP_TEST_PATH_CAP];
(void)snprintf(root, sizeof(root), "%s/cbm-app-profile-root-XXXXXX", cbm_tmpdir());
(void)snprintf(cache, sizeof(cache), "%s/cbm-app-profile-cache-XXXXXX", cbm_tmpdir());
bool dirs_ok = cbm_mkdtemp(root) != NULL && cbm_mkdtemp(cache) != NULL;
bool env_ok =
dirs_ok && (!had_cache || saved_cache) && cbm_setenv("CBM_CACHE_DIR", cache, 1) == 0;
cbm_ui_config_t initial_ui = {.ui_enabled = false, .ui_port = 9749};
bool ui_ready = env_ok && cbm_ui_config_save(&initial_ui);
char *project = dirs_ok ? cbm_project_name_from_path(root) : NULL;
char db_path[APP_TEST_PATH_CAP] = {0};
if (project) {
(void)snprintf(db_path, sizeof(db_path), "%s/%s.db", cache, project);
}
bool db_ok = project && app_test_create_empty_file(db_path);
cbm_store_t *store = cbm_store_open_memory();
cbm_watcher_t *watcher = cbm_watcher_new(store, app_test_index_noop, NULL);
cbm_daemon_application_config_t config = {.watcher = watcher};
cbm_daemon_application_t *application = cbm_daemon_application_new(&config);
cbm_daemon_runtime_application_callbacks_t callbacks =
cbm_daemon_application_runtime_callbacks(application);
cbm_daemon_runtime_application_session_t *session = app_test_open(&callbacks, 304);
uint8_t *context = NULL;
uint32_t context_length = 0;
uint8_t *initialize = NULL;
uint32_t initialize_length = 0;
uint8_t *index_call = NULL;
uint32_t index_call_length = 0;
bool encoded =
db_ok && session &&
app_test_context_request_options(root, root, CBM_MCP_TOOL_PROFILE_SCOUT, NULL, NULL,
&context, &context_length) &&
app_test_text_request(
CBM_DAEMON_APPLICATION_REQUEST_MCP,
"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\"params\":{}}", &initialize,
&initialize_length) &&
app_test_text_request(CBM_DAEMON_APPLICATION_REQUEST_MCP,
"{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"tools/call\","
"\"params\":{\"name\":\"index_repository\",\"arguments\":{}}}",
&index_call, &index_call_length);
uint8_t *response = NULL;
uint32_t response_length = 0;
cbm_daemon_runtime_application_status_t context_status =
encoded ? app_test_request(&callbacks, session, context, context_length, &response,
&response_length)
: CBM_DAEMON_RUNTIME_APPLICATION_TRANSPORT_ERROR;
free(response);
response = NULL;
cbm_daemon_runtime_application_status_t initialize_status =
context_status == CBM_DAEMON_RUNTIME_APPLICATION_OK
? app_test_request(&callbacks, session, initialize, initialize_length, &response,
&response_length)
: CBM_DAEMON_RUNTIME_APPLICATION_TRANSPORT_ERROR;
bool scout_surface = response && strstr((const char *)response, "scout tool profile") &&
!strstr((const char *)response, "index_repository");
free(response);
response = NULL;
cbm_daemon_runtime_application_status_t index_status =
initialize_status == CBM_DAEMON_RUNTIME_APPLICATION_OK
? app_test_request(&callbacks, session, index_call, index_call_length, &response,
&response_length)
: CBM_DAEMON_RUNTIME_APPLICATION_TRANSPORT_ERROR;
bool index_refused =
response && strstr((const char *)response, "not available in the scout tool profile");
free(response);
response = NULL;
cbm_daemon_runtime_application_status_t ui_status = app_test_ui_config_request(
&callbacks, session, CBM_DAEMON_APPLICATION_UI_CONFIG_ENABLED, 1, 0, 7);
int watch_count = watcher ? cbm_watcher_watch_count(watcher) : -1;
size_t active_jobs = application ? cbm_daemon_application_active_jobs(application) : SIZE_MAX;
cbm_ui_config_t final_ui = {0};
cbm_ui_config_load(&final_ui);
if (session) {
callbacks.session_close(callbacks.context, session);
}
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
if (watcher) {
cbm_watcher_stop(watcher);
cbm_watcher_free(watcher);
}
cbm_store_close(store);
free(context);
free(initialize);
free(index_call);
free(project);
(void)cbm_unlink(db_path);
(void)cbm_rmdir(root);
if (had_cache) {
(void)cbm_setenv("CBM_CACHE_DIR", saved_cache, 1);
} else {
(void)cbm_unsetenv("CBM_CACHE_DIR");
}
free(saved_cache);
(void)th_rmtree(cache);
ASSERT_TRUE(dirs_ok);
ASSERT_TRUE(env_ok);
ASSERT_TRUE(ui_ready);
ASSERT_TRUE(db_ok);
ASSERT_TRUE(encoded);
ASSERT_EQ(context_status, CBM_DAEMON_RUNTIME_APPLICATION_OK);
ASSERT_EQ(initialize_status, CBM_DAEMON_RUNTIME_APPLICATION_OK);
ASSERT_TRUE(scout_surface);
ASSERT_EQ(index_status, CBM_DAEMON_RUNTIME_APPLICATION_OK);
ASSERT_TRUE(index_refused);
ASSERT_EQ(ui_status, CBM_DAEMON_RUNTIME_APPLICATION_REJECTED);
ASSERT_EQ(watch_count, 0);
ASSERT_EQ(active_jobs, 0);
ASSERT_FALSE(final_ui.ui_enabled);
ASSERT_TRUE(stopped);
PASS();
}
/* The thin hook process parses CLI metadata, but the daemon owns the MCP
* session and therefore must retain that metadata with the session context.
* Copilot deliberately omits the event from stdin, making this non-vacuous. */
TEST(daemon_application_hook_context_preserves_event_and_dialect) {
char root[APP_TEST_PATH_CAP];
(void)snprintf(root, sizeof(root), "%s/cbm-app-hook-context-XXXXXX", cbm_tmpdir());
bool root_ok = cbm_mkdtemp(root) != NULL;
cbm_daemon_application_t *application = cbm_daemon_application_new(NULL);
cbm_daemon_runtime_application_callbacks_t callbacks =
cbm_daemon_application_runtime_callbacks(application);
cbm_daemon_runtime_application_session_t *session = app_test_open(&callbacks, 305);
uint8_t *context = NULL;
uint32_t context_length = 0;
char input[APP_TEST_PATH_CAP + 32];
(void)snprintf(input, sizeof(input), "{\"cwd\":\"%s\"}", root);
uint8_t *hook = NULL;
uint32_t hook_length = 0;
bool encoded =
root_ok && session &&
app_test_context_request_options(root, root, CBM_MCP_TOOL_PROFILE_ALL, "SessionStart",
"copilot", &context, &context_length) &&
app_test_text_request(CBM_DAEMON_APPLICATION_REQUEST_HOOK_AUGMENT, input, &hook,
&hook_length);
uint8_t *response = NULL;
uint32_t response_length = 0;
cbm_daemon_runtime_application_status_t context_status =
encoded ? app_test_request(&callbacks, session, context, context_length, &response,
&response_length)
: CBM_DAEMON_RUNTIME_APPLICATION_TRANSPORT_ERROR;
free(response);
response = NULL;
cbm_daemon_runtime_application_status_t hook_status =
context_status == CBM_DAEMON_RUNTIME_APPLICATION_OK
? app_test_request(&callbacks, session, hook, hook_length, &response, &response_length)
: CBM_DAEMON_RUNTIME_APPLICATION_TRANSPORT_ERROR;
bool copilot_shape = response && strstr((const char *)response, "additionalContext") &&
strstr((const char *)response, "Session context") &&
!strstr((const char *)response, "hookSpecificOutput");
free(response);
free(context);
free(hook);
if (session) {
callbacks.session_close(callbacks.context, session);
}
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
if (root_ok) {
(void)cbm_rmdir(root);
}
ASSERT_TRUE(root_ok);
ASSERT_TRUE(encoded);
ASSERT_EQ(context_status, CBM_DAEMON_RUNTIME_APPLICATION_OK);
ASSERT_EQ(hook_status, CBM_DAEMON_RUNTIME_APPLICATION_OK);
ASSERT_TRUE(copilot_shape);
ASSERT_TRUE(stopped);
PASS();
}
TEST(daemon_application_reference_counts_one_shared_watch) {
const char *old_cache = getenv("CBM_CACHE_DIR");
bool had_cache = old_cache != NULL;
char *saved_cache = old_cache ? cbm_strdup(old_cache) : NULL;
bool saved_cache_ok = !had_cache || saved_cache;
char root[APP_TEST_PATH_CAP];
char cache[APP_TEST_PATH_CAP];
snprintf(root, sizeof(root), "%s/cbm-app-watch-root-XXXXXX", cbm_tmpdir());
snprintf(cache, sizeof(cache), "%s/cbm-app-watch-cache-XXXXXX", cbm_tmpdir());
bool dirs_ok = cbm_mkdtemp(root) != NULL && cbm_mkdtemp(cache) != NULL;
bool env_ok = saved_cache_ok && cbm_setenv("CBM_CACHE_DIR", cache, 1) == 0;
char *project = dirs_ok ? cbm_project_name_from_path(root) : NULL;
char db_path[APP_TEST_PATH_CAP] = {0};
FILE *db = NULL;
if (project) {
snprintf(db_path, sizeof(db_path), "%s/%s.db", cache, project);
db = cbm_fopen(db_path, "wb");
}
bool db_ok = db && fclose(db) == 0;
cbm_store_t *store = cbm_store_open_memory();
cbm_watcher_t *watcher = cbm_watcher_new(store, app_test_index_noop, NULL);
cbm_daemon_application_config_t config = {.watcher = watcher, .config = NULL};
cbm_daemon_application_t *application = cbm_daemon_application_new(&config);
cbm_daemon_runtime_application_callbacks_t callbacks =
cbm_daemon_application_runtime_callbacks(application);
cbm_daemon_runtime_application_session_t *first_session = app_test_open(&callbacks, 10);
cbm_daemon_runtime_application_session_t *second_session = app_test_open(&callbacks, 11);
uint8_t *context = NULL;
uint32_t context_length = 0;
uint8_t *ping = NULL;
uint32_t ping_length = 0;
uint8_t *response = NULL;
uint32_t response_length = 0;
bool encoded = dirs_ok && db_ok &&
app_test_context_request(root, root, &context, &context_length) &&
app_test_text_request(CBM_DAEMON_APPLICATION_REQUEST_MCP,
"{\"jsonrpc\":\"2.0\",\"id\":7,\"method\":\"ping\"}",
&ping, &ping_length);
bool requests_ok = encoded;
cbm_daemon_runtime_application_session_t *sessions[2] = {first_session, second_session};
for (size_t i = 0; requests_ok && i < 2; i++) {
requests_ok = app_test_request(&callbacks, sessions[i], context, context_length, &response,
&response_length) == CBM_DAEMON_RUNTIME_APPLICATION_OK &&
app_test_request(&callbacks, sessions[i], ping, ping_length, &response,
&response_length) == CBM_DAEMON_RUNTIME_APPLICATION_OK;
free(response);
response = NULL;
}
int shared_count = cbm_watcher_watch_count(watcher);
callbacks.session_cancel(callbacks.context, first_session);
int after_first_cancel = cbm_watcher_watch_count(watcher);
callbacks.session_cancel(callbacks.context, second_session);
int after_final_cancel = cbm_watcher_watch_count(watcher);
callbacks.session_close(callbacks.context, first_session);
int after_first_close = cbm_watcher_watch_count(watcher);
callbacks.session_close(callbacks.context, second_session);
int after_final_close = cbm_watcher_watch_count(watcher);
cbm_daemon_application_free(application);
cbm_watcher_stop(watcher);
cbm_watcher_free(watcher);
cbm_store_close(store);
free(context);
free(ping);
free(response);
free(project);
(void)cbm_unlink(db_path);
(void)cbm_rmdir(root);
(void)cbm_rmdir(cache);
if (saved_cache) {
(void)cbm_setenv("CBM_CACHE_DIR", saved_cache, 1);
} else if (!had_cache) {
(void)cbm_unsetenv("CBM_CACHE_DIR");
}
free(saved_cache);
ASSERT_TRUE(env_ok);
ASSERT_TRUE(requests_ok);
ASSERT_EQ(shared_count, 1);
ASSERT_EQ(after_first_cancel, 1);
ASSERT_EQ(after_final_cancel, 0);
ASSERT_EQ(after_first_close, 0);
ASSERT_EQ(after_final_close, 0);
PASS();
}
TEST(daemon_application_free_releases_live_watch_once) {
const char *old_cache = getenv("CBM_CACHE_DIR");
bool had_cache = old_cache != NULL;
char *saved_cache = old_cache ? cbm_strdup(old_cache) : NULL;
const char *old_grace = getenv("CBM_WATCHER_PRUNE_GRACE_S");
bool had_grace = old_grace != NULL;
char *saved_grace = old_grace ? cbm_strdup(old_grace) : NULL;
char root[APP_TEST_PATH_CAP];
char survivor_root[APP_TEST_PATH_CAP];
char cache[APP_TEST_PATH_CAP];
snprintf(root, sizeof(root), "%s/cbm-app-free-watch-root-XXXXXX", cbm_tmpdir());
snprintf(survivor_root, sizeof(survivor_root), "%s/cbm-app-free-survivor-root-XXXXXX",
cbm_tmpdir());
snprintf(cache, sizeof(cache), "%s/cbm-app-free-watch-cache-XXXXXX", cbm_tmpdir());
bool dirs_ok = cbm_mkdtemp(root) != NULL && cbm_mkdtemp(survivor_root) != NULL &&
cbm_mkdtemp(cache) != NULL;
bool env_ok = (!had_cache || saved_cache) && (!had_grace || saved_grace) &&
cbm_setenv("CBM_CACHE_DIR", cache, 1) == 0 &&
cbm_setenv("CBM_WATCHER_PRUNE_GRACE_S", "0", 1) == 0;
char *project = dirs_ok ? cbm_project_name_from_path(root) : NULL;
char *survivor_project = dirs_ok ? cbm_project_name_from_path(survivor_root) : NULL;
char db_path[APP_TEST_PATH_CAP] = {0};
char survivor_db_path[APP_TEST_PATH_CAP] = {0};
if (project) {
snprintf(db_path, sizeof(db_path), "%s/%s.db", cache, project);
}
if (survivor_project) {
snprintf(survivor_db_path, sizeof(survivor_db_path), "%s/%s.db", cache, survivor_project);
}
bool db_ok =
app_test_create_empty_file(db_path) && app_test_create_empty_file(survivor_db_path);
cbm_store_t *store = cbm_store_open_memory();
cbm_watcher_t *watcher = cbm_watcher_new(store, app_test_index_noop, NULL);
cbm_daemon_application_config_t config = {.watcher = watcher};
cbm_daemon_application_t *application = cbm_daemon_application_new(&config);
cbm_daemon_runtime_application_callbacks_t callbacks =
cbm_daemon_application_runtime_callbacks(application);
cbm_daemon_runtime_application_session_t *session = app_test_open(&callbacks, 12);
uint8_t *context = NULL;
uint32_t context_length = 0;
uint8_t *ping = NULL;
uint32_t ping_length = 0;
uint8_t *response = NULL;
uint32_t response_length = 0;
bool encoded = dirs_ok && env_ok && db_ok && session &&
app_test_context_request(root, root, &context, &context_length) &&
app_test_text_request(CBM_DAEMON_APPLICATION_REQUEST_MCP,
"{\"jsonrpc\":\"2.0\",\"id\":8,\"method\":\"ping\"}",
&ping, &ping_length);
bool requested =
encoded && app_test_request(&callbacks, session, context, context_length, &response,
&response_length) == CBM_DAEMON_RUNTIME_APPLICATION_OK;
free(response);
response = NULL;
requested =
requested && app_test_request(&callbacks, session, ping, ping_length, &response,
&response_length) == CBM_DAEMON_RUNTIME_APPLICATION_OK;
bool watch_live = requested && cbm_watcher_watch_count(watcher) == 1;
/* Keep one watcher entry outside the application's subscription table.
* It survives application teardown and then enters the destructive prune
* path. If application_free leaves its borrowed callbacks installed, the
* third poll calls through the freed application context (caught by ASan). */
if (watch_live) {
cbm_watcher_watch(watcher, survivor_project, survivor_root);
}
bool survivor_live = watch_live && cbm_watcher_watch_count(watcher) == 2;
bool survivor_removed = survivor_live && cbm_rmdir(survivor_root) == 0;
/* Deliberately leave the session open: application teardown owns both
* the logical session and its one physical watch. ASan catches a second
* release of the detached watch node. */
cbm_daemon_application_free(application);
bool survivor_still_watched = cbm_watcher_watch_count(watcher) == 1;
for (int miss = 0; survivor_removed && miss < 3; miss++) {
cbm_watcher_touch(watcher, survivor_project);
(void)cbm_watcher_poll_once(watcher);
}
bool post_free_poll_safe = survivor_removed && cbm_watcher_watch_count(watcher) == 0 &&
!cbm_file_exists(survivor_db_path);
cbm_watcher_stop(watcher);
cbm_watcher_free(watcher);
cbm_store_close(store);
free(context);
free(ping);
free(response);
free(project);
free(survivor_project);
(void)cbm_unlink(db_path);
(void)cbm_unlink(survivor_db_path);
(void)cbm_rmdir(root);
(void)cbm_rmdir(survivor_root);
(void)cbm_rmdir(cache);
if (saved_cache) {
(void)cbm_setenv("CBM_CACHE_DIR", saved_cache, 1);
} else if (!had_cache) {
(void)cbm_unsetenv("CBM_CACHE_DIR");
}
if (saved_grace) {
(void)cbm_setenv("CBM_WATCHER_PRUNE_GRACE_S", saved_grace, 1);
} else if (!had_grace) {
(void)cbm_unsetenv("CBM_WATCHER_PRUNE_GRACE_S");
}
free(saved_cache);
free(saved_grace);
ASSERT_TRUE(encoded);
ASSERT_TRUE(requested);
ASSERT_TRUE(watch_live);
ASSERT_TRUE(survivor_live);
ASSERT_TRUE(survivor_still_watched);
ASSERT_TRUE(post_free_poll_safe);
PASS();
}
TEST(daemon_application_prune_clears_logical_watch_for_reregistration) {
const char *old_cache = getenv("CBM_CACHE_DIR");
bool had_cache = old_cache != NULL;
char *saved_cache = old_cache ? cbm_strdup(old_cache) : NULL;
const char *old_grace = getenv("CBM_WATCHER_PRUNE_GRACE_S");
bool had_grace = old_grace != NULL;
char *saved_grace = old_grace ? cbm_strdup(old_grace) : NULL;
char root[APP_TEST_PATH_CAP];
char cache[APP_TEST_PATH_CAP];
snprintf(root, sizeof(root), "%s/cbm-app-prune-watch-root-XXXXXX", cbm_tmpdir());
snprintf(cache, sizeof(cache), "%s/cbm-app-prune-watch-cache-XXXXXX", cbm_tmpdir());
bool dirs_ok = cbm_mkdtemp(root) != NULL && cbm_mkdtemp(cache) != NULL;
bool env_ok = (!had_cache || saved_cache) && (!had_grace || saved_grace) &&
cbm_setenv("CBM_CACHE_DIR", cache, 1) == 0 &&
cbm_setenv("CBM_WATCHER_PRUNE_GRACE_S", "0", 1) == 0;
char *project = dirs_ok ? cbm_project_name_from_path(root) : NULL;
char db_path[APP_TEST_PATH_CAP] = {0};
char wal_path[APP_TEST_PATH_CAP] = {0};
char shm_path[APP_TEST_PATH_CAP] = {0};
if (project) {
snprintf(db_path, sizeof(db_path), "%s/%s.db", cache, project);
snprintf(wal_path, sizeof(wal_path), "%s/%s.db-wal", cache, project);
snprintf(shm_path, sizeof(shm_path), "%s/%s.db-shm", cache, project);
}
bool files_ok = app_test_create_empty_file(db_path) && app_test_create_empty_file(wal_path) &&
app_test_create_empty_file(shm_path);
cbm_store_t *store = cbm_store_open_memory();
cbm_watcher_t *watcher = cbm_watcher_new(store, app_test_index_noop, NULL);
cbm_daemon_application_config_t config = {.watcher = watcher};
cbm_daemon_application_t *application = cbm_daemon_application_new(&config);
cbm_daemon_runtime_application_callbacks_t callbacks =
cbm_daemon_application_runtime_callbacks(application);
cbm_daemon_runtime_application_session_t *first = app_test_open(&callbacks, 13);
cbm_daemon_runtime_application_session_t *second = NULL;
uint8_t *context = NULL;
uint32_t context_length = 0;
uint8_t *ping = NULL;
uint32_t ping_length = 0;
uint8_t *response = NULL;
uint32_t response_length = 0;
bool encoded = dirs_ok && env_ok && files_ok && store && watcher && application && first &&
app_test_context_request(root, root, &context, &context_length) &&
app_test_text_request(CBM_DAEMON_APPLICATION_REQUEST_MCP,
"{\"jsonrpc\":\"2.0\",\"id\":9,\"method\":\"ping\"}",
&ping, &ping_length);
bool first_registered =
encoded && app_test_request(&callbacks, first, context, context_length, &response,
&response_length) == CBM_DAEMON_RUNTIME_APPLICATION_OK;
free(response);
response = NULL;
first_registered = first_registered &&
app_test_request(&callbacks, first, ping, ping_length, &response,
&response_length) == CBM_DAEMON_RUNTIME_APPLICATION_OK &&
cbm_watcher_watch_count(watcher) == 1;
free(response);
response = NULL;
bool root_removed = first_registered && cbm_rmdir(root) == 0;
for (int miss = 0; root_removed && miss < 3; miss++) {
cbm_watcher_touch(watcher, project);
(void)cbm_watcher_poll_once(watcher);
}
bool pruned = root_removed && cbm_watcher_watch_count(watcher) == 0 &&
!cbm_file_exists(db_path) && !cbm_file_exists(wal_path) &&
!cbm_file_exists(shm_path);
bool restored = pruned && cbm_mkdir_p(root, 0755) && app_test_create_empty_file(db_path);
if (restored) {
second = app_test_open(&callbacks, 14);
}
bool second_registered =
second && app_test_request(&callbacks, second, context, context_length, &response,
&response_length) == CBM_DAEMON_RUNTIME_APPLICATION_OK;
free(response);
response = NULL;
second_registered = second_registered &&
app_test_request(&callbacks, second, ping, ping_length, &response,
&response_length) == CBM_DAEMON_RUNTIME_APPLICATION_OK &&
cbm_watcher_watch_count(watcher) == 1;
free(response);
response = NULL;
if (first) {
callbacks.session_close(callbacks.context, first);
first = NULL;
}
bool first_close_kept_reregistered_watch =
second_registered && cbm_watcher_watch_count(watcher) == 1;
if (second) {
callbacks.session_close(callbacks.context, second);
second = NULL;
}
bool final_close_released_watch = cbm_watcher_watch_count(watcher) == 0;
cbm_daemon_application_free(application);
cbm_watcher_stop(watcher);
cbm_watcher_free(watcher);
cbm_store_close(store);
free(context);
free(ping);
free(response);
free(project);
(void)cbm_unlink(db_path);
(void)cbm_unlink(wal_path);
(void)cbm_unlink(shm_path);
(void)cbm_rmdir(root);
(void)cbm_rmdir(cache);
if (saved_cache) {
(void)cbm_setenv("CBM_CACHE_DIR", saved_cache, 1);
} else if (!had_cache) {
(void)cbm_unsetenv("CBM_CACHE_DIR");
}
if (saved_grace) {
(void)cbm_setenv("CBM_WATCHER_PRUNE_GRACE_S", saved_grace, 1);
} else if (!had_grace) {
(void)cbm_unsetenv("CBM_WATCHER_PRUNE_GRACE_S");
}
free(saved_cache);
free(saved_grace);
ASSERT_TRUE(encoded);
ASSERT_TRUE(first_registered);
ASSERT_TRUE(pruned);
ASSERT_TRUE(restored);
ASSERT_TRUE(second_registered);
ASSERT_TRUE(first_close_kept_reregistered_watch);
ASSERT_TRUE(final_close_released_watch);
PASS();
}
enum { APP_FAKE_MAX_ATTEMPTS = 16 };
typedef struct {
atomic_int starts;
atomic_int cancels;
atomic_int destroys;
atomic_bool allow_completion;
atomic_bool unsafe_clean;
atomic_bool scripted;
atomic_int hold_destroy_attempt;
atomic_bool release_destroy;
atomic_int project_lock_attempts;
atomic_int project_lock_acquisitions;
cbm_project_lock_manager_t *project_locks;
bool project_lock_busy_is_error;
cbm_proc_outcome_t outcomes[APP_FAKE_MAX_ATTEMPTS];
const char *responses[APP_FAKE_MAX_ATTEMPTS];
const char *marker_payloads[APP_FAKE_MAX_ATTEMPTS];
char marker_paths[APP_FAKE_MAX_ATTEMPTS][APP_TEST_PATH_CAP];
char quarantine_paths[APP_FAKE_MAX_ATTEMPTS][APP_TEST_PATH_CAP];
char quarantine_seen[APP_FAKE_MAX_ATTEMPTS][APP_TEST_PATH_CAP];
size_t memory_budgets[APP_FAKE_MAX_ATTEMPTS];
} app_fake_worker_context_t;
typedef struct {
app_fake_worker_context_t *context;
int attempt;
atomic_bool cancelled;
char *project_key;
cbm_project_lock_lease_t *project_lock_lease;
cbm_index_worker_result_t result;
} app_fake_worker_t;
static void app_fake_worker_context_init(app_fake_worker_context_t *context) {
memset(context, 0, sizeof(*context));
atomic_init(&context->starts, 0);
atomic_init(&context->cancels, 0);
atomic_init(&context->destroys, 0);
atomic_init(&context->allow_completion, false);
atomic_init(&context->unsafe_clean, false);
atomic_init(&context->scripted, false);
atomic_init(&context->hold_destroy_attempt, -1);
atomic_init(&context->release_destroy, false);
atomic_init(&context->project_lock_attempts, 0);
atomic_init(&context->project_lock_acquisitions, 0);
}
static void app_fake_worker_read_file(const char *path, char *out, size_t out_size) {
if (!path || !path[0] || !out || out_size == 0) {
return;
}
FILE *file = cbm_fopen(path, "rb");
if (!file) {
return;
}
size_t read = fread(out, 1, out_size - 1U, file);
out[read] = '\0';
(void)fclose(file);
}
static int app_fake_worker_start(void *opaque, const char *args_json, size_t memory_budget_bytes,
const char *marker_file, const char *quarantine_file,
cbm_daemon_application_worker_t *worker_out) {
app_fake_worker_context_t *context = opaque;
app_fake_worker_t *worker = calloc(1, sizeof(*worker));
if (!worker) {
return -1;
}
if (context->project_locks) {
char *repo_path = cbm_mcp_get_string_arg(args_json, "repo_path");
char *name_override = cbm_mcp_get_string_arg(args_json, "name");
worker->project_key = cbm_project_name_from_path(
name_override && name_override[0] ? name_override : repo_path);
free(name_override);
free(repo_path);
if (!worker->project_key) {
free(worker);
return -1;
}
}
worker->context = context;
worker->attempt = atomic_fetch_add(&context->starts, 1);
atomic_init(&worker->cancelled, false);
worker->result.exit_code = -1;
if (worker->attempt < APP_FAKE_MAX_ATTEMPTS) {
context->memory_budgets[worker->attempt] = memory_budget_bytes;
if (marker_file) {
(void)snprintf(context->marker_paths[worker->attempt], APP_TEST_PATH_CAP, "%s",
marker_file);
}
if (quarantine_file) {
(void)snprintf(context->quarantine_paths[worker->attempt], APP_TEST_PATH_CAP, "%s",
quarantine_file);
app_fake_worker_read_file(quarantine_file, context->quarantine_seen[worker->attempt],
sizeof(context->quarantine_seen[worker->attempt]));
}
const char *payload = context->marker_payloads[worker->attempt];
if (marker_file && payload) {
FILE *marker = cbm_fopen(marker_file, "ab");
if (marker) {
(void)fputs(payload, marker);
(void)fclose(marker);
}
}
}
*worker_out = worker;
return 0;
}
static cbm_index_worker_poll_t app_fake_worker_poll(void *opaque,
cbm_daemon_application_worker_t handle,
const cbm_index_worker_result_t **result_out) {
(void)opaque;
app_fake_worker_t *worker = handle;
*result_out = NULL;
if (atomic_load(&worker->cancelled)) {
worker->result.outcome = CBM_PROC_KILLED;
worker->result.cancellation_requested = true;
worker->result.tree_quiesced = true;
*result_out = &worker->result;
return CBM_INDEX_WORKER_POLL_TERMINAL;
}
if (worker->context->project_locks && !worker->project_lock_lease) {
cbm_project_lock_lease_t *lease = NULL;
atomic_fetch_add(&worker->context->project_lock_attempts, 1);
cbm_private_file_lock_status_t status = cbm_project_lock_try_acquire(
worker->context->project_locks, worker->project_key, &lease);
if (status == CBM_PRIVATE_FILE_LOCK_OK && lease) {
worker->project_lock_lease = lease;
atomic_fetch_add(&worker->context->project_lock_acquisitions, 1);
} else {
(void)app_project_lock_release(&lease);
if (status == CBM_PRIVATE_FILE_LOCK_BUSY &&
!worker->context->project_lock_busy_is_error) {
return CBM_INDEX_WORKER_POLL_RUNNING;
}
return CBM_INDEX_WORKER_POLL_ERROR;
}
}
if (atomic_load(&worker->context->scripted)) {
cbm_proc_outcome_t outcome = worker->attempt < APP_FAKE_MAX_ATTEMPTS
? worker->context->outcomes[worker->attempt]
: CBM_PROC_SPAWN_FAILED;
worker->result.outcome = outcome;
worker->result.exit_code = outcome == CBM_PROC_CLEAN ? 0 : -1;
worker->result.tree_quiesced = true;
const char *response =
worker->attempt < APP_FAKE_MAX_ATTEMPTS && worker->context->responses[worker->attempt]
? worker->context->responses[worker->attempt]
: "{\"content\":[{\"type\":\"text\",\"text\":\"{"
"\\\"status\\\":\\\"indexed\\\"}\"}]}";
worker->result.response = outcome == CBM_PROC_CLEAN ? cbm_strdup(response) : NULL;
*result_out = &worker->result;
return CBM_INDEX_WORKER_POLL_TERMINAL;
}
if (atomic_load(&worker->context->allow_completion)) {
worker->result.outcome = CBM_PROC_CLEAN;
worker->result.exit_code = 0;
worker->result.tree_quiesced = !atomic_load(&worker->context->unsafe_clean);
worker->result.supervision_failed = atomic_load(&worker->context->unsafe_clean);
worker->result.response = cbm_strdup(
"{\"content\":[{\"type\":\"text\",\"text\":\"{\\\"status\\\":\\\"indexed\\\"}\"}]}");
*result_out = &worker->result;
return CBM_INDEX_WORKER_POLL_TERMINAL;
}
return CBM_INDEX_WORKER_POLL_RUNNING;
}
static bool app_fake_worker_cancel(void *opaque, cbm_daemon_application_worker_t handle) {
app_fake_worker_context_t *context = opaque;
app_fake_worker_t *worker = handle;
bool first = !atomic_exchange(&worker->cancelled, true);
if (first) {
atomic_fetch_add(&context->cancels, 1);
}
return true;
}
static const char *app_fake_worker_log_path(void *opaque, cbm_daemon_application_worker_t handle) {
(void)opaque;
(void)handle;
return "/tmp/cbm-fake-worker.log";
}
static void app_fake_worker_destroy(void *opaque, cbm_daemon_application_worker_t handle) {
app_fake_worker_context_t *context = opaque;
app_fake_worker_t *worker = handle;
atomic_fetch_add(&context->destroys, 1);
while (worker->attempt == atomic_load(&context->hold_destroy_attempt) &&
!atomic_load(&context->release_destroy)) {
cbm_usleep(1000);
}
(void)app_project_lock_release(&worker->project_lock_lease);
cbm_index_worker_result_free(&worker->result);
free(worker->project_key);
free(handle);
}
typedef struct {
cbm_daemon_runtime_application_callbacks_t callbacks;
cbm_daemon_runtime_application_session_t *session;
cbm_daemon_runtime_application_token_t request_token;
const uint8_t *request;
uint32_t request_length;
cbm_daemon_runtime_application_status_t status;
uint8_t *response;
uint32_t response_length;
atomic_bool done;
} app_request_thread_t;
static void *app_request_thread(void *opaque) {
app_request_thread_t *request = opaque;
request->status =
request->request_token == CBM_DAEMON_RUNTIME_APPLICATION_TOKEN_INVALID
? app_test_request(&request->callbacks, request->session, request->request,
request->request_length, &request->response,
&request->response_length)
: app_test_request_tagged(&request->callbacks, request->session, request->request_token,
request->request, request->request_length, &request->response,
&request->response_length);
atomic_store(&request->done, true);
return NULL;
}
static bool app_wait_for_subscribers(cbm_daemon_application_t *application, const char *project,
size_t expected) {
uint64_t deadline = cbm_now_ms() + APP_TEST_TIMEOUT_MS;
while (cbm_now_ms() < deadline) {
if (cbm_daemon_application_job_subscribers(application, project) == expected) {
return true;
}
cbm_usleep(1000);
}
return false;
}
static bool app_wait_for_atomic_int(atomic_int *value, int expected) {
uint64_t deadline = cbm_now_ms() + APP_TEST_TIMEOUT_MS;
while (cbm_now_ms() < deadline) {
if (atomic_load(value) == expected) {
return true;
}
cbm_usleep(1000);
}
return false;
}
static bool app_wait_for_atomic_int_at_least(atomic_int *value, int minimum) {
uint64_t deadline = cbm_now_ms() + APP_TEST_TIMEOUT_MS;
while (cbm_now_ms() < deadline) {
if (atomic_load(value) >= minimum) {
return true;
}
cbm_usleep(1000);
}
return false;
}
static bool app_wait_for_background_initializes(int minimum) {
/* Positive completion signal for the request-handler's background tail:
* negatives about auto-index admission ("nothing was started") are only
* meaningful once the admission pass has RUN — a fixed sleep loses that
* race in both directions. */
uint64_t deadline = cbm_now_ms() + APP_TEST_TIMEOUT_MS;
while (cbm_now_ms() < deadline) {
if (cbm_daemon_application_background_initializes_for_test() >= minimum) {
return true;
}
cbm_usleep(1000);
}
return false;
}
static bool app_wait_for_atomic_bool(atomic_bool *value, bool expected) {
uint64_t deadline = cbm_now_ms() + APP_TEST_TIMEOUT_MS;
while (cbm_now_ms() < deadline) {
if (atomic_load(value) == expected) {
return true;
}
cbm_usleep(1000);
}
return false;
}
static bool app_wait_for_active_jobs(cbm_daemon_application_t *application, size_t expected) {
uint64_t deadline = cbm_now_ms() + APP_TEST_TIMEOUT_MS;
while (cbm_now_ms() < deadline) {
if (cbm_daemon_application_active_jobs(application) == expected) {
return true;
}
cbm_usleep(1000);
}
return false;
}
typedef struct {
cbm_daemon_application_t *application;
const char *project;
const char *root;
int result;
} app_index_thread_t;
static void *app_index_thread(void *opaque) {
app_index_thread_t *request = opaque;
request->result =
cbm_daemon_application_index(request->application, request->project, request->root);
return NULL;
}
typedef struct {
bool had_cache;
char *saved_cache;
char root[APP_TEST_PATH_CAP];
char cache[APP_TEST_PATH_CAP];
char db_path[APP_TEST_PATH_CAP];
char *project;
app_fake_worker_context_t fake;
cbm_store_t *store;
cbm_watcher_t *watcher;
cbm_daemon_application_t *application;
cbm_daemon_runtime_application_callbacks_t callbacks;
cbm_daemon_runtime_application_session_t *session;
} app_watch_race_fixture_t;
static bool app_watch_race_fixture_init(app_watch_race_fixture_t *fixture,
cbm_daemon_client_id_t client_id) {
memset(fixture, 0, sizeof(*fixture));
app_fake_worker_context_init(&fixture->fake);
const char *old_cache = getenv("CBM_CACHE_DIR");
fixture->had_cache = old_cache != NULL;
fixture->saved_cache = old_cache ? cbm_strdup(old_cache) : NULL;
bool saved_cache_ok = !fixture->had_cache || fixture->saved_cache;
(void)snprintf(fixture->root, sizeof(fixture->root), "%s/cbm-app-watch-race-root-XXXXXX",
cbm_tmpdir());
(void)snprintf(fixture->cache, sizeof(fixture->cache), "%s/cbm-app-watch-race-cache-XXXXXX",
cbm_tmpdir());
bool dirs_ok = cbm_mkdtemp(fixture->root) != NULL && cbm_mkdtemp(fixture->cache) != NULL;
bool env_ok = saved_cache_ok && cbm_setenv("CBM_CACHE_DIR", fixture->cache, 1) == 0;
fixture->project = dirs_ok ? cbm_project_name_from_path(fixture->root) : NULL;
if (fixture->project) {
(void)snprintf(fixture->db_path, sizeof(fixture->db_path), "%s/%s.db", fixture->cache,
fixture->project);
}
bool db_ok = fixture->project && app_test_create_empty_file(fixture->db_path);
cbm_daemon_application_worker_ops_t worker_ops = {
.context = &fixture->fake,
.start = app_fake_worker_start,
.poll = app_fake_worker_poll,
.cancel = app_fake_worker_cancel,
.log_path = app_fake_worker_log_path,
.destroy = app_fake_worker_destroy,
};
fixture->store = cbm_store_open_memory();
fixture->watcher = cbm_watcher_new(fixture->store, app_test_index_noop, NULL);
cbm_daemon_application_config_t config = {
.watcher = fixture->watcher,
.worker_ops = &worker_ops,
};
fixture->application = cbm_daemon_application_new(&config);
fixture->callbacks = cbm_daemon_application_runtime_callbacks(fixture->application);
if (fixture->application) {
fixture->session = app_test_open(&fixture->callbacks, client_id);
}
uint8_t *context = NULL;
uint32_t context_length = 0;
uint8_t *ping = NULL;
uint32_t ping_length = 0;
uint8_t *response = NULL;
uint32_t response_length = 0;
bool encoded =
env_ok && dirs_ok && db_ok && fixture->store && fixture->watcher && fixture->application &&
fixture->session &&
app_test_context_request(fixture->root, fixture->root, &context, &context_length) &&
app_test_text_request(CBM_DAEMON_APPLICATION_REQUEST_MCP,
"{\"jsonrpc\":\"2.0\",\"id\":19,\"method\":\"ping\"}", &ping,
&ping_length);
bool initialized = encoded && app_test_request(&fixture->callbacks, fixture->session, context,
context_length, &response, &response_length) ==
CBM_DAEMON_RUNTIME_APPLICATION_OK;
free(response);
response = NULL;
response_length = 0;
initialized = initialized && app_test_request(&fixture->callbacks, fixture->session, ping,
ping_length, &response, &response_length) ==
CBM_DAEMON_RUNTIME_APPLICATION_OK;
free(response);
free(context);
free(ping);
return initialized && cbm_watcher_watch_count(fixture->watcher) == 1;
}
static bool app_watch_race_fixture_finish(app_watch_race_fixture_t *fixture) {
if (fixture->session) {
fixture->callbacks.session_close(fixture->callbacks.context, fixture->session);
fixture->session = NULL;
}
bool stopped = !fixture->application ||
cbm_daemon_application_shutdown(fixture->application, APP_TEST_TIMEOUT_MS);
if (stopped) {
cbm_daemon_application_free(fixture->application);
fixture->application = NULL;
cbm_watcher_stop(fixture->watcher);
cbm_watcher_free(fixture->watcher);
fixture->watcher = NULL;
cbm_store_close(fixture->store);
fixture->store = NULL;
}
free(fixture->project);
fixture->project = NULL;
(void)th_rmtree(fixture->root);
(void)th_rmtree(fixture->cache);
bool restored = fixture->saved_cache
? cbm_setenv("CBM_CACHE_DIR", fixture->saved_cache, 1) == 0
: (!fixture->had_cache && cbm_unsetenv("CBM_CACHE_DIR") == 0);
free(fixture->saved_cache);
fixture->saved_cache = NULL;
return stopped && restored;
}
typedef struct {
cbm_daemon_application_t *application;
const char *project;
const char *root;
bool pause_before_subscribe;
atomic_bool ready;
atomic_bool proceed;
atomic_bool done;
int result;
} app_watcher_index_thread_t;
static void *app_watcher_index_thread(void *opaque) {
app_watcher_index_thread_t *request = opaque;
if (request->pause_before_subscribe) {
atomic_store(&request->ready, true);
while (!atomic_load(&request->proceed)) {
cbm_usleep(1000);
}
}
request->result =
cbm_daemon_application_watcher_index(request->project, request->root, request->application);
atomic_store(&request->done, true);
return NULL;
}
typedef struct {
const char *name;
bool had_value;
char *value;
} app_env_backup_t;
static bool app_env_backup_capture(app_env_backup_t *backup, const char *name) {
memset(backup, 0, sizeof(*backup));
backup->name = name;
const char *value = getenv(name);
backup->had_value = value != NULL;
backup->value = value ? cbm_strdup(value) : NULL;
return !backup->had_value || backup->value != NULL;
}
static bool app_env_backup_restore(app_env_backup_t *backup) {
if (!backup || !backup->name) {
return true;
}
if (backup->had_value && !backup->value) {
backup->name = NULL;
return false;
}
int status =
backup->had_value ? cbm_setenv(backup->name, backup->value, 1) : cbm_unsetenv(backup->name);
free(backup->value);
backup->value = NULL;
backup->name = NULL;
return status == 0;
}
typedef struct {
atomic_int starts;
atomic_int start_failures_remaining;
atomic_int cancels;
atomic_int destroys;
atomic_bool allow_completion;
const char *latest_version;
} app_fake_update_context_t;
typedef struct {
app_fake_update_context_t *context;
atomic_bool cancelled;
} app_fake_update_worker_t;
static void app_fake_update_context_init(app_fake_update_context_t *context,
bool allow_completion) {
memset(context, 0, sizeof(*context));
atomic_init(&context->starts, 0);
atomic_init(&context->start_failures_remaining, 0);
atomic_init(&context->cancels, 0);
atomic_init(&context->destroys, 0);
atomic_init(&context->allow_completion, allow_completion);
context->latest_version = "v999.0.0";
}
static int app_fake_update_start(void *opaque, cbm_daemon_application_update_worker_t *worker_out) {
app_fake_update_context_t *context = opaque;
if (!context || !worker_out) {
return -1;
}
atomic_fetch_add(&context->starts, 1);
int failures = atomic_load(&context->start_failures_remaining);
while (failures > 0 && !atomic_compare_exchange_weak(&context->start_failures_remaining,
&failures, failures - 1)) {}
if (failures > 0) {
*worker_out = NULL;
return -1;
}
app_fake_update_worker_t *worker = calloc(1, sizeof(*worker));
if (!worker) {
return -1;
}
worker->context = context;
atomic_init(&worker->cancelled, false);
*worker_out = worker;
return 0;
}
static cbm_daemon_application_update_poll_t app_fake_update_poll(
void *opaque, cbm_daemon_application_update_worker_t handle, const char **latest_version_out) {
(void)opaque;
app_fake_update_worker_t *worker = handle;
*latest_version_out = NULL;
if (atomic_load(&worker->cancelled)) {
return CBM_DAEMON_APPLICATION_UPDATE_POLL_TERMINAL;
}
if (!atomic_load(&worker->context->allow_completion)) {
return CBM_DAEMON_APPLICATION_UPDATE_POLL_RUNNING;
}
*latest_version_out = worker->context->latest_version;
return CBM_DAEMON_APPLICATION_UPDATE_POLL_TERMINAL;
}
static bool app_fake_update_cancel(void *opaque, cbm_daemon_application_update_worker_t handle) {
app_fake_update_context_t *context = opaque;
app_fake_update_worker_t *worker = handle;
if (!atomic_exchange(&worker->cancelled, true)) {
atomic_fetch_add(&context->cancels, 1);
}
return true;
}
static void app_fake_update_destroy(void *opaque, cbm_daemon_application_update_worker_t handle) {
app_fake_update_context_t *context = opaque;
atomic_fetch_add(&context->destroys, 1);
free(handle);
}
static cbm_daemon_application_update_ops_t app_fake_update_ops(app_fake_update_context_t *context) {
return (cbm_daemon_application_update_ops_t){
.context = context,
.start = app_fake_update_start,
.poll = app_fake_update_poll,
.cancel = app_fake_update_cancel,
.destroy = app_fake_update_destroy,
};
}
static bool app_test_initialize_profile(const cbm_daemon_runtime_application_callbacks_t *callbacks,
cbm_daemon_runtime_application_session_t *session,
const char *root, cbm_mcp_tool_profile_t profile,
const char *hook_event, const char *hook_dialect) {
uint8_t *context = NULL;
uint32_t context_length = 0;
uint8_t *initialize = NULL;
uint32_t initialize_length = 0;
bool encoded =
session &&
app_test_context_request_options(root, root, profile, hook_event, hook_dialect, &context,
&context_length) &&
app_test_text_request(CBM_DAEMON_APPLICATION_REQUEST_MCP,
"{\"jsonrpc\":\"2.0\",\"id\":4100,\"method\":\"initialize\","
"\"params\":{}}",
&initialize, &initialize_length);
uint8_t *response = NULL;
uint32_t response_length = 0;
cbm_daemon_runtime_application_status_t context_status =
encoded ? app_test_request(callbacks, session, context, context_length, &response,
&response_length)
: CBM_DAEMON_RUNTIME_APPLICATION_TRANSPORT_ERROR;
free(response);
response = NULL;
response_length = 0;
cbm_daemon_runtime_application_status_t initialize_status =
context_status == CBM_DAEMON_RUNTIME_APPLICATION_OK
? app_test_request(callbacks, session, initialize, initialize_length, &response,
&response_length)
: CBM_DAEMON_RUNTIME_APPLICATION_TRANSPORT_ERROR;
bool initialized = initialize_status == CBM_DAEMON_RUNTIME_APPLICATION_OK && response &&
strstr((char *)response, "\"result\"");
free(response);
free(context);
free(initialize);
return initialized;
}
static cbm_daemon_runtime_application_status_t app_test_list_projects(
const cbm_daemon_runtime_application_callbacks_t *callbacks,
cbm_daemon_runtime_application_session_t *session, uint64_t id, uint8_t **response_out,
uint32_t *response_length_out) {
char message[256];
(void)snprintf(message, sizeof(message),
"{\"jsonrpc\":\"2.0\",\"id\":%llu,\"method\":\"tools/call\","
"\"params\":{\"name\":\"list_projects\",\"arguments\":{}}}",
(unsigned long long)id);
uint8_t *request = NULL;
uint32_t request_length = 0;
if (!app_test_text_request(CBM_DAEMON_APPLICATION_REQUEST_MCP, message, &request,
&request_length)) {
return CBM_DAEMON_RUNTIME_APPLICATION_TRANSPORT_ERROR;
}
cbm_daemon_runtime_application_status_t status = app_test_request(
callbacks, session, request, request_length, response_out, response_length_out);
free(request);
return status;
}
static bool app_wait_for_update_notice(const cbm_daemon_runtime_application_callbacks_t *callbacks,
cbm_daemon_runtime_application_session_t *session,
uint64_t *id, uint8_t **notice_out) {
uint64_t deadline = cbm_now_ms() + APP_TEST_TIMEOUT_MS;
while (cbm_now_ms() < deadline) {
uint8_t *response = NULL;
uint32_t response_length = 0;
cbm_daemon_runtime_application_status_t status =
app_test_list_projects(callbacks, session, (*id)++, &response, &response_length);
if (status == CBM_DAEMON_RUNTIME_APPLICATION_OK && response &&
strstr((char *)response, "Update available:")) {
*notice_out = response;
return true;
}
free(response);
cbm_usleep(1000);
}
*notice_out = NULL;
return false;
}
/* Regression contract: initialize is the ownership boundary for daemon background
* indexing. Only normal full MCP sessions participate; identical roots share
* one physical worker but retain one subscription per live session. */
TEST(daemon_application_initialize_coalesces_auto_index_for_full_sessions) {
app_env_backup_t cache_environment;
bool cache_saved = app_env_backup_capture(&cache_environment, "CBM_CACHE_DIR");
app_fake_update_context_t update;
app_fake_update_context_init(&update, false);
cbm_daemon_application_update_ops_t update_ops = app_fake_update_ops(&update);
char root[APP_TEST_PATH_CAP];
char cache[APP_TEST_PATH_CAP];
(void)snprintf(root, sizeof(root), "%s/cbm-app-auto-index-root-XXXXXX", cbm_tmpdir());
(void)snprintf(cache, sizeof(cache), "%s/cbm-app-auto-index-cache-XXXXXX", cbm_tmpdir());
bool dirs_ok = cbm_mkdtemp(root) != NULL && cbm_mkdtemp(cache) != NULL;
bool cache_set = dirs_ok && cache_saved && cbm_setenv("CBM_CACHE_DIR", cache, 1) == 0;
cbm_config_t *stored_config = cache_set ? cbm_config_open(cache) : NULL;
bool config_ready = stored_config &&
cbm_config_set(stored_config, CBM_CONFIG_AUTO_INDEX, "true") == 0 &&
cbm_config_set(stored_config, CBM_CONFIG_AUTO_WATCH, "false") == 0;
char canonical_root[APP_TEST_PATH_CAP] = {0};
bool canonical = dirs_ok && cbm_canonical_path(root, canonical_root, sizeof(canonical_root));
char *project = canonical ? cbm_project_name_from_path(canonical_root) : NULL;
app_fake_worker_context_t fake;
app_fake_worker_context_init(&fake);
cbm_daemon_application_worker_ops_t worker_ops = {
.context = &fake,
.start = app_fake_worker_start,
.poll = app_fake_worker_poll,
.cancel = app_fake_worker_cancel,
.log_path = app_fake_worker_log_path,
.destroy = app_fake_worker_destroy,
};
cbm_daemon_application_config_t config = {
.config = stored_config,
.worker_ops = &worker_ops,
.update_ops = &update_ops,
};
cbm_daemon_application_t *application =
config_ready ? cbm_daemon_application_new(&config) : NULL;
cbm_daemon_runtime_application_callbacks_t callbacks =
cbm_daemon_application_runtime_callbacks(application);
cbm_daemon_runtime_application_session_t *sessions[4] = {NULL, NULL, NULL, NULL};
for (size_t i = 0; application && i < 4; i++) {
sessions[i] = app_test_open(&callbacks, (cbm_daemon_client_id_t)(4110U + i));
}
int bg_baseline = cbm_daemon_application_background_initializes_for_test();
bool scout_initialized = app_test_initialize_profile(&callbacks, sessions[2], root,
CBM_MCP_TOOL_PROFILE_SCOUT, NULL, NULL);
bool hook_initialized = app_test_initialize_profile(
&callbacks, sessions[3], root, CBM_MCP_TOOL_PROFILE_ALL, "SessionStart", "copilot");
bool admissions_evaluated = app_wait_for_background_initializes(bg_baseline + 2);
bool restricted_started_nothing =
admissions_evaluated && atomic_load(&fake.starts) == 0 && application && project &&
cbm_daemon_application_job_subscribers(application, project) == 0 &&
atomic_load(&update.starts) == 0;
bool first_initialized = app_test_initialize_profile(&callbacks, sessions[0], root,
CBM_MCP_TOOL_PROFILE_ALL, NULL, NULL);
bool first_owned = first_initialized && project &&
app_wait_for_subscribers(application, project, 1) &&
app_wait_for_atomic_int(&fake.starts, 1);
bool second_initialized = app_test_initialize_profile(&callbacks, sessions[1], root,
CBM_MCP_TOOL_PROFILE_ALL, NULL, NULL);
bool coalesced = first_owned && second_initialized && project &&
app_wait_for_subscribers(application, project, 2) &&
cbm_daemon_application_active_jobs(application) == 1 &&
atomic_load(&fake.starts) == 1;
for (size_t i = 2; i < 4; i++) {
if (sessions[i]) {
callbacks.session_cancel(callbacks.context, sessions[i]);
callbacks.session_close(callbacks.context, sessions[i]);
sessions[i] = NULL;
}
}
bool restricted_disconnect_kept_job =
coalesced && project && cbm_daemon_application_job_subscribers(application, project) == 2 &&
atomic_load(&fake.cancels) == 0;
if (sessions[0]) {
callbacks.session_cancel(callbacks.context, sessions[0]);
callbacks.session_close(callbacks.context, sessions[0]);
sessions[0] = NULL;
}
bool one_owner_left = coalesced && project &&
app_wait_for_subscribers(application, project, 1) &&
atomic_load(&fake.cancels) == 0;
if (sessions[1]) {
callbacks.session_cancel(callbacks.context, sessions[1]);
callbacks.session_close(callbacks.context, sessions[1]);
sessions[1] = NULL;
}
bool final_cancelled = coalesced && app_wait_for_atomic_int(&fake.cancels, 1);
bool final_reaped = final_cancelled && app_wait_for_atomic_int(&fake.destroys, 1) &&
app_wait_for_active_jobs(application, 0);
for (size_t i = 0; i < 4; i++) {
if (sessions[i]) {
callbacks.session_cancel(callbacks.context, sessions[i]);
callbacks.session_close(callbacks.context, sessions[i]);
}
}
if (!final_reaped) {
atomic_store(&fake.allow_completion, true);
}
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
cbm_config_close(stored_config);
free(project);
(void)th_rmtree(root);
(void)th_rmtree(cache);
bool cache_restored = app_env_backup_restore(&cache_environment);
ASSERT_TRUE(cache_saved);
ASSERT_TRUE(dirs_ok);
ASSERT_TRUE(cache_set);
ASSERT_TRUE(config_ready);
ASSERT_TRUE(canonical);
ASSERT_TRUE(scout_initialized);
ASSERT_TRUE(hook_initialized);
ASSERT_TRUE(restricted_started_nothing);
ASSERT_TRUE(first_initialized);
ASSERT_TRUE(first_owned);
ASSERT_TRUE(second_initialized);
ASSERT_TRUE(coalesced);
ASSERT_TRUE(restricted_disconnect_kept_job);
ASSERT_TRUE(one_owner_left);
ASSERT_TRUE(final_cancelled);
ASSERT_TRUE(final_reaped);
ASSERT_EQ(atomic_load(&update.starts), 1);
ASSERT_EQ(atomic_load(&update.cancels), 1);
ASSERT_EQ(atomic_load(&update.destroys), 1);
ASSERT_TRUE(stopped);
ASSERT_TRUE(cache_restored);
PASS();
}
TEST(daemon_application_auto_index_honors_tracked_file_limit) {
app_env_backup_t cache_environment;
bool cache_saved = app_env_backup_capture(&cache_environment, "CBM_CACHE_DIR");
char root[APP_TEST_PATH_CAP];
char cache[APP_TEST_PATH_CAP];
char tracked[APP_TEST_PATH_CAP];
(void)snprintf(root, sizeof(root), "%s/cbm-app-auto-limit-root-XXXXXX", cbm_tmpdir());
(void)snprintf(cache, sizeof(cache), "%s/cbm-app-auto-limit-cache-XXXXXX", cbm_tmpdir());
bool dirs_ok = cbm_mkdtemp(root) != NULL && cbm_mkdtemp(cache) != NULL;
bool cache_set = dirs_ok && cache_saved && cbm_setenv("CBM_CACHE_DIR", cache, 1) == 0;
(void)snprintf(tracked, sizeof(tracked), "%s/tracked.c", root);
bool git_ready = dirs_ok && app_test_create_empty_file(tracked) &&
app_test_git(root, "init", "-q", NULL) == 0 &&
app_test_git(root, "add", "--", "tracked.c") == 0;
cbm_config_t *stored_config = cache_set ? cbm_config_open(cache) : NULL;
bool config_ready = stored_config &&
cbm_config_set(stored_config, CBM_CONFIG_AUTO_INDEX, "true") == 0 &&
cbm_config_set(stored_config, CBM_CONFIG_AUTO_INDEX_LIMIT, "0") == 0 &&
cbm_config_set(stored_config, CBM_CONFIG_AUTO_WATCH, "false") == 0;
app_fake_worker_context_t fake;
app_fake_worker_context_init(&fake);
cbm_daemon_application_worker_ops_t worker_ops = {
.context = &fake,
.start = app_fake_worker_start,
.poll = app_fake_worker_poll,
.cancel = app_fake_worker_cancel,
.log_path = app_fake_worker_log_path,
.destroy = app_fake_worker_destroy,
};
app_fake_update_context_t update;
app_fake_update_context_init(&update, true);
cbm_daemon_application_update_ops_t update_ops = app_fake_update_ops(&update);
cbm_daemon_application_config_t config = {
.config = stored_config,
.worker_ops = &worker_ops,
.update_ops = &update_ops,
};
cbm_daemon_application_t *application =
config_ready && git_ready ? cbm_daemon_application_new(&config) : NULL;
cbm_daemon_runtime_application_callbacks_t callbacks =
cbm_daemon_application_runtime_callbacks(application);
cbm_daemon_runtime_application_session_t *session =
application ? app_test_open(&callbacks, 4190) : NULL;
int bg_baseline = cbm_daemon_application_background_initializes_for_test();
bool initialized = app_test_initialize_profile(&callbacks, session, root,
CBM_MCP_TOOL_PROFILE_ALL, NULL, NULL);
bool admission_evaluated = app_wait_for_background_initializes(bg_baseline + 1);
bool limit_prevented_admission = initialized && admission_evaluated &&
atomic_load(&fake.starts) == 0 &&
cbm_daemon_application_active_jobs(application) == 0;
if (session) {
callbacks.session_cancel(callbacks.context, session);
callbacks.session_close(callbacks.context, session);
}
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
cbm_config_close(stored_config);
(void)th_rmtree(root);
(void)th_rmtree(cache);
bool cache_restored = app_env_backup_restore(&cache_environment);
ASSERT_TRUE(cache_saved);
ASSERT_TRUE(dirs_ok);
ASSERT_TRUE(cache_set);
ASSERT_TRUE(git_ready);
ASSERT_TRUE(config_ready);
ASSERT_TRUE(initialized);
ASSERT_TRUE(limit_prevented_admission);
ASSERT_TRUE(stopped);
ASSERT_TRUE(cache_restored);
PASS();
}
/* Native discovery never interprets a repository path as command text. Valid
* non-Git paths containing shell metacharacters remain countable literally. */
TEST(daemon_application_auto_index_file_count_handles_literal_metacharacter_path) {
char original[APP_TEST_PATH_CAP];
char literal[APP_TEST_PATH_CAP];
char tracked[APP_TEST_PATH_CAP];
(void)snprintf(original, sizeof(original), "%s/cbm-app-auto-count-XXXXXX", cbm_tmpdir());
bool root_ready = cbm_mkdtemp(original) != NULL;
(void)snprintf(tracked, sizeof(tracked), "%s/tracked.c", original);
bool source_ready = root_ready && app_test_create_empty_file(tracked);
int literal_written = snprintf(literal, sizeof(literal), "%s&literal", original);
bool renamed = source_ready && literal_written > 0 &&
(size_t)literal_written < sizeof(literal) && rename(original, literal) == 0;
int file_count = -1;
bool within_limit = renamed && cbm_mcp_auto_index_within_file_limit(literal, 1, &file_count);
(void)th_rmtree(renamed ? literal : original);
ASSERT_TRUE(root_ready);
ASSERT_TRUE(source_ready);
ASSERT_TRUE(renamed);
ASSERT_TRUE(within_limit);
ASSERT_EQ(file_count, 1);
PASS();
}
/* Auto-index is documented for every project, not only Git worktrees. The
* native count must preserve that behavior while enforcing the same limit. */
TEST(daemon_application_auto_index_file_count_supports_non_git_roots) {
char root[APP_TEST_PATH_CAP];
char source[APP_TEST_PATH_CAP];
(void)snprintf(root, sizeof(root), "%s/cbm-app-auto-count-non-git-XXXXXX", cbm_tmpdir());
bool root_ready = cbm_mkdtemp(root) != NULL;
(void)snprintf(source, sizeof(source), "%s/source.c", root);
bool source_ready = root_ready && app_test_create_empty_file(source);
int rejected_count = -1;
bool rejected = source_ready && cbm_mcp_auto_index_within_file_limit(root, 0, &rejected_count);
int admitted_count = -1;
bool admitted = source_ready && cbm_mcp_auto_index_within_file_limit(root, 1, &admitted_count);
(void)th_rmtree(root);
ASSERT_TRUE(root_ready);
ASSERT_TRUE(source_ready);
ASSERT_FALSE(rejected);
ASSERT_EQ(rejected_count, 1);
ASSERT_TRUE(admitted);
ASSERT_EQ(admitted_count, 1);
PASS();
}
TEST(daemon_application_auto_index_retries_transient_busy_admission) {
app_env_backup_t cache_environment;
bool cache_saved = app_env_backup_capture(&cache_environment, "CBM_CACHE_DIR");
char occupied_root[APP_TEST_PATH_CAP];
char auto_root[APP_TEST_PATH_CAP];
char cache[APP_TEST_PATH_CAP];
(void)snprintf(occupied_root, sizeof(occupied_root), "%s/cbm-app-auto-busy-first-XXXXXX",
cbm_tmpdir());
(void)snprintf(auto_root, sizeof(auto_root), "%s/cbm-app-auto-busy-second-XXXXXX",
cbm_tmpdir());
(void)snprintf(cache, sizeof(cache), "%s/cbm-app-auto-busy-cache-XXXXXX", cbm_tmpdir());
bool dirs_ok = cbm_mkdtemp(occupied_root) != NULL && cbm_mkdtemp(auto_root) != NULL &&
cbm_mkdtemp(cache) != NULL;
bool cache_set = dirs_ok && cache_saved && cbm_setenv("CBM_CACHE_DIR", cache, 1) == 0;
cbm_config_t *stored_config = cache_set ? cbm_config_open(cache) : NULL;
bool config_ready = stored_config &&
cbm_config_set(stored_config, CBM_CONFIG_AUTO_INDEX, "true") == 0 &&
cbm_config_set(stored_config, CBM_CONFIG_AUTO_WATCH, "false") == 0;
app_fake_worker_context_t fake;
app_fake_worker_context_init(&fake);
cbm_daemon_application_worker_ops_t worker_ops = {
.context = &fake,
.start = app_fake_worker_start,
.poll = app_fake_worker_poll,
.cancel = app_fake_worker_cancel,
.log_path = app_fake_worker_log_path,
.destroy = app_fake_worker_destroy,
};
app_fake_update_context_t update;
app_fake_update_context_init(&update, true);
cbm_daemon_application_update_ops_t update_ops = app_fake_update_ops(&update);
cbm_daemon_application_config_t config = {
.config = stored_config,
.worker_ops = &worker_ops,
.update_ops = &update_ops,
.physical_job_limit = 1,
};
cbm_daemon_application_t *application =
config_ready ? cbm_daemon_application_new(&config) : NULL;
app_index_thread_t occupied = {
.application = application,
.project = "auto-busy-first",
.root = occupied_root,
.result = -1,
};
cbm_thread_t occupied_thread;
bool occupied_started =
application && dirs_ok &&
cbm_thread_create(&occupied_thread, 0, app_index_thread, &occupied) == 0;
bool occupied_admitted = occupied_started && app_wait_for_atomic_int(&fake.starts, 1);
cbm_daemon_runtime_application_callbacks_t callbacks =
cbm_daemon_application_runtime_callbacks(application);
cbm_daemon_runtime_application_session_t *session =
application ? app_test_open(&callbacks, 4191) : NULL;
int bg_baseline = cbm_daemon_application_background_initializes_for_test();
bool initialized =
occupied_admitted && app_test_initialize_profile(&callbacks, session, auto_root,
CBM_MCP_TOOL_PROFILE_ALL, NULL, NULL);
bool admission_evaluated = app_wait_for_background_initializes(bg_baseline + 1);
bool initially_deferred = initialized && admission_evaluated && atomic_load(&fake.starts) == 1;
atomic_store(&fake.allow_completion, true);
if (occupied_started) {
(void)cbm_thread_join(&occupied_thread);
}
bool retried_without_request =
occupied.result == 0 && app_wait_for_atomic_int(&fake.starts, 2) &&
app_wait_for_atomic_int(&fake.destroys, 2) && app_wait_for_active_jobs(application, 0);
uint8_t *response = NULL;
uint32_t response_length = 0;
cbm_daemon_runtime_application_status_t retry_request =
initialized
? app_test_list_projects(&callbacks, session, 41910, &response, &response_length)
: CBM_DAEMON_RUNTIME_APPLICATION_TRANSPORT_ERROR;
free(response);
bool retry_admitted =
retried_without_request && retry_request == CBM_DAEMON_RUNTIME_APPLICATION_OK;
if (session) {
callbacks.session_cancel(callbacks.context, session);
callbacks.session_close(callbacks.context, session);
}
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
cbm_config_close(stored_config);
(void)th_rmtree(occupied_root);
(void)th_rmtree(auto_root);
(void)th_rmtree(cache);
bool cache_restored = app_env_backup_restore(&cache_environment);
ASSERT_TRUE(cache_saved);
ASSERT_TRUE(dirs_ok);
ASSERT_TRUE(cache_set);
ASSERT_TRUE(config_ready);
ASSERT_TRUE(occupied_started);
ASSERT_TRUE(occupied_admitted);
ASSERT_TRUE(initialized);
ASSERT_TRUE(initially_deferred);
ASSERT_EQ(occupied.result, 0);
ASSERT_TRUE(retried_without_request);
ASSERT_TRUE(retry_admitted);
ASSERT_TRUE(stopped);
ASSERT_TRUE(cache_restored);
PASS();
}
/* Regression contract: the daemon owns one update generation, not one update thread
* per MCP server. Its completed result is replayed exactly once to every
* eligible full session, including a session initialized after completion. */
TEST(daemon_application_update_generation_notifies_initial_and_late_sessions_once) {
app_fake_update_context_t update;
app_fake_update_context_init(&update, true);
cbm_daemon_application_update_ops_t update_ops = app_fake_update_ops(&update);
cbm_daemon_application_config_t config = {.update_ops = &update_ops};
char root[APP_TEST_PATH_CAP];
(void)snprintf(root, sizeof(root), "%s/cbm-app-update-root-XXXXXX", cbm_tmpdir());
bool root_ok = cbm_mkdtemp(root) != NULL;
cbm_daemon_application_t *application = root_ok ? cbm_daemon_application_new(&config) : NULL;
cbm_daemon_runtime_application_callbacks_t callbacks =
cbm_daemon_application_runtime_callbacks(application);
cbm_daemon_runtime_application_session_t *initial =
application ? app_test_open(&callbacks, 4211) : NULL;
bool initial_initialized = app_test_initialize_profile(&callbacks, initial, root,
CBM_MCP_TOOL_PROFILE_ALL, NULL, NULL);
uint64_t request_id = 42000;
uint8_t *initial_notice = NULL;
bool initial_notified =
initial_initialized &&
app_wait_for_update_notice(&callbacks, initial, &request_id, &initial_notice);
uint8_t *initial_second = NULL;
uint32_t initial_second_length = 0;
cbm_daemon_runtime_application_status_t initial_second_status =
initial_notified ? app_test_list_projects(&callbacks, initial, request_id++,
&initial_second, &initial_second_length)
: CBM_DAEMON_RUNTIME_APPLICATION_TRANSPORT_ERROR;
bool initial_once = initial_second_status == CBM_DAEMON_RUNTIME_APPLICATION_OK &&
initial_second && !strstr((char *)initial_second, "Update available:");
cbm_daemon_runtime_application_session_t *late =
application ? app_test_open(&callbacks, 4212) : NULL;
bool late_initialized =
initial_notified &&
app_test_initialize_profile(&callbacks, late, root, CBM_MCP_TOOL_PROFILE_ALL, NULL, NULL);
uint8_t *late_notice = NULL;
bool late_notified =
late_initialized && app_wait_for_update_notice(&callbacks, late, &request_id, &late_notice);
uint8_t *late_second = NULL;
uint32_t late_second_length = 0;
cbm_daemon_runtime_application_status_t late_second_status =
late_notified ? app_test_list_projects(&callbacks, late, request_id++, &late_second,
&late_second_length)
: CBM_DAEMON_RUNTIME_APPLICATION_TRANSPORT_ERROR;
bool late_once = late_second_status == CBM_DAEMON_RUNTIME_APPLICATION_OK && late_second &&
!strstr((char *)late_second, "Update available:");
int generation_starts = atomic_load(&update.starts);
bool generation_completed = initial_notified && app_wait_for_atomic_int(&update.destroys, 1);
if (initial) {
callbacks.session_cancel(callbacks.context, initial);
callbacks.session_close(callbacks.context, initial);
}
if (late) {
callbacks.session_cancel(callbacks.context, late);
callbacks.session_close(callbacks.context, late);
}
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
free(initial_notice);
free(initial_second);
free(late_notice);
free(late_second);
(void)th_rmtree(root);
ASSERT_TRUE(root_ok);
ASSERT_TRUE(initial_initialized);
ASSERT_TRUE(initial_notified);
ASSERT_TRUE(initial_once);
ASSERT_TRUE(late_initialized);
ASSERT_TRUE(late_notified);
ASSERT_TRUE(late_once);
ASSERT_EQ(generation_starts, 1);
ASSERT_TRUE(generation_completed);
ASSERT_TRUE(stopped);
ASSERT_EQ(atomic_load(&update.cancels), 0);
ASSERT_EQ(atomic_load(&update.destroys), 1);
PASS();
}
TEST(daemon_application_update_generation_retries_worker_start_failure) {
app_fake_update_context_t update;
app_fake_update_context_init(&update, true);
atomic_store(&update.start_failures_remaining, 1);
cbm_daemon_application_update_ops_t update_ops = app_fake_update_ops(&update);
cbm_daemon_application_config_t config = {.update_ops = &update_ops};
char root[APP_TEST_PATH_CAP];
(void)snprintf(root, sizeof(root), "%s/cbm-app-update-retry-start-XXXXXX", cbm_tmpdir());
bool root_ok = cbm_mkdtemp(root) != NULL;
cbm_daemon_application_t *application = root_ok ? cbm_daemon_application_new(&config) : NULL;
cbm_daemon_runtime_application_callbacks_t callbacks =
cbm_daemon_application_runtime_callbacks(application);
cbm_daemon_runtime_application_session_t *session =
application ? app_test_open(&callbacks, 4261) : NULL;
bool initialized = app_test_initialize_profile(&callbacks, session, root,
CBM_MCP_TOOL_PROFILE_ALL, NULL, NULL);
bool failed_generation_started = initialized && app_wait_for_atomic_int(&update.starts, 1);
uint64_t request_id = 42610;
uint8_t *notice = NULL;
bool retry_notified = failed_generation_started &&
app_wait_for_update_notice(&callbacks, session, &request_id, &notice);
if (session) {
callbacks.session_cancel(callbacks.context, session);
callbacks.session_close(callbacks.context, session);
}
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
free(notice);
(void)th_rmtree(root);
ASSERT_TRUE(root_ok);
ASSERT_TRUE(initialized);
ASSERT_TRUE(failed_generation_started);
ASSERT_TRUE(retry_notified);
ASSERT_EQ(atomic_load(&update.starts), 2);
ASSERT_EQ(atomic_load(&update.cancels), 0);
ASSERT_EQ(atomic_load(&update.destroys), 1);
ASSERT_TRUE(stopped);
PASS();
}
TEST(daemon_application_update_generation_retries_cancelled_check) {
app_fake_update_context_t update;
app_fake_update_context_init(&update, false);
cbm_daemon_application_update_ops_t update_ops = app_fake_update_ops(&update);
cbm_daemon_application_config_t config = {.update_ops = &update_ops};
char root[APP_TEST_PATH_CAP];
(void)snprintf(root, sizeof(root), "%s/cbm-app-update-retry-cancel-XXXXXX", cbm_tmpdir());
bool root_ok = cbm_mkdtemp(root) != NULL;
cbm_daemon_application_t *application = root_ok ? cbm_daemon_application_new(&config) : NULL;
cbm_daemon_runtime_application_callbacks_t callbacks =
cbm_daemon_application_runtime_callbacks(application);
cbm_daemon_runtime_application_session_t *first =
application ? app_test_open(&callbacks, 4271) : NULL;
cbm_daemon_runtime_application_session_t *scout =
application ? app_test_open(&callbacks, 4272) : NULL;
bool scout_initialized = app_test_initialize_profile(&callbacks, scout, root,
CBM_MCP_TOOL_PROFILE_SCOUT, NULL, NULL);
bool first_initialized =
scout_initialized &&
app_test_initialize_profile(&callbacks, first, root, CBM_MCP_TOOL_PROFILE_ALL, NULL, NULL);
bool first_started = first_initialized && app_wait_for_atomic_int(&update.starts, 1);
if (first) {
callbacks.session_cancel(callbacks.context, first);
callbacks.session_close(callbacks.context, first);
first = NULL;
}
bool first_cancelled =
first_started && atomic_load(&update.cancels) == 1 && atomic_load(&update.destroys) == 1;
atomic_store(&update.allow_completion, true);
cbm_daemon_runtime_application_session_t *retry =
application ? app_test_open(&callbacks, 4273) : NULL;
bool retry_initialized =
first_cancelled &&
app_test_initialize_profile(&callbacks, retry, root, CBM_MCP_TOOL_PROFILE_ALL, NULL, NULL);
uint64_t request_id = 42710;
uint8_t *notice = NULL;
bool retry_notified =
retry_initialized && app_wait_for_update_notice(&callbacks, retry, &request_id, &notice);
if (retry) {
callbacks.session_cancel(callbacks.context, retry);
callbacks.session_close(callbacks.context, retry);
}
if (scout) {
callbacks.session_cancel(callbacks.context, scout);
callbacks.session_close(callbacks.context, scout);
}
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
free(notice);
(void)th_rmtree(root);
ASSERT_TRUE(root_ok);
ASSERT_TRUE(scout_initialized);
ASSERT_TRUE(first_initialized);
ASSERT_TRUE(first_started);
ASSERT_TRUE(first_cancelled);
ASSERT_TRUE(retry_initialized);
ASSERT_TRUE(retry_notified);
ASSERT_EQ(atomic_load(&update.starts), 2);
ASSERT_EQ(atomic_load(&update.cancels), 1);
ASSERT_EQ(atomic_load(&update.destroys), 2);
ASSERT_TRUE(stopped);
PASS();
}
typedef struct {
cbm_daemon_runtime_application_callbacks_t callbacks;
cbm_daemon_runtime_application_session_t *session;
cbm_daemon_application_t *application;
bool shutdown_ok;
atomic_bool done;
} app_final_disconnect_thread_t;
static void *app_final_disconnect_thread(void *opaque) {
app_final_disconnect_thread_t *disconnect = opaque;
disconnect->callbacks.session_cancel(disconnect->callbacks.context, disconnect->session);
disconnect->callbacks.session_close(disconnect->callbacks.context, disconnect->session);
disconnect->shutdown_ok =
cbm_daemon_application_shutdown(disconnect->application, APP_TEST_TIMEOUT_MS);
atomic_store(&disconnect->done, true);
return NULL;
}
/* Regression contract: the last eligible disconnect is also the update-generation
* cancellation and join boundary. */
TEST(daemon_application_final_disconnect_cancels_and_joins_update_generation) {
app_fake_update_context_t update;
app_fake_update_context_init(&update, false);
cbm_daemon_application_update_ops_t update_ops = app_fake_update_ops(&update);
cbm_daemon_application_config_t config = {.update_ops = &update_ops};
char root[APP_TEST_PATH_CAP];
(void)snprintf(root, sizeof(root), "%s/cbm-app-update-cancel-root-XXXXXX", cbm_tmpdir());
bool root_ok = cbm_mkdtemp(root) != NULL;
cbm_daemon_application_t *application = root_ok ? cbm_daemon_application_new(&config) : NULL;
cbm_daemon_runtime_application_callbacks_t callbacks =
cbm_daemon_application_runtime_callbacks(application);
cbm_daemon_runtime_application_session_t *session =
application ? app_test_open(&callbacks, 4311) : NULL;
bool initialized = app_test_initialize_profile(&callbacks, session, root,
CBM_MCP_TOOL_PROFILE_ALL, NULL, NULL);
bool generation_started = initialized && app_wait_for_atomic_int(&update.starts, 1);
app_final_disconnect_thread_t disconnect = {
.callbacks = callbacks,
.session = session,
.application = application,
.shutdown_ok = false,
};
atomic_init(&disconnect.done, false);
cbm_thread_t disconnect_thread;
bool disconnect_started =
generation_started &&
cbm_thread_create(&disconnect_thread, 0, app_final_disconnect_thread, &disconnect) == 0;
bool joined_before_return =
disconnect_started && app_wait_for_atomic_bool(&disconnect.done, true);
if (disconnect_started) {
(void)cbm_thread_join(&disconnect_thread);
session = NULL;
} else if (session) {
callbacks.session_cancel(callbacks.context, session);
callbacks.session_close(callbacks.context, session);
session = NULL;
disconnect.shutdown_ok = cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
}
cbm_daemon_application_free(application);
(void)th_rmtree(root);
ASSERT_TRUE(root_ok);
ASSERT_TRUE(initialized);
ASSERT_TRUE(generation_started);
ASSERT_TRUE(disconnect_started);
ASSERT_TRUE(joined_before_return);
ASSERT_EQ(atomic_load(&update.cancels), 1);
ASSERT_EQ(atomic_load(&update.destroys), 1);
ASSERT_TRUE(disconnect.shutdown_ok);
PASS();
}
TEST(daemon_application_coalesces_semantically_identical_index_requests) {
app_fake_worker_context_t fake;
app_fake_worker_context_init(&fake);
cbm_daemon_application_worker_ops_t worker_ops = {
.context = &fake,
.start = app_fake_worker_start,
.poll = app_fake_worker_poll,
.cancel = app_fake_worker_cancel,
.log_path = app_fake_worker_log_path,
.destroy = app_fake_worker_destroy,
};
cbm_daemon_application_config_t config = {
.worker_ops = &worker_ops,
.physical_job_limit = 1,
};
cbm_daemon_application_t *application = cbm_daemon_application_new(&config);
cbm_daemon_runtime_application_callbacks_t callbacks =
cbm_daemon_application_runtime_callbacks(application);
cbm_daemon_runtime_application_session_t *sessions[2] = {app_test_open(&callbacks, 21),
app_test_open(&callbacks, 22)};
char root[APP_TEST_PATH_CAP];
snprintf(root, sizeof(root), "%s/cbm-app-coalesce-XXXXXX", cbm_tmpdir());
bool root_ok = cbm_mkdtemp(root) != NULL;
char *project = root_ok ? cbm_project_name_from_path(root) : NULL;
uint8_t *context = NULL;
uint32_t context_length = 0;
char first_args[APP_TEST_PATH_CAP + 96];
char second_args[APP_TEST_PATH_CAP + 96];
snprintf(first_args, sizeof(first_args),
"{\"repo_path\":\"%s\",\"mode\":\"full\","
"\"persistence\":false}",
root);
snprintf(second_args, sizeof(second_args),
"{ \"persistence\" : false, \"repo_path\" : \"%s\" }", root);
uint8_t *tools[2] = {NULL, NULL};
uint32_t tool_lengths[2] = {0, 0};
bool encoded =
root_ok && project && app_test_context_request(root, root, &context, &context_length) &&
app_test_tool_request("index_repository", first_args, &tools[0], &tool_lengths[0]) &&
app_test_tool_request("index_repository", second_args, &tools[1], &tool_lengths[1]);
bool contexts_ok = encoded;
uint8_t *empty = NULL;
uint32_t empty_length = 0;
for (size_t i = 0; contexts_ok && i < 2; i++) {
contexts_ok = app_test_request(&callbacks, sessions[i], context, context_length, &empty,
&empty_length) == CBM_DAEMON_RUNTIME_APPLICATION_OK;
free(empty);
empty = NULL;
}
app_request_thread_t requests[2] = {
{.callbacks = callbacks,
.session = sessions[0],
.request = tools[0],
.request_length = tool_lengths[0]},
{.callbacks = callbacks,
.session = sessions[1],
.request = tools[1],
.request_length = tool_lengths[1]},
};
cbm_thread_t threads[2];
bool thread_started[2] = {false, false};
if (contexts_ok) {
thread_started[0] =
cbm_thread_create(&threads[0], 0, app_request_thread, &requests[0]) == 0;
}
bool first_joined = thread_started[0] && app_wait_for_subscribers(application, project, 1);
if (first_joined) {
thread_started[1] =
cbm_thread_create(&threads[1], 0, app_request_thread, &requests[1]) == 0;
}
bool both_joined = thread_started[1] && app_wait_for_subscribers(application, project, 2);
atomic_store(&fake.allow_completion, true);
for (size_t i = 0; i < 2; i++) {
if (thread_started[i]) {
(void)cbm_thread_join(&threads[i]);
}
}
for (size_t i = 0; i < 2; i++) {
callbacks.session_close(callbacks.context, sessions[i]);
}
(void)cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
ASSERT_TRUE(contexts_ok);
ASSERT_TRUE(first_joined);
ASSERT_TRUE(both_joined);
ASSERT_EQ(atomic_load(&fake.starts), 1);
ASSERT_EQ(atomic_load(&fake.cancels), 0);
ASSERT_EQ(atomic_load(&fake.destroys), 1);
ASSERT_EQ(requests[0].status, CBM_DAEMON_RUNTIME_APPLICATION_OK);
ASSERT_EQ(requests[1].status, CBM_DAEMON_RUNTIME_APPLICATION_OK);
ASSERT_TRUE(requests[0].response && strstr((char *)requests[0].response, "indexed"));
ASSERT_TRUE(requests[1].response && strstr((char *)requests[1].response, "indexed"));
free(requests[0].response);
free(requests[1].response);
free(context);
free(tools[0]);
free(tools[1]);
free(project);
(void)cbm_rmdir(root);
PASS();
}
TEST(daemon_application_fresh_request_does_not_reuse_terminal_subscribed_job) {
enum { PRIOR_SUBSCRIBERS = 16 };
static const char stale_response[] =
"{\"content\":[{\"type\":\"text\",\"text\":\"stale-generation\"}]}";
static const char fresh_response[] =
"{\"content\":[{\"type\":\"text\",\"text\":\"fresh-generation\"}]}";
app_fake_worker_context_t fake;
app_fake_worker_context_init(&fake);
atomic_store(&fake.scripted, true);
atomic_store(&fake.hold_destroy_attempt, 0);
fake.outcomes[0] = CBM_PROC_CLEAN;
fake.outcomes[1] = CBM_PROC_CLEAN;
fake.responses[0] = stale_response;
fake.responses[1] = fresh_response;
cbm_daemon_application_worker_ops_t worker_ops = {
.context = &fake,
.start = app_fake_worker_start,
.poll = app_fake_worker_poll,
.cancel = app_fake_worker_cancel,
.log_path = app_fake_worker_log_path,
.destroy = app_fake_worker_destroy,
};
cbm_daemon_application_config_t config = {.worker_ops = &worker_ops};
cbm_daemon_application_t *application = cbm_daemon_application_new(&config);
cbm_daemon_runtime_application_callbacks_t callbacks =
cbm_daemon_application_runtime_callbacks(application);
cbm_daemon_runtime_application_session_t *sessions[PRIOR_SUBSCRIBERS + 1] = {0};
bool sessions_ok = application != NULL;
for (size_t i = 0; sessions_ok && i < PRIOR_SUBSCRIBERS + 1U; i++) {
sessions[i] = app_test_open(&callbacks, (cbm_daemon_client_id_t)(51U + i));
sessions_ok = sessions[i] != NULL;
}
char root[APP_TEST_PATH_CAP];
snprintf(root, sizeof(root), "%s/cbm-app-terminal-fresh-XXXXXX", cbm_tmpdir());
bool root_ok = cbm_mkdtemp(root) != NULL;
char *project = root_ok ? cbm_project_name_from_path(root) : NULL;
uint8_t *context = NULL;
uint32_t context_length = 0;
char args[APP_TEST_PATH_CAP + 32];
snprintf(args, sizeof(args), "{\"repo_path\":\"%s\"}", root);
uint8_t *tool = NULL;
uint32_t tool_length = 0;
bool setup = sessions_ok && root_ok && project &&
app_test_context_request(root, root, &context, &context_length) &&
app_test_tool_request("index_repository", args, &tool, &tool_length);
for (size_t i = 0; setup && i < PRIOR_SUBSCRIBERS + 1U; i++) {
uint8_t *empty = NULL;
uint32_t empty_length = 0;
setup = app_test_request(&callbacks, sessions[i], context, context_length, &empty,
&empty_length) == CBM_DAEMON_RUNTIME_APPLICATION_OK;
free(empty);
}
app_request_thread_t prior[PRIOR_SUBSCRIBERS];
cbm_thread_t threads[PRIOR_SUBSCRIBERS];
bool started[PRIOR_SUBSCRIBERS] = {0};
memset(prior, 0, sizeof(prior));
bool all_subscribed = setup;
for (size_t i = 0; all_subscribed && i < PRIOR_SUBSCRIBERS; i++) {
prior[i] = (app_request_thread_t){
.callbacks = callbacks,
.session = sessions[i],
.request = tool,
.request_length = tool_length,
};
atomic_init(&prior[i].done, false);
started[i] = cbm_thread_create(&threads[i], 0, app_request_thread, &prior[i]) == 0;
all_subscribed = started[i] && app_wait_for_subscribers(application, project, i + 1U);
if (all_subscribed) {
cbm_usleep(1000);
}
}
bool first_worker_ready_to_publish =
all_subscribed && app_wait_for_atomic_int(&fake.destroys, 1);
atomic_store(&fake.release_destroy, true);
/* Wait only for the stable end-state. Publish flips terminal and lets the
* blocked prior requests drain in the same breath, so "terminal AND still
* subscribed" is a transient window no poll cadence can pin (release runs
* 30305464193 and 30309182389 missed it from both directions). The
* production guard ignores subscriber counts — find_active_job skips any
* terminal job — and the stale/fresh assertions below catch a reuse in
* every interleaving. */
bool job_terminal = first_worker_ready_to_publish && app_wait_for_active_jobs(application, 0);
uint8_t *fresh = NULL;
uint32_t fresh_length = 0;
cbm_daemon_runtime_application_status_t fresh_status =
job_terminal ? app_test_request(&callbacks, sessions[PRIOR_SUBSCRIBERS], tool, tool_length,
&fresh, &fresh_length)
: CBM_DAEMON_RUNTIME_APPLICATION_REJECTED;
for (size_t i = 0; i < PRIOR_SUBSCRIBERS; i++) {
if (started[i]) {
(void)cbm_thread_join(&threads[i]);
}
}
for (size_t i = 0; i < PRIOR_SUBSCRIBERS + 1U; i++) {
if (sessions[i]) {
callbacks.session_close(callbacks.context, sessions[i]);
}
}
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
ASSERT_TRUE(setup);
ASSERT_TRUE(all_subscribed);
ASSERT_TRUE(first_worker_ready_to_publish);
ASSERT_TRUE(job_terminal);
ASSERT_EQ(fresh_status, CBM_DAEMON_RUNTIME_APPLICATION_OK);
ASSERT_EQ(atomic_load(&fake.starts), 2);
ASSERT_EQ(atomic_load(&fake.destroys), 2);
ASSERT_TRUE(fresh && strstr((char *)fresh, "fresh-generation"));
ASSERT_TRUE(!fresh || !strstr((char *)fresh, "stale-generation"));
ASSERT_TRUE(!fresh || !strstr((char *)fresh, "different options"));
ASSERT_TRUE(stopped);
for (size_t i = 0; i < PRIOR_SUBSCRIBERS; i++) {
ASSERT_EQ(prior[i].status, CBM_DAEMON_RUNTIME_APPLICATION_OK);
ASSERT_TRUE(prior[i].response && strstr((char *)prior[i].response, "stale-generation"));
free(prior[i].response);
}
free(fresh);
free(context);
free(tool);
free(project);
(void)cbm_rmdir(root);
PASS();
}
TEST(daemon_application_request_cancel_detaches_only_one_coalesced_subscriber) {
app_fake_worker_context_t fake;
app_fake_worker_context_init(&fake);
cbm_daemon_application_worker_ops_t worker_ops = {
.context = &fake,
.start = app_fake_worker_start,
.poll = app_fake_worker_poll,
.cancel = app_fake_worker_cancel,
.log_path = app_fake_worker_log_path,
.destroy = app_fake_worker_destroy,
};
cbm_daemon_application_config_t config = {.worker_ops = &worker_ops};
cbm_daemon_application_t *application = cbm_daemon_application_new(&config);
cbm_daemon_runtime_application_callbacks_t callbacks =
cbm_daemon_application_runtime_callbacks(application);
cbm_daemon_runtime_application_session_t *sessions[2] = {app_test_open(&callbacks, 3011),
app_test_open(&callbacks, 3012)};
char root[APP_TEST_PATH_CAP];
(void)snprintf(root, sizeof(root), "%s/cbm-app-request-cancel-coalesce-XXXXXX", cbm_tmpdir());
bool root_ok = cbm_mkdtemp(root) != NULL;
char *project = root_ok ? cbm_project_name_from_path(root) : NULL;
uint8_t *context = NULL;
uint32_t context_length = 0;
char args[APP_TEST_PATH_CAP + 32];
(void)snprintf(args, sizeof(args), "{\"repo_path\":\"%s\"}", root);
uint8_t *tool = NULL;
uint32_t tool_length = 0;
bool setup = application && sessions[0] && sessions[1] && root_ok && project &&
app_test_context_request(root, root, &context, &context_length) &&
app_test_tool_request("index_repository", args, &tool, &tool_length);
uint8_t *empty = NULL;
uint32_t empty_length = 0;
for (size_t i = 0; setup && i < 2; i++) {
setup = app_test_request(&callbacks, sessions[i], context, context_length, &empty,
&empty_length) == CBM_DAEMON_RUNTIME_APPLICATION_OK;
free(empty);
empty = NULL;
empty_length = 0;
}
app_request_thread_t requests[2] = {
{.callbacks = callbacks,
.session = sessions[0],
.request_token = UINT64_C(8101),
.request = tool,
.request_length = tool_length},
{.callbacks = callbacks,
.session = sessions[1],
.request_token = UINT64_C(8102),
.request = tool,
.request_length = tool_length},
};
atomic_init(&requests[0].done, false);
atomic_init(&requests[1].done, false);
cbm_thread_t threads[2];
bool started[2] = {false, false};
if (setup) {
started[0] = cbm_thread_create(&threads[0], 0, app_request_thread, &requests[0]) == 0;
}
bool first_joined = started[0] && app_wait_for_subscribers(application, project, 1);
if (first_joined) {
started[1] = cbm_thread_create(&threads[1], 0, app_request_thread, &requests[1]) == 0;
}
bool both_joined = started[1] && app_wait_for_subscribers(application, project, 2);
if (both_joined) {
callbacks.request_cancel(callbacks.context, sessions[0], requests[0].request_token);
}
bool one_left = both_joined && app_wait_for_subscribers(application, project, 1);
bool cancelled_returned = one_left && app_wait_for_atomic_bool(&requests[0].done, true);
int cancels_after_first = atomic_load(&fake.cancels);
/* Always release the fake worker so a failed RED assertion cannot strand
* either joinable request thread. */
atomic_store(&fake.allow_completion, true);
bool thread_joined[2] = {false, false};
for (size_t i = 0; i < 2; i++) {
if (started[i]) {
thread_joined[i] = cbm_thread_join(&threads[i]) == 0;
}
}
for (size_t i = 0; i < 2; i++) {
if (sessions[i]) {
callbacks.session_close(callbacks.context, sessions[i]);
}
}
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
bool second_indexed =
requests[1].response && strstr((char *)requests[1].response, "indexed") != NULL;
int final_cancels = atomic_load(&fake.cancels);
int final_starts = atomic_load(&fake.starts);
int final_destroys = atomic_load(&fake.destroys);
free(requests[0].response);
free(requests[1].response);
free(context);
free(tool);
free(project);
(void)cbm_rmdir(root);
ASSERT_TRUE(setup);
ASSERT_TRUE(first_joined);
ASSERT_TRUE(both_joined);
ASSERT_TRUE(one_left);
ASSERT_TRUE(cancelled_returned);
ASSERT_EQ(cancels_after_first, 0);
ASSERT_TRUE(thread_joined[0]);
ASSERT_TRUE(thread_joined[1]);
ASSERT_EQ(requests[0].status, CBM_DAEMON_RUNTIME_APPLICATION_CANCELLED);
ASSERT_NULL(requests[0].response);
ASSERT_EQ(requests[0].response_length, 0);
ASSERT_EQ(requests[1].status, CBM_DAEMON_RUNTIME_APPLICATION_OK);
ASSERT_TRUE(second_indexed);
ASSERT_EQ(final_cancels, 0);
ASSERT_EQ(final_starts, 1);
ASSERT_EQ(final_destroys, 1);
ASSERT_TRUE(stopped);
PASS();
}
TEST(daemon_application_cancels_physical_job_only_after_final_session) {
app_fake_worker_context_t fake;
app_fake_worker_context_init(&fake);
cbm_daemon_application_worker_ops_t worker_ops = {
.context = &fake,
.start = app_fake_worker_start,
.poll = app_fake_worker_poll,
.cancel = app_fake_worker_cancel,
.log_path = app_fake_worker_log_path,
.destroy = app_fake_worker_destroy,
};
cbm_daemon_application_config_t config = {.worker_ops = &worker_ops};
cbm_daemon_application_t *application = cbm_daemon_application_new(&config);
cbm_daemon_runtime_application_callbacks_t callbacks =
cbm_daemon_application_runtime_callbacks(application);
cbm_daemon_runtime_application_session_t *sessions[2] = {app_test_open(&callbacks, 31),
app_test_open(&callbacks, 32)};
char root[APP_TEST_PATH_CAP];
snprintf(root, sizeof(root), "%s/cbm-app-cancel-XXXXXX", cbm_tmpdir());
bool root_ok = cbm_mkdtemp(root) != NULL;
char *project = root_ok ? cbm_project_name_from_path(root) : NULL;
uint8_t *context = NULL;
uint32_t context_length = 0;
char args[APP_TEST_PATH_CAP + 32];
snprintf(args, sizeof(args), "{\"repo_path\":\"%s\"}", root);
uint8_t *tool = NULL;
uint32_t tool_length = 0;
bool setup = root_ok && project &&
app_test_context_request(root, root, &context, &context_length) &&
app_test_tool_request("index_repository", args, &tool, &tool_length);
uint8_t *empty = NULL;
uint32_t empty_length = 0;
for (size_t i = 0; setup && i < 2; i++) {
setup = app_test_request(&callbacks, sessions[i], context, context_length, &empty,
&empty_length) == CBM_DAEMON_RUNTIME_APPLICATION_OK;
free(empty);
empty = NULL;
}
app_request_thread_t requests[2] = {
{.callbacks = callbacks,
.session = sessions[0],
.request = tool,
.request_length = tool_length},
{.callbacks = callbacks,
.session = sessions[1],
.request = tool,
.request_length = tool_length},
};
cbm_thread_t threads[2];
bool started[2] = {false, false};
for (size_t i = 0; setup && i < 2; i++) {
started[i] = cbm_thread_create(&threads[i], 0, app_request_thread, &requests[i]) == 0;
}
bool subscribed = started[0] && started[1] && app_wait_for_subscribers(application, project, 2);
/* The physical job starts asynchronously once a session subscribes, so a
* subscriber count of 2 does NOT imply it has started. Cancelling both
* sessions before that point leaves starts == 0 -- arguably the correct
* outcome, and the reason this test failed intermittently on Windows while
* passing everywhere else. Wait for the state the assertions below actually
* require; `starts` only increments, so this cannot miss the transition. */
bool job_started = subscribed && app_wait_for_atomic_int(&fake.starts, 1);
if (job_started) {
callbacks.session_cancel(callbacks.context, sessions[0]);
}
bool one_left = job_started && app_wait_for_subscribers(application, project, 1);
int cancels_after_first = atomic_load(&fake.cancels);
if (one_left) {
callbacks.session_cancel(callbacks.context, sessions[1]);
}
for (size_t i = 0; i < 2; i++) {
if (started[i]) {
(void)cbm_thread_join(&threads[i]);
}
callbacks.session_close(callbacks.context, sessions[i]);
}
(void)cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
ASSERT_TRUE(setup);
ASSERT_TRUE(subscribed);
ASSERT_TRUE(job_started);
ASSERT_TRUE(one_left);
ASSERT_EQ(atomic_load(&fake.starts), 1);
ASSERT_EQ(cancels_after_first, 0);
ASSERT_EQ(atomic_load(&fake.cancels), 1);
ASSERT_EQ(atomic_load(&fake.destroys), 1);
ASSERT_EQ(requests[0].status, CBM_DAEMON_RUNTIME_APPLICATION_CANCELLED);
ASSERT_NULL(requests[0].response);
ASSERT_EQ(requests[1].status, CBM_DAEMON_RUNTIME_APPLICATION_CANCELLED);
ASSERT_NULL(requests[1].response);
free(requests[0].response);
free(requests[1].response);
free(context);
free(tool);
free(project);
(void)cbm_rmdir(root);
PASS();
}
TEST(daemon_application_disconnect_before_request_callback_is_sticky) {
app_fake_worker_context_t fake;
app_fake_worker_context_init(&fake);
atomic_store(&fake.scripted, true);
fake.outcomes[0] = CBM_PROC_CLEAN;
cbm_daemon_application_worker_ops_t worker_ops = {
.context = &fake,
.start = app_fake_worker_start,
.poll = app_fake_worker_poll,
.cancel = app_fake_worker_cancel,
.log_path = app_fake_worker_log_path,
.destroy = app_fake_worker_destroy,
};
cbm_daemon_application_config_t config = {.worker_ops = &worker_ops};
cbm_daemon_application_t *application = cbm_daemon_application_new(&config);
cbm_daemon_runtime_application_callbacks_t callbacks =
cbm_daemon_application_runtime_callbacks(application);
cbm_daemon_runtime_application_session_t *cancelled_session = app_test_open(&callbacks, 37);
cbm_daemon_runtime_application_session_t *healthy_session = app_test_open(&callbacks, 38);
char root[APP_TEST_PATH_CAP];
snprintf(root, sizeof(root), "%s/cbm-app-pre-callback-cancel-XXXXXX", cbm_tmpdir());
bool root_ok = cbm_mkdtemp(root) != NULL;
uint8_t *context = NULL;
uint32_t context_length = 0;
char args[APP_TEST_PATH_CAP + 32];
snprintf(args, sizeof(args), "{\"repo_path\":\"%s\"}", root);
uint8_t *tool = NULL;
uint32_t tool_length = 0;
bool setup = application && cancelled_session && healthy_session && root_ok &&
app_test_context_request(root, root, &context, &context_length) &&
app_test_tool_request("index_repository", args, &tool, &tool_length);
uint8_t *response = NULL;
uint32_t response_length = 0;
for (size_t i = 0; setup && i < 2; i++) {
cbm_daemon_runtime_application_session_t *session =
i == 0 ? cancelled_session : healthy_session;
setup = app_test_request(&callbacks, session, context, context_length, &response,
&response_length) == CBM_DAEMON_RUNTIME_APPLICATION_OK;
free(response);
response = NULL;
response_length = 0;
}
if (setup) {
/* Runtime disconnect may win immediately after it creates the request
* thread but before that thread enters the application callback. */
callbacks.session_cancel(callbacks.context, cancelled_session);
}
cbm_daemon_runtime_application_status_t cancelled_status =
setup ? app_test_request(&callbacks, cancelled_session, tool, tool_length, &response,
&response_length)
: CBM_DAEMON_RUNTIME_APPLICATION_HANDLER_ERROR;
uint8_t *cancelled_response = response;
uint32_t cancelled_response_length = response_length;
response = NULL;
response_length = 0;
int starts_after_cancelled_callback = atomic_load(&fake.starts);
cbm_daemon_runtime_application_status_t healthy_status =
setup ? app_test_request(&callbacks, healthy_session, tool, tool_length, &response,
&response_length)
: CBM_DAEMON_RUNTIME_APPLICATION_HANDLER_ERROR;
if (cancelled_session) {
callbacks.session_close(callbacks.context, cancelled_session);
}
if (healthy_session) {
callbacks.session_close(callbacks.context, healthy_session);
}
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
ASSERT_TRUE(setup);
ASSERT_EQ(cancelled_status, CBM_DAEMON_RUNTIME_APPLICATION_REJECTED);
ASSERT_NULL(cancelled_response);
ASSERT_EQ(cancelled_response_length, 0);
ASSERT_EQ(starts_after_cancelled_callback, 0);
ASSERT_EQ(healthy_status, CBM_DAEMON_RUNTIME_APPLICATION_OK);
ASSERT_TRUE(response && strstr((char *)response, "indexed"));
ASSERT_EQ(atomic_load(&fake.starts), 1);
ASSERT_EQ(atomic_load(&fake.cancels), 0);
ASSERT_EQ(atomic_load(&fake.destroys), 1);
ASSERT_TRUE(stopped);
free(cancelled_response);
free(response);
free(context);
free(tool);
(void)cbm_rmdir(root);
PASS();
}
TEST(daemon_application_request_cancel_preserves_persistent_watch_and_session) {
app_watch_race_fixture_t fixture;
bool fixture_ready = app_watch_race_fixture_init(&fixture, 3021);
char args[APP_TEST_PATH_CAP + 32];
(void)snprintf(args, sizeof(args), "{\"repo_path\":\"%s\"}", fixture.root);
uint8_t *tool = NULL;
uint32_t tool_length = 0;
uint8_t *ping = NULL;
uint32_t ping_length = 0;
bool encoded = fixture_ready &&
app_test_tool_request("index_repository", args, &tool, &tool_length) &&
app_test_text_request(CBM_DAEMON_APPLICATION_REQUEST_MCP,
"{\"jsonrpc\":\"2.0\",\"id\":3021,\"method\":\"ping\"}",
&ping, &ping_length);
app_request_thread_t request = {
.callbacks = fixture.callbacks,
.session = fixture.session,
.request_token = UINT64_C(8201),
.request = tool,
.request_length = tool_length,
.status = CBM_DAEMON_RUNTIME_APPLICATION_HANDLER_ERROR,
};
atomic_init(&request.done, false);
cbm_thread_t request_thread;
bool thread_started =
encoded && cbm_thread_create(&request_thread, 0, app_request_thread, &request) == 0;
bool subscribed =
thread_started && app_wait_for_subscribers(fixture.application, fixture.project, 1);
/* Cancel only after the WORKER exists: subscription precedes worker
* start, and a cancel that wins that window legally aborts the job with
* no worker (and no worker-cancel op) ever invoked — that interleaving
* has its own deterministic test below. From worker start onward, cancel
* delivery to the worker is guaranteed. */
bool worker_started = subscribed && app_wait_for_atomic_int_at_least(&fixture.fake.starts, 1);
if (worker_started) {
fixture.callbacks.request_cancel(fixture.callbacks.context, fixture.session,
request.request_token);
}
bool request_returned = worker_started && app_wait_for_atomic_bool(&request.done, true);
if (!request_returned) {
/* Keep cleanup bounded if the RED path fails to detach the request. */
atomic_store(&fixture.fake.allow_completion, true);
}
bool thread_joined = thread_started && cbm_thread_join(&request_thread) == 0;
int watch_count_after_cancel = fixture.watcher ? cbm_watcher_watch_count(fixture.watcher) : -1;
uint8_t *ping_response = NULL;
uint32_t ping_response_length = 0;
cbm_daemon_runtime_application_status_t ping_status =
CBM_DAEMON_RUNTIME_APPLICATION_TRANSPORT_ERROR;
if (thread_joined && encoded) {
ping_status =
app_test_request_tagged(&fixture.callbacks, fixture.session, UINT64_C(8202), ping,
ping_length, &ping_response, &ping_response_length);
}
int watch_count_after_ping = fixture.watcher ? cbm_watcher_watch_count(fixture.watcher) : -1;
bool worker_cancelled = app_wait_for_atomic_int(&fixture.fake.cancels, 1);
bool cleaned = app_watch_race_fixture_finish(&fixture);
bool ping_matches = ping_response && strstr((char *)ping_response, "\"id\":3021") != NULL;
int worker_starts = atomic_load(&fixture.fake.starts);
int worker_destroys = atomic_load(&fixture.fake.destroys);
free(request.response);
free(ping_response);
free(tool);
free(ping);
ASSERT_TRUE(fixture_ready);
ASSERT_TRUE(encoded);
ASSERT_TRUE(thread_started);
ASSERT_TRUE(subscribed);
ASSERT_TRUE(worker_started);
ASSERT_TRUE(request_returned);
ASSERT_TRUE(thread_joined);
ASSERT_EQ(request.status, CBM_DAEMON_RUNTIME_APPLICATION_CANCELLED);
ASSERT_NULL(request.response);
ASSERT_EQ(request.response_length, 0);
ASSERT_EQ(watch_count_after_cancel, 1);
ASSERT_EQ(ping_status, CBM_DAEMON_RUNTIME_APPLICATION_OK);
ASSERT_TRUE(ping_matches);
ASSERT_GT(ping_response_length, 0);
ASSERT_EQ(watch_count_after_ping, 1);
ASSERT_TRUE(worker_cancelled);
ASSERT_EQ(worker_starts, 1);
ASSERT_EQ(worker_destroys, 1);
ASSERT_TRUE(cleaned);
PASS();
}
/* Deterministic sibling of the test above: with every job thread parked
* before its first pre-start cancel check, the cancel WINS the pre-start
* window by construction — the request must still report CANCELLED with the
* watch preserved, while the worker is never started and none of its ops
* are ever invoked. A loaded CI runner produced this interleaving by
* chance; here it is forced (CI is never a lottery). */
TEST(daemon_application_cancel_before_worker_start_skips_worker) {
app_watch_race_fixture_t fixture;
bool fixture_ready = app_watch_race_fixture_init(&fixture, 3022);
char args[APP_TEST_PATH_CAP + 32];
(void)snprintf(args, sizeof(args), "{\"repo_path\":\"%s\"}", fixture.root);
uint8_t *tool = NULL;
uint32_t tool_length = 0;
bool encoded =
fixture_ready && app_test_tool_request("index_repository", args, &tool, &tool_length);
cbm_daemon_application_hold_job_before_start_for_test(true);
app_request_thread_t request = {
.callbacks = fixture.callbacks,
.session = fixture.session,
.request_token = UINT64_C(8301),
.request = tool,
.request_length = tool_length,
.status = CBM_DAEMON_RUNTIME_APPLICATION_HANDLER_ERROR,
};
atomic_init(&request.done, false);
cbm_thread_t request_thread;
bool thread_started =
encoded && cbm_thread_create(&request_thread, 0, app_request_thread, &request) == 0;
bool subscribed =
thread_started && app_wait_for_subscribers(fixture.application, fixture.project, 1);
if (subscribed) {
fixture.callbacks.request_cancel(fixture.callbacks.context, fixture.session,
request.request_token);
}
bool request_returned = subscribed && app_wait_for_atomic_bool(&request.done, true);
cbm_daemon_application_hold_job_before_start_for_test(false);
bool thread_joined = thread_started && cbm_thread_join(&request_thread) == 0;
int watch_count_after_cancel = fixture.watcher ? cbm_watcher_watch_count(fixture.watcher) : -1;
bool cleaned = app_watch_race_fixture_finish(&fixture);
int worker_starts = atomic_load(&fixture.fake.starts);
int worker_cancels = atomic_load(&fixture.fake.cancels);
int worker_destroys = atomic_load(&fixture.fake.destroys);
free(request.response);
free(tool);
ASSERT_TRUE(fixture_ready);
ASSERT_TRUE(encoded);
ASSERT_TRUE(thread_started);
ASSERT_TRUE(subscribed);
ASSERT_TRUE(request_returned);
ASSERT_TRUE(thread_joined);
ASSERT_EQ(request.status, CBM_DAEMON_RUNTIME_APPLICATION_CANCELLED);
ASSERT_NULL(request.response);
ASSERT_EQ(watch_count_after_cancel, 1);
ASSERT_EQ(worker_starts, 0);
ASSERT_EQ(worker_cancels, 0);
ASSERT_EQ(worker_destroys, 0);
ASSERT_TRUE(cleaned);
PASS();
}
TEST(daemon_application_cancel_drops_watch_before_inflight_request_returns) {
const char *old_cache = getenv("CBM_CACHE_DIR");
bool had_cache = old_cache != NULL;
char *saved_cache = old_cache ? cbm_strdup(old_cache) : NULL;
bool saved_cache_ok = !had_cache || saved_cache;
char root[APP_TEST_PATH_CAP];
char cache[APP_TEST_PATH_CAP];
(void)snprintf(root, sizeof(root), "%s/cbm-app-cancel-watch-root-XXXXXX", cbm_tmpdir());
(void)snprintf(cache, sizeof(cache), "%s/cbm-app-cancel-watch-cache-XXXXXX", cbm_tmpdir());
bool dirs_ok = cbm_mkdtemp(root) != NULL && cbm_mkdtemp(cache) != NULL;
bool env_ok = saved_cache_ok && cbm_setenv("CBM_CACHE_DIR", cache, 1) == 0;
char *project = dirs_ok ? cbm_project_name_from_path(root) : NULL;
char db_path[APP_TEST_PATH_CAP] = {0};
if (project) {
(void)snprintf(db_path, sizeof(db_path), "%s/%s.db", cache, project);
}
bool db_ok = project && app_test_create_empty_file(db_path);
app_fake_worker_context_t fake;
app_fake_worker_context_init(&fake);
cbm_daemon_application_worker_ops_t worker_ops = {
.context = &fake,
.start = app_fake_worker_start,
.poll = app_fake_worker_poll,
.cancel = app_fake_worker_cancel,
.log_path = app_fake_worker_log_path,
.destroy = app_fake_worker_destroy,
};
cbm_store_t *store = cbm_store_open_memory();
cbm_watcher_t *watcher = cbm_watcher_new(store, app_test_index_noop, NULL);
cbm_daemon_application_config_t config = {
.watcher = watcher,
.worker_ops = &worker_ops,
};
cbm_daemon_application_t *application = cbm_daemon_application_new(&config);
cbm_daemon_runtime_application_callbacks_t callbacks =
cbm_daemon_application_runtime_callbacks(application);
cbm_daemon_runtime_application_session_t *session = app_test_open(&callbacks, 39);
uint8_t *context = NULL;
uint32_t context_length = 0;
uint8_t *ping = NULL;
uint32_t ping_length = 0;
char args[APP_TEST_PATH_CAP + 32];
(void)snprintf(args, sizeof(args), "{\"repo_path\":\"%s\"}", root);
uint8_t *tool = NULL;
uint32_t tool_length = 0;
uint8_t *response = NULL;
uint32_t response_length = 0;
bool setup = env_ok && db_ok && store && watcher && application && session &&
app_test_context_request(root, root, &context, &context_length) &&
app_test_text_request(CBM_DAEMON_APPLICATION_REQUEST_MCP,
"{\"jsonrpc\":\"2.0\",\"id\":17,\"method\":\"ping\"}", &ping,
&ping_length) &&
app_test_tool_request("index_repository", args, &tool, &tool_length);
if (setup) {
setup = app_test_request(&callbacks, session, context, context_length, &response,
&response_length) == CBM_DAEMON_RUNTIME_APPLICATION_OK;
free(response);
response = NULL;
response_length = 0;
}
if (setup) {
setup = app_test_request(&callbacks, session, ping, ping_length, &response,
&response_length) == CBM_DAEMON_RUNTIME_APPLICATION_OK;
free(response);
response = NULL;
response_length = 0;
}
bool watch_registered = setup && cbm_watcher_watch_count(watcher) == 1;
app_request_thread_t request = {
.callbacks = callbacks,
.session = session,
.request = tool,
.request_length = tool_length,
.status = CBM_DAEMON_RUNTIME_APPLICATION_HANDLER_ERROR,
};
atomic_init(&request.done, false);
cbm_thread_t request_thread;
bool thread_started = watch_registered &&
cbm_thread_create(&request_thread, 0, app_request_thread, &request) == 0;
bool worker_started = thread_started && app_wait_for_atomic_int(&fake.starts, 1);
if (thread_started) {
/* Runtime close is intentionally delayed until the in-flight request
* returns; cancel is the disconnect ownership boundary. */
callbacks.session_cancel(callbacks.context, session);
}
int watch_count_at_cancel = cbm_watcher_watch_count(watcher);
bool thread_joined = thread_started && cbm_thread_join(&request_thread) == 0;
int watch_count_after_request = cbm_watcher_watch_count(watcher);
bool response_cancelled = request.status == CBM_DAEMON_RUNTIME_APPLICATION_CANCELLED &&
request.response == NULL && request.response_length == 0;
if (session) {
callbacks.session_close(callbacks.context, session);
}
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
cbm_watcher_stop(watcher);
cbm_watcher_free(watcher);
cbm_store_close(store);
free(request.response);
free(context);
free(ping);
free(tool);
free(project);
(void)cbm_unlink(db_path);
(void)cbm_rmdir(root);
(void)cbm_rmdir(cache);
if (saved_cache) {
(void)cbm_setenv("CBM_CACHE_DIR", saved_cache, 1);
} else if (!had_cache) {
(void)cbm_unsetenv("CBM_CACHE_DIR");
}
free(saved_cache);
ASSERT_TRUE(setup);
ASSERT_TRUE(watch_registered);
ASSERT_TRUE(thread_started);
ASSERT_TRUE(worker_started);
ASSERT_EQ(watch_count_at_cancel, 0);
ASSERT_TRUE(thread_joined);
ASSERT_EQ(watch_count_after_request, 0);
ASSERT_TRUE(response_cancelled);
ASSERT_EQ(atomic_load(&fake.cancels), 1);
ASSERT_EQ(atomic_load(&fake.destroys), 1);
ASSERT_TRUE(stopped);
PASS();
}
TEST(daemon_application_stale_watcher_callback_is_rejected_at_job_admission) {
app_watch_race_fixture_t fixture;
bool fixture_ready = app_watch_race_fixture_init(&fixture, 40);
app_watcher_index_thread_t request = {
.application = fixture.application,
.project = fixture.project,
.root = fixture.root,
.pause_before_subscribe = true,
.result = -1,
};
atomic_init(&request.ready, false);
atomic_init(&request.proceed, false);
atomic_init(&request.done, false);
cbm_thread_t thread;
bool thread_started =
fixture_ready && cbm_thread_create(&thread, 0, app_watcher_index_thread, &request) == 0;
bool paused_before_subscribe = thread_started && app_wait_for_atomic_bool(&request.ready, true);
if (thread_started) {
/* Models a watcher callback selected before unwatch but scheduled
* after the last owning session crosses its cancel boundary. */
fixture.callbacks.session_cancel(fixture.callbacks.context, fixture.session);
}
int watches_after_cancel = cbm_watcher_watch_count(fixture.watcher);
atomic_store(&request.proceed, true);
/* On unfixed code the stale callback starts a worker. Let that RED path
* finish instead of hanging the suite; the zero-start assertion remains
* the reproduction oracle. */
atomic_store(&fixture.fake.allow_completion, true);
bool request_done = thread_started && app_wait_for_atomic_bool(&request.done, true);
bool thread_joined = request_done && cbm_thread_join(&thread) == 0;
int worker_starts = atomic_load(&fixture.fake.starts);
size_t active_jobs = cbm_daemon_application_active_jobs(fixture.application);
bool cleaned = app_watch_race_fixture_finish(&fixture);
ASSERT_TRUE(fixture_ready);
ASSERT_TRUE(thread_started);
ASSERT_TRUE(paused_before_subscribe);
ASSERT_EQ(watches_after_cancel, 0);
ASSERT_TRUE(thread_joined);
ASSERT_EQ(request.result, 1);
ASSERT_EQ(worker_starts, 0);
ASSERT_EQ(active_jobs, 0);
ASSERT_TRUE(cleaned);
PASS();
}
TEST(daemon_application_final_cancel_drains_admitted_watcher_job) {
app_watch_race_fixture_t fixture;
bool fixture_ready = app_watch_race_fixture_init(&fixture, 41);
app_watcher_index_thread_t request = {
.application = fixture.application,
.project = fixture.project,
.root = fixture.root,
.pause_before_subscribe = false,
.result = -1,
};
atomic_init(&request.ready, false);
atomic_init(&request.proceed, true);
atomic_init(&request.done, false);
cbm_thread_t thread;
bool thread_started =
fixture_ready && cbm_thread_create(&thread, 0, app_watcher_index_thread, &request) == 0;
bool worker_started = thread_started && app_wait_for_atomic_int(&fixture.fake.starts, 1);
bool job_active =
worker_started && cbm_daemon_application_active_jobs(fixture.application) == 1;
if (thread_started) {
fixture.callbacks.session_cancel(fixture.callbacks.context, fixture.session);
}
int watches_after_cancel = cbm_watcher_watch_count(fixture.watcher);
bool cancelled_by_final_session =
thread_started && app_wait_for_atomic_int(&fixture.fake.cancels, 1);
if (!cancelled_by_final_session) {
/* RED cleanup: prior behavior leaves this watcher-owned worker alive
* while another daemon client could keep the process running. */
atomic_store(&fixture.fake.allow_completion, true);
}
bool request_done = thread_started && app_wait_for_atomic_bool(&request.done, true);
bool thread_joined = request_done && cbm_thread_join(&thread) == 0;
size_t active_jobs_after_join = cbm_daemon_application_active_jobs(fixture.application);
int worker_cancels = atomic_load(&fixture.fake.cancels);
int worker_destroys = atomic_load(&fixture.fake.destroys);
bool cleaned = app_watch_race_fixture_finish(&fixture);
ASSERT_TRUE(fixture_ready);
ASSERT_TRUE(thread_started);
ASSERT_TRUE(worker_started);
ASSERT_TRUE(job_active);
ASSERT_EQ(watches_after_cancel, 0);
ASSERT_TRUE(thread_joined);
ASSERT_EQ(request.result, 1);
ASSERT_EQ(active_jobs_after_join, 0);
ASSERT_TRUE(cancelled_by_final_session);
ASSERT_EQ(worker_cancels, 1);
ASSERT_EQ(worker_destroys, 1);
ASSERT_TRUE(cleaned);
PASS();
}
TEST(daemon_application_watcher_job_follows_exact_live_watch_owners) {
app_watch_race_fixture_t fixture;
bool fixture_ready = app_watch_race_fixture_init(&fixture, 42);
cbm_daemon_runtime_application_session_t *second =
fixture_ready ? app_test_open(&fixture.callbacks, 43) : NULL;
/* This unrelated live session keeps the daemon generation alive after
* both matching watch owners disconnect. The watcher job must still be
* cancelled when its own final owner disappears. */
cbm_daemon_runtime_application_session_t *unrelated =
fixture_ready ? app_test_open(&fixture.callbacks, 44) : NULL;
uint8_t *context = NULL;
uint32_t context_length = 0;
uint8_t *ping = NULL;
uint32_t ping_length = 0;
uint8_t *response = NULL;
uint32_t response_length = 0;
bool encoded =
fixture_ready && second && unrelated &&
app_test_context_request(fixture.root, fixture.root, &context, &context_length) &&
app_test_text_request(CBM_DAEMON_APPLICATION_REQUEST_MCP,
"{\"jsonrpc\":\"2.0\",\"id\":20,\"method\":\"ping\"}", &ping,
&ping_length);
bool second_watching =
encoded && app_test_request(&fixture.callbacks, second, context, context_length, &response,
&response_length) == CBM_DAEMON_RUNTIME_APPLICATION_OK;
free(response);
response = NULL;
response_length = 0;
second_watching = second_watching &&
app_test_request(&fixture.callbacks, second, ping, ping_length, &response,
&response_length) == CBM_DAEMON_RUNTIME_APPLICATION_OK &&
cbm_watcher_watch_count(fixture.watcher) == 1;
free(response);
response = NULL;
app_watcher_index_thread_t request = {
.application = fixture.application,
.project = fixture.project,
.root = fixture.root,
.pause_before_subscribe = false,
.result = -1,
};
atomic_init(&request.ready, false);
atomic_init(&request.proceed, true);
atomic_init(&request.done, false);
cbm_thread_t thread;
bool thread_started =
second_watching && cbm_thread_create(&thread, 0, app_watcher_index_thread, &request) == 0;
bool worker_started = thread_started && app_wait_for_atomic_int(&fixture.fake.starts, 1);
size_t initial_subscribers =
worker_started
? cbm_daemon_application_job_subscribers(fixture.application, fixture.project)
: 0;
if (fixture.session) {
fixture.callbacks.session_close(fixture.callbacks.context, fixture.session);
fixture.session = NULL;
}
int watches_after_first = cbm_watcher_watch_count(fixture.watcher);
size_t subscribers_after_first =
cbm_daemon_application_job_subscribers(fixture.application, fixture.project);
int cancels_after_first = atomic_load(&fixture.fake.cancels);
if (second) {
fixture.callbacks.session_close(fixture.callbacks.context, second);
second = NULL;
}
int watches_after_second = cbm_watcher_watch_count(fixture.watcher);
size_t subscribers_after_second =
cbm_daemon_application_job_subscribers(fixture.application, fixture.project);
bool cancelled_without_final_session =
thread_started && app_wait_for_atomic_int(&fixture.fake.cancels, 1);
if (!cancelled_without_final_session) {
/* RED cleanup: the old anonymous watcher subscription keeps the job
* alive here, so let it finish without leaking the test thread. */
atomic_store(&fixture.fake.allow_completion, true);
}
bool request_done = thread_started && app_wait_for_atomic_bool(&request.done, true);
bool thread_joined = request_done && cbm_thread_join(&thread) == 0;
if (unrelated) {
fixture.callbacks.session_close(fixture.callbacks.context, unrelated);
unrelated = NULL;
}
bool cleaned = app_watch_race_fixture_finish(&fixture);
free(context);
free(ping);
free(response);
ASSERT_TRUE(fixture_ready);
ASSERT_TRUE(second_watching);
ASSERT_TRUE(thread_started);
ASSERT_TRUE(worker_started);
ASSERT_EQ(initial_subscribers, 2);
ASSERT_EQ(watches_after_first, 1);
ASSERT_EQ(subscribers_after_first, 1);
ASSERT_EQ(cancels_after_first, 0);
ASSERT_EQ(watches_after_second, 0);
ASSERT_EQ(subscribers_after_second, 0);
ASSERT_TRUE(cancelled_without_final_session);
ASSERT_TRUE(thread_joined);
ASSERT_EQ(request.result, 1);
ASSERT_EQ(atomic_load(&fixture.fake.destroys), 1);
ASSERT_TRUE(cleaned);
PASS();
}
TEST(daemon_application_late_watcher_session_owns_active_watcher_job) {
app_watch_race_fixture_t fixture;
bool fixture_ready = app_watch_race_fixture_init(&fixture, 45);
cbm_daemon_runtime_application_session_t *unrelated =
fixture_ready ? app_test_open(&fixture.callbacks, 46) : NULL;
app_watcher_index_thread_t request = {
.application = fixture.application,
.project = fixture.project,
.root = fixture.root,
.pause_before_subscribe = false,
.result = -1,
};
atomic_init(&request.ready, false);
atomic_init(&request.proceed, true);
atomic_init(&request.done, false);
cbm_thread_t thread;
bool thread_started = fixture_ready && unrelated &&
cbm_thread_create(&thread, 0, app_watcher_index_thread, &request) == 0;
bool worker_started = thread_started && app_wait_for_atomic_int(&fixture.fake.starts, 1);
size_t initial_subscribers =
worker_started
? cbm_daemon_application_job_subscribers(fixture.application, fixture.project)
: 0;
cbm_daemon_runtime_application_session_t *late =
worker_started ? app_test_open(&fixture.callbacks, 47) : NULL;
uint8_t *context = NULL;
uint32_t context_length = 0;
uint8_t *ping = NULL;
uint32_t ping_length = 0;
uint8_t *response = NULL;
uint32_t response_length = 0;
bool encoded =
late && app_test_context_request(fixture.root, fixture.root, &context, &context_length) &&
app_test_text_request(CBM_DAEMON_APPLICATION_REQUEST_MCP,
"{\"jsonrpc\":\"2.0\",\"id\":47,\"method\":\"ping\"}", &ping,
&ping_length);
bool late_watching =
encoded && app_test_request(&fixture.callbacks, late, context, context_length, &response,
&response_length) == CBM_DAEMON_RUNTIME_APPLICATION_OK;
free(response);
response = NULL;
response_length = 0;
late_watching = late_watching &&
app_test_request(&fixture.callbacks, late, ping, ping_length, &response,
&response_length) == CBM_DAEMON_RUNTIME_APPLICATION_OK &&
cbm_watcher_watch_count(fixture.watcher) == 1;
free(response);
response = NULL;
size_t subscribers_with_late =
late_watching ? cbm_daemon_application_job_subscribers(fixture.application, fixture.project)
: 0;
if (fixture.session) {
fixture.callbacks.session_close(fixture.callbacks.context, fixture.session);
fixture.session = NULL;
}
size_t subscribers_after_initial =
cbm_daemon_application_job_subscribers(fixture.application, fixture.project);
int cancels_after_initial = atomic_load(&fixture.fake.cancels);
if (late) {
fixture.callbacks.session_close(fixture.callbacks.context, late);
late = NULL;
}
bool cancelled_after_late = app_wait_for_atomic_int(&fixture.fake.cancels, 1);
if (!cancelled_after_late) {
atomic_store(&fixture.fake.allow_completion, true);
}
bool request_done = app_wait_for_atomic_bool(&request.done, true);
bool thread_joined = request_done && cbm_thread_join(&thread) == 0;
if (unrelated) {
fixture.callbacks.session_close(fixture.callbacks.context, unrelated);
unrelated = NULL;
}
bool cleaned = app_watch_race_fixture_finish(&fixture);
free(context);
free(ping);
free(response);
ASSERT_TRUE(fixture_ready);
ASSERT_TRUE(thread_started);
ASSERT_TRUE(worker_started);
ASSERT_EQ(initial_subscribers, 1);
ASSERT_TRUE(late_watching);
ASSERT_EQ(subscribers_with_late, 2);
ASSERT_EQ(subscribers_after_initial, 1);
ASSERT_EQ(cancels_after_initial, 0);
ASSERT_TRUE(cancelled_after_late);
ASSERT_TRUE(thread_joined);
ASSERT_EQ(request.result, 1);
ASSERT_TRUE(cleaned);
PASS();
}
/* A staged index snapshots the current DB before it publishes a replacement.
* Any ADR write accepted during that lifetime can therefore be silently lost
* (and can make replacement fail on Windows). The daemon job must keep the
* project reserved against mutations until the physical worker is terminal. */
TEST(daemon_application_serializes_adr_mutation_with_index_job) {
const char *old_cache = getenv("CBM_CACHE_DIR");
bool had_cache = old_cache != NULL;
char *saved_cache = old_cache ? cbm_strdup(old_cache) : NULL;
app_fake_worker_context_t fake;
app_fake_worker_context_init(&fake);
cbm_daemon_application_worker_ops_t worker_ops = {
.context = &fake,
.start = app_fake_worker_start,
.poll = app_fake_worker_poll,
.cancel = app_fake_worker_cancel,
.log_path = app_fake_worker_log_path,
.destroy = app_fake_worker_destroy,
};
cbm_daemon_application_config_t config = {.worker_ops = &worker_ops};
char root[APP_TEST_PATH_CAP];
char cache[APP_TEST_PATH_CAP];
snprintf(root, sizeof(root), "%s/cbm-app-mutation-root-XXXXXX", cbm_tmpdir());
snprintf(cache, sizeof(cache), "%s/cbm-app-mutation-cache-XXXXXX", cbm_tmpdir());
bool dirs_ok = cbm_mkdtemp(root) != NULL && cbm_mkdtemp(cache) != NULL;
bool env_ok =
dirs_ok && (!had_cache || saved_cache) && cbm_setenv("CBM_CACHE_DIR", cache, 1) == 0;
char *project = env_ok ? cbm_project_name_from_path(root) : NULL;
char db_path[APP_TEST_PATH_CAP] = {0};
cbm_store_t *seed = NULL;
bool seeded = false;
if (project) {
snprintf(db_path, sizeof(db_path), "%s/%s.db", cache, project);
seed = cbm_store_open_path(db_path);
seeded = seed && cbm_store_upsert_project(seed, project, root) == CBM_STORE_OK;
cbm_store_close(seed);
}
cbm_daemon_application_t *application = seeded ? cbm_daemon_application_new(&config) : NULL;
cbm_daemon_runtime_application_callbacks_t callbacks =
cbm_daemon_application_runtime_callbacks(application);
cbm_daemon_runtime_application_session_t *index_session =
application ? app_test_open(&callbacks, 33) : NULL;
cbm_daemon_runtime_application_session_t *adr_session =
application ? app_test_open(&callbacks, 34) : NULL;
uint8_t *context = NULL;
uint32_t context_length = 0;
char index_args[APP_TEST_PATH_CAP + 32];
snprintf(index_args, sizeof(index_args), "{\"repo_path\":\"%s\"}", root);
uint8_t *index_tool = NULL;
uint32_t index_tool_length = 0;
char adr_args[APP_TEST_PATH_CAP + 128];
snprintf(adr_args, sizeof(adr_args),
"{\"project\":\"%s\",\"mode\":\"update\","
"\"content\":\"serialized ADR\"}",
project ? project : "");
uint8_t *adr_tool = NULL;
uint32_t adr_tool_length = 0;
bool setup =
application && index_session && adr_session &&
app_test_context_request(root, root, &context, &context_length) &&
app_test_tool_request("index_repository", index_args, &index_tool, &index_tool_length) &&
app_test_tool_request("manage_adr", adr_args, &adr_tool, &adr_tool_length);
uint8_t *response = NULL;
uint32_t response_length = 0;
if (setup) {
setup = app_test_request(&callbacks, index_session, context, context_length, &response,
&response_length) == CBM_DAEMON_RUNTIME_APPLICATION_OK;
free(response);
response = NULL;
}
if (setup) {
setup = app_test_request(&callbacks, adr_session, context, context_length, &response,
&response_length) == CBM_DAEMON_RUNTIME_APPLICATION_OK;
free(response);
response = NULL;
}
app_request_thread_t index_request = {
.callbacks = callbacks,
.session = index_session,
.request = index_tool,
.request_length = index_tool_length,
};
app_request_thread_t adr_request = {
.callbacks = callbacks,
.session = adr_session,
.request = adr_tool,
.request_length = adr_tool_length,
};
cbm_thread_t index_thread;
cbm_thread_t adr_thread;
bool index_started =
setup && cbm_thread_create(&index_thread, 0, app_request_thread, &index_request) == 0;
bool index_active = index_started && project &&
app_wait_for_subscribers(application, project, 1) &&
app_wait_for_atomic_int(&fake.starts, 1);
bool adr_started =
index_active && cbm_thread_create(&adr_thread, 0, app_request_thread, &adr_request) == 0;
uint64_t early_deadline = cbm_now_ms() + 100;
while (adr_started && !atomic_load(&adr_request.done) && cbm_now_ms() < early_deadline) {
cbm_usleep(1000);
}
bool adr_completed_while_index_active = adr_started && atomic_load(&adr_request.done);
atomic_store(&fake.allow_completion, true);
if (index_started) {
(void)cbm_thread_join(&index_thread);
}
if (adr_started) {
(void)cbm_thread_join(&adr_thread);
}
cbm_store_t *check = cbm_store_open_path_query(db_path);
cbm_adr_t adr = {0};
bool adr_persisted = check && cbm_store_adr_get(check, project, &adr) == CBM_STORE_OK &&
adr.content && strcmp(adr.content, "serialized ADR") == 0;
if (adr.content) {
cbm_store_adr_free(&adr);
}
cbm_store_close(check);
if (index_session) {
callbacks.session_close(callbacks.context, index_session);
}
if (adr_session) {
callbacks.session_close(callbacks.context, adr_session);
}
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
bool adr_response_updated =
adr_request.response && strstr((char *)adr_request.response, "updated") != NULL;
free(index_request.response);
free(adr_request.response);
free(context);
free(index_tool);
free(adr_tool);
(void)cbm_unlink(db_path);
char sidecar[APP_TEST_PATH_CAP];
snprintf(sidecar, sizeof(sidecar), "%s-wal", db_path);
(void)cbm_unlink(sidecar);
snprintf(sidecar, sizeof(sidecar), "%s-shm", db_path);
(void)cbm_unlink(sidecar);
free(project);
(void)cbm_rmdir(root);
(void)cbm_rmdir(cache);
if (saved_cache) {
(void)cbm_setenv("CBM_CACHE_DIR", saved_cache, 1);
} else if (!had_cache) {
(void)cbm_unsetenv("CBM_CACHE_DIR");
}
free(saved_cache);
ASSERT_TRUE(env_ok);
ASSERT_TRUE(seeded);
ASSERT_TRUE(setup);
ASSERT_TRUE(index_active);
ASSERT_TRUE(adr_started);
ASSERT_FALSE(adr_completed_while_index_active);
ASSERT_EQ(index_request.status, CBM_DAEMON_RUNTIME_APPLICATION_OK);
ASSERT_EQ(adr_request.status, CBM_DAEMON_RUNTIME_APPLICATION_OK);
ASSERT_TRUE(adr_response_updated);
ASSERT_TRUE(adr_persisted);
ASSERT_TRUE(stopped);
PASS();
}
TEST(daemon_application_reserved_mutation_delays_worker_start) {
app_fake_worker_context_t fake;
app_fake_worker_context_init(&fake);
cbm_daemon_application_worker_ops_t worker_ops = {
.context = &fake,
.start = app_fake_worker_start,
.poll = app_fake_worker_poll,
.cancel = app_fake_worker_cancel,
.log_path = app_fake_worker_log_path,
.destroy = app_fake_worker_destroy,
};
cbm_daemon_application_config_t config = {.worker_ops = &worker_ops};
cbm_daemon_application_t *application = cbm_daemon_application_new(&config);
char root[APP_TEST_PATH_CAP];
snprintf(root, sizeof(root), "%s/cbm-app-mutation-first-XXXXXX", cbm_tmpdir());
bool root_ok = cbm_mkdtemp(root) != NULL;
const char *project = "mutation-first";
bool lease_held =
application && cbm_daemon_application_project_mutation_try_begin(application, project);
app_index_thread_t request = {
.application = application,
.project = project,
.root = root,
.result = -1,
};
cbm_thread_t thread;
bool thread_started =
root_ok && lease_held && cbm_thread_create(&thread, 0, app_index_thread, &request) == 0;
bool reserved = thread_started && app_wait_for_subscribers(application, project, 1);
uint64_t early_deadline = cbm_now_ms() + 100;
while (atomic_load(&fake.starts) == 0 && cbm_now_ms() < early_deadline) {
cbm_usleep(1000);
}
bool worker_started_while_mutating = atomic_load(&fake.starts) != 0;
if (lease_held) {
cbm_daemon_application_project_mutation_end(application, project);
}
bool worker_started = thread_started && app_wait_for_atomic_int(&fake.starts, 1);
atomic_store(&fake.allow_completion, true);
if (thread_started) {
(void)cbm_thread_join(&thread);
}
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
(void)cbm_rmdir(root);
ASSERT_TRUE(root_ok);
ASSERT_TRUE(lease_held);
ASSERT_TRUE(thread_started);
ASSERT_TRUE(reserved);
ASSERT_FALSE(worker_started_while_mutating);
ASSERT_TRUE(worker_started);
ASSERT_EQ(request.result, 0);
ASSERT_TRUE(stopped);
PASS();
}
TEST(daemon_application_mutation_does_not_block_other_project) {
app_fake_worker_context_t fake;
app_fake_worker_context_init(&fake);
cbm_daemon_application_worker_ops_t worker_ops = {
.context = &fake,
.start = app_fake_worker_start,
.poll = app_fake_worker_poll,
.cancel = app_fake_worker_cancel,
.log_path = app_fake_worker_log_path,
.destroy = app_fake_worker_destroy,
};
cbm_daemon_application_config_t config = {.worker_ops = &worker_ops};
cbm_daemon_application_t *application = cbm_daemon_application_new(&config);
char root[APP_TEST_PATH_CAP];
snprintf(root, sizeof(root), "%s/cbm-app-mutation-other-XXXXXX", cbm_tmpdir());
bool root_ok = cbm_mkdtemp(root) != NULL;
bool lease_held =
application && cbm_daemon_application_project_mutation_try_begin(application, "project-a");
app_index_thread_t request = {
.application = application,
.project = "project-b",
.root = root,
.result = -1,
};
cbm_thread_t thread;
bool thread_started =
root_ok && lease_held && cbm_thread_create(&thread, 0, app_index_thread, &request) == 0;
bool worker_started = thread_started && app_wait_for_atomic_int(&fake.starts, 1);
atomic_store(&fake.allow_completion, true);
if (thread_started) {
(void)cbm_thread_join(&thread);
}
if (lease_held) {
cbm_daemon_application_project_mutation_end(application, "project-a");
}
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
(void)cbm_rmdir(root);
ASSERT_TRUE(root_ok);
ASSERT_TRUE(lease_held);
ASSERT_TRUE(thread_started);
ASSERT_TRUE(worker_started);
ASSERT_EQ(request.result, 0);
ASSERT_TRUE(stopped);
PASS();
}
TEST(daemon_application_mutation_still_owns_os_project_lock) {
app_project_lock_fixture_t locks;
bool locks_ready = app_project_lock_fixture_init(&locks);
cbm_daemon_application_config_t config = {
.project_locks = locks_ready ? locks.daemon_locks : NULL,
};
cbm_daemon_application_t *application =
locks_ready ? cbm_daemon_application_new(&config) : NULL;
const char *project = "daemon-mutation-native-lock";
bool mutation_started =
application && cbm_daemon_application_project_mutation_try_begin(application, project);
cbm_project_lock_lease_t *active_probe = NULL;
cbm_private_file_lock_status_t active_status =
mutation_started ? cbm_project_lock_try_acquire(locks.probe_locks, project, &active_probe)
: CBM_PRIVATE_FILE_LOCK_IO;
bool mutation_holds_lock = active_status == CBM_PRIVATE_FILE_LOCK_BUSY && !active_probe;
bool active_probe_released = app_project_lock_release(&active_probe);
if (mutation_started) {
cbm_daemon_application_project_mutation_end(application, project);
}
cbm_project_lock_lease_t *terminal_probe = NULL;
cbm_private_file_lock_status_t terminal_status =
mutation_started ? cbm_project_lock_try_acquire(locks.probe_locks, project, &terminal_probe)
: CBM_PRIVATE_FILE_LOCK_IO;
bool mutation_released_lock = terminal_status == CBM_PRIVATE_FILE_LOCK_OK && terminal_probe;
bool terminal_probe_released = app_project_lock_release(&terminal_probe);
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
bool locks_clean = app_project_lock_fixture_cleanup(&locks);
ASSERT_TRUE(locks_ready);
ASSERT_NOT_NULL(application);
ASSERT_TRUE(mutation_started);
ASSERT_TRUE(mutation_holds_lock);
ASSERT_TRUE(active_probe_released);
ASSERT_TRUE(mutation_released_lock);
ASSERT_TRUE(terminal_probe_released);
ASSERT_TRUE(stopped);
ASSERT_TRUE(locks_clean);
PASS();
}
/* RED on daemon-held project leases: the injected physical-worker boundary
* attempts the exact same native lock. A parent lease makes that attempt fail;
* the correct ownership has the worker acquire it and hold it until destroy. */
TEST(daemon_application_worker_boundary_is_sole_os_project_lock_owner) {
app_project_lock_fixture_t locks;
bool locks_ready = app_project_lock_fixture_init(&locks);
app_fake_worker_context_t fake;
app_fake_worker_context_init(&fake);
fake.project_locks = locks_ready ? locks.worker_locks : NULL;
fake.project_lock_busy_is_error = true;
cbm_daemon_application_worker_ops_t worker_ops = {
.context = &fake,
.start = app_fake_worker_start,
.poll = app_fake_worker_poll,
.cancel = app_fake_worker_cancel,
.log_path = app_fake_worker_log_path,
.destroy = app_fake_worker_destroy,
};
cbm_daemon_application_config_t config = {
.worker_ops = &worker_ops,
.project_locks = locks_ready ? locks.daemon_locks : NULL,
};
cbm_daemon_application_t *application =
locks_ready ? cbm_daemon_application_new(&config) : NULL;
char root[APP_TEST_PATH_CAP];
(void)snprintf(root, sizeof(root), "%s/cbm-app-worker-lock-owner-XXXXXX", cbm_tmpdir());
bool root_ok = cbm_mkdtemp(root) != NULL;
const char *project = "worker-lock-owner";
app_index_thread_t request = {
.application = application,
.project = project,
.root = root,
.result = -1,
};
cbm_thread_t thread;
bool thread_started =
root_ok && application && cbm_thread_create(&thread, 0, app_index_thread, &request) == 0;
bool boundary_started = thread_started && app_wait_for_atomic_int(&fake.starts, 1);
bool worker_acquired =
boundary_started && app_wait_for_atomic_int(&fake.project_lock_acquisitions, 1);
cbm_project_lock_lease_t *active_probe = NULL;
cbm_private_file_lock_status_t active_probe_status =
worker_acquired ? cbm_project_lock_try_acquire(locks.probe_locks, project, &active_probe)
: CBM_PRIVATE_FILE_LOCK_IO;
bool worker_holds_lock = active_probe_status == CBM_PRIVATE_FILE_LOCK_BUSY && !active_probe;
bool active_probe_released = app_project_lock_release(&active_probe);
atomic_store(&fake.allow_completion, true);
if (thread_started) {
(void)cbm_thread_join(&thread);
}
cbm_project_lock_lease_t *terminal_probe = NULL;
cbm_private_file_lock_status_t terminal_probe_status =
thread_started ? cbm_project_lock_try_acquire(locks.probe_locks, project, &terminal_probe)
: CBM_PRIVATE_FILE_LOCK_IO;
bool worker_released_lock = terminal_probe_status == CBM_PRIVATE_FILE_LOCK_OK && terminal_probe;
bool terminal_probe_released = app_project_lock_release(&terminal_probe);
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
bool locks_clean = app_project_lock_fixture_cleanup(&locks);
(void)th_rmtree(root);
ASSERT_TRUE(locks_ready);
ASSERT_TRUE(root_ok);
ASSERT_TRUE(thread_started);
ASSERT_TRUE(boundary_started);
ASSERT_TRUE(worker_acquired);
ASSERT_EQ(atomic_load(&fake.project_lock_attempts), 1);
ASSERT_TRUE(worker_holds_lock);
ASSERT_TRUE(active_probe_released);
ASSERT_EQ(request.result, 0);
ASSERT_TRUE(worker_released_lock);
ASSERT_TRUE(terminal_probe_released);
ASSERT_TRUE(stopped);
ASSERT_TRUE(locks_clean);
ASSERT_EQ(atomic_load(&fake.starts), 1);
ASSERT_EQ(atomic_load(&fake.destroys), 1);
PASS();
}
TEST(daemon_application_worker_lock_serializes_external_mutation) {
app_project_lock_fixture_t locks;
bool locks_ready = app_project_lock_fixture_init(&locks);
app_fake_worker_context_t fake;
app_fake_worker_context_init(&fake);
fake.project_locks = locks_ready ? locks.worker_locks : NULL;
cbm_daemon_application_worker_ops_t worker_ops = {
.context = &fake,
.start = app_fake_worker_start,
.poll = app_fake_worker_poll,
.cancel = app_fake_worker_cancel,
.log_path = app_fake_worker_log_path,
.destroy = app_fake_worker_destroy,
};
cbm_daemon_application_config_t config = {
.worker_ops = &worker_ops,
.project_locks = locks_ready ? locks.daemon_locks : NULL,
};
cbm_daemon_application_t *application =
locks_ready ? cbm_daemon_application_new(&config) : NULL;
char blocked_root[APP_TEST_PATH_CAP];
char other_root[APP_TEST_PATH_CAP];
(void)snprintf(blocked_root, sizeof(blocked_root), "%s/cbm-app-external-blocked-XXXXXX",
cbm_tmpdir());
(void)snprintf(other_root, sizeof(other_root), "%s/cbm-app-external-other-XXXXXX",
cbm_tmpdir());
bool blocked_root_ok = cbm_mkdtemp(blocked_root) != NULL;
bool other_root_ok = cbm_mkdtemp(other_root) != NULL;
const char *blocked_project = "external-blocked";
const char *other_project = "external-other";
cbm_project_lock_lease_t *external_lease = NULL;
bool external_lease_held =
application && blocked_root_ok && other_root_ok &&
cbm_project_lock_acquire(locks.external_locks, blocked_project, UINT64_MAX, NULL,
&external_lease) == CBM_PRIVATE_FILE_LOCK_OK &&
external_lease;
app_index_thread_t blocked_request = {
.application = application,
.project = blocked_project,
.root = blocked_root,
.result = -1,
};
cbm_thread_t blocked_thread;
bool blocked_thread_started =
external_lease_held &&
cbm_thread_create(&blocked_thread, 0, app_index_thread, &blocked_request) == 0;
bool blocked_boundary_started =
blocked_thread_started && app_wait_for_atomic_int(&fake.starts, 1);
bool blocked_boundary_attempted =
blocked_boundary_started &&
app_wait_for_atomic_int_at_least(&fake.project_lock_attempts, 1);
bool blocked_waiting =
blocked_boundary_attempted && atomic_load(&fake.project_lock_acquisitions) == 0;
app_index_thread_t other_request = {
.application = application,
.project = other_project,
.root = other_root,
.result = -1,
};
cbm_thread_t other_thread;
bool other_thread_started =
blocked_waiting &&
cbm_thread_create(&other_thread, 0, app_index_thread, &other_request) == 0;
bool both_boundaries_started = other_thread_started && app_wait_for_atomic_int(&fake.starts, 2);
bool other_worker_acquired =
both_boundaries_started && app_wait_for_atomic_int(&fake.project_lock_acquisitions, 1);
cbm_project_lock_lease_t *other_probe = NULL;
cbm_private_file_lock_status_t active_other_status =
other_worker_acquired
? cbm_project_lock_try_acquire(locks.probe_locks, other_project, &other_probe)
: CBM_PRIVATE_FILE_LOCK_IO;
bool other_lock_held_while_worker_active =
active_other_status == CBM_PRIVATE_FILE_LOCK_BUSY && !other_probe;
bool active_other_probe_released = app_project_lock_release(&other_probe);
atomic_store(&fake.allow_completion, true);
if (other_thread_started && !other_worker_acquired && application) {
(void)cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
}
if (other_thread_started) {
(void)cbm_thread_join(&other_thread);
}
cbm_project_lock_lease_t *terminal_other_probe = NULL;
cbm_private_file_lock_status_t terminal_other_status =
other_worker_acquired ? app_project_lock_try_acquire_until_released(
locks.probe_locks, other_project, &terminal_other_probe)
: CBM_PRIVATE_FILE_LOCK_IO;
bool other_lock_released_at_terminal =
terminal_other_status == CBM_PRIVATE_FILE_LOCK_OK && terminal_other_probe;
bool terminal_other_probe_released = app_project_lock_release(&terminal_other_probe);
bool blocked_still_waiting = atomic_load(&fake.project_lock_acquisitions) == 1;
bool external_lease_released = app_project_lock_release(&external_lease);
bool blocked_worker_acquired =
blocked_thread_started && app_wait_for_atomic_int(&fake.project_lock_acquisitions, 2);
if (blocked_thread_started && !blocked_worker_acquired && application) {
(void)cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
}
if (blocked_thread_started) {
(void)cbm_thread_join(&blocked_thread);
}
cbm_project_lock_lease_t *blocked_probe = NULL;
cbm_private_file_lock_status_t terminal_blocked_status =
blocked_worker_acquired ? app_project_lock_try_acquire_until_released(
locks.probe_locks, blocked_project, &blocked_probe)
: CBM_PRIVATE_FILE_LOCK_IO;
bool blocked_lock_released_at_terminal =
terminal_blocked_status == CBM_PRIVATE_FILE_LOCK_OK && blocked_probe;
bool blocked_probe_released = app_project_lock_release(&blocked_probe);
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
bool locks_clean = app_project_lock_fixture_cleanup(&locks);
(void)th_rmtree(blocked_root);
(void)th_rmtree(other_root);
ASSERT_TRUE(locks_ready);
ASSERT_TRUE(blocked_root_ok);
ASSERT_TRUE(other_root_ok);
ASSERT_TRUE(external_lease_held);
ASSERT_TRUE(blocked_thread_started);
ASSERT_TRUE(blocked_boundary_started);
ASSERT_TRUE(blocked_boundary_attempted);
ASSERT_TRUE(blocked_waiting);
ASSERT_TRUE(other_thread_started);
ASSERT_TRUE(both_boundaries_started);
ASSERT_TRUE(other_worker_acquired);
ASSERT_TRUE(other_lock_held_while_worker_active);
ASSERT_TRUE(active_other_probe_released);
ASSERT_EQ(other_request.result, 0);
ASSERT_TRUE(other_lock_released_at_terminal);
ASSERT_TRUE(terminal_other_probe_released);
ASSERT_TRUE(blocked_still_waiting);
ASSERT_TRUE(external_lease_released);
ASSERT_TRUE(blocked_worker_acquired);
ASSERT_EQ(blocked_request.result, 0);
ASSERT_TRUE(blocked_lock_released_at_terminal);
ASSERT_TRUE(blocked_probe_released);
ASSERT_TRUE(stopped);
ASSERT_TRUE(locks_clean);
ASSERT_EQ(atomic_load(&fake.starts), 2);
ASSERT_EQ(atomic_load(&fake.project_lock_acquisitions), 2);
ASSERT_EQ(atomic_load(&fake.destroys), 2);
PASS();
}
TEST(daemon_application_final_unsubscribe_cancels_external_lock_wait) {
app_project_lock_fixture_t locks;
bool locks_ready = app_project_lock_fixture_init(&locks);
app_fake_worker_context_t fake;
app_fake_worker_context_init(&fake);
fake.project_locks = locks_ready ? locks.worker_locks : NULL;
cbm_daemon_application_worker_ops_t worker_ops = {
.context = &fake,
.start = app_fake_worker_start,
.poll = app_fake_worker_poll,
.cancel = app_fake_worker_cancel,
.log_path = app_fake_worker_log_path,
.destroy = app_fake_worker_destroy,
};
cbm_daemon_application_config_t config = {
.worker_ops = &worker_ops,
.project_locks = locks_ready ? locks.daemon_locks : NULL,
};
cbm_daemon_application_t *application =
locks_ready ? cbm_daemon_application_new(&config) : NULL;
cbm_daemon_runtime_application_callbacks_t callbacks =
cbm_daemon_application_runtime_callbacks(application);
cbm_daemon_runtime_application_session_t *session =
application ? app_test_open(&callbacks, 37) : NULL;
char root[APP_TEST_PATH_CAP];
(void)snprintf(root, sizeof(root), "%s/cbm-app-external-cancel-XXXXXX", cbm_tmpdir());
bool root_ok = cbm_mkdtemp(root) != NULL;
char *project = root_ok ? cbm_project_name_from_path(root) : NULL;
cbm_project_lock_lease_t *external_lease = NULL;
bool external_lease_held =
session && project &&
cbm_project_lock_acquire(locks.external_locks, project, UINT64_MAX, NULL,
&external_lease) == CBM_PRIVATE_FILE_LOCK_OK &&
external_lease;
uint8_t *context = NULL;
uint32_t context_length = 0;
char args[APP_TEST_PATH_CAP + 32];
(void)snprintf(args, sizeof(args), "{\"repo_path\":\"%s\"}", root);
uint8_t *tool = NULL;
uint32_t tool_length = 0;
bool setup = external_lease_held &&
app_test_context_request(root, root, &context, &context_length) &&
app_test_tool_request("index_repository", args, &tool, &tool_length);
uint8_t *empty = NULL;
uint32_t empty_length = 0;
if (setup) {
setup = app_test_request(&callbacks, session, context, context_length, &empty,
&empty_length) == CBM_DAEMON_RUNTIME_APPLICATION_OK;
}
free(empty);
app_request_thread_t request = {
.callbacks = callbacks,
.session = session,
.request = tool,
.request_length = tool_length,
};
atomic_init(&request.done, false);
cbm_thread_t request_thread;
bool request_started =
setup && cbm_thread_create(&request_thread, 0, app_request_thread, &request) == 0;
bool subscribed = request_started && app_wait_for_subscribers(application, project, 1);
bool worker_boundary_started = subscribed && app_wait_for_atomic_int(&fake.starts, 1);
bool worker_lock_attempted =
worker_boundary_started && app_wait_for_atomic_int_at_least(&fake.project_lock_attempts, 1);
bool worker_waiting =
worker_lock_attempted && atomic_load(&fake.project_lock_acquisitions) == 0;
if (subscribed) {
callbacks.session_cancel(callbacks.context, session);
}
bool final_unsubscribed = subscribed && app_wait_for_subscribers(application, project, 0);
uint64_t request_deadline = cbm_now_ms() + APP_TEST_TIMEOUT_MS;
while (request_started && !atomic_load(&request.done) && cbm_now_ms() < request_deadline) {
cbm_usleep(1000);
}
bool request_finished = request_started && atomic_load(&request.done);
bool job_terminal = final_unsubscribed && app_wait_for_active_jobs(application, 0);
if (request_started) {
(void)cbm_thread_join(&request_thread);
}
bool cancellation_reported = request.status == CBM_DAEMON_RUNTIME_APPLICATION_CANCELLED &&
request.response == NULL && request.response_length == 0;
bool worker_never_acquired = atomic_load(&fake.project_lock_acquisitions) == 0;
bool external_lease_released = app_project_lock_release(&external_lease);
cbm_project_lock_lease_t *leak_probe = NULL;
bool can_probe_for_leak = job_terminal && external_lease_held && external_lease_released &&
locks.probe_locks && project;
cbm_private_file_lock_status_t leak_probe_status =
can_probe_for_leak ? cbm_project_lock_try_acquire(locks.probe_locks, project, &leak_probe)
: CBM_PRIVATE_FILE_LOCK_IO;
bool no_daemon_lease_leaked = leak_probe_status == CBM_PRIVATE_FILE_LOCK_OK && leak_probe;
bool leak_probe_released = app_project_lock_release(&leak_probe);
if (session) {
callbacks.session_close(callbacks.context, session);
}
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
bool locks_clean = app_project_lock_fixture_cleanup(&locks);
free(request.response);
free(context);
free(tool);
free(project);
(void)th_rmtree(root);
ASSERT_TRUE(locks_ready);
ASSERT_TRUE(root_ok);
ASSERT_TRUE(external_lease_held);
ASSERT_TRUE(setup);
ASSERT_TRUE(request_started);
ASSERT_TRUE(subscribed);
ASSERT_TRUE(worker_boundary_started);
ASSERT_TRUE(worker_lock_attempted);
ASSERT_TRUE(worker_waiting);
ASSERT_TRUE(final_unsubscribed);
ASSERT_TRUE(request_finished);
ASSERT_TRUE(job_terminal);
ASSERT_EQ(request.status, CBM_DAEMON_RUNTIME_APPLICATION_CANCELLED);
ASSERT_TRUE(cancellation_reported);
ASSERT_TRUE(worker_never_acquired);
ASSERT_EQ(atomic_load(&fake.starts), 1);
ASSERT_EQ(atomic_load(&fake.cancels), 1);
ASSERT_EQ(atomic_load(&fake.destroys), 1);
ASSERT_TRUE(external_lease_released);
ASSERT_TRUE(no_daemon_lease_leaked);
ASSERT_TRUE(leak_probe_released);
ASSERT_TRUE(stopped);
ASSERT_TRUE(locks_clean);
PASS();
}
TEST(daemon_application_cancels_mutation_wait_without_cancelling_shared_job) {
app_fake_worker_context_t fake;
app_fake_worker_context_init(&fake);
cbm_daemon_application_worker_ops_t worker_ops = {
.context = &fake,
.start = app_fake_worker_start,
.poll = app_fake_worker_poll,
.cancel = app_fake_worker_cancel,
.log_path = app_fake_worker_log_path,
.destroy = app_fake_worker_destroy,
};
cbm_daemon_application_config_t config = {.worker_ops = &worker_ops};
cbm_daemon_application_t *application = cbm_daemon_application_new(&config);
cbm_daemon_runtime_application_callbacks_t callbacks =
cbm_daemon_application_runtime_callbacks(application);
cbm_daemon_runtime_application_session_t *index_session = app_test_open(&callbacks, 35);
cbm_daemon_runtime_application_session_t *mutation_session = app_test_open(&callbacks, 36);
char root[APP_TEST_PATH_CAP];
snprintf(root, sizeof(root), "%s/cbm-app-mutation-cancel-XXXXXX", cbm_tmpdir());
bool root_ok = cbm_mkdtemp(root) != NULL;
char *project = root_ok ? cbm_project_name_from_path(root) : NULL;
uint8_t *context = NULL;
uint32_t context_length = 0;
char index_args[APP_TEST_PATH_CAP + 32];
snprintf(index_args, sizeof(index_args), "{\"repo_path\":\"%s\"}", root);
uint8_t *index_tool = NULL;
uint32_t index_tool_length = 0;
char mutation_args[APP_TEST_PATH_CAP + 96];
snprintf(mutation_args, sizeof(mutation_args),
"{\"project\":\"%s\",\"mode\":\"update\",\"content\":\"x\"}", project ? project : "");
uint8_t *mutation_tool = NULL;
uint32_t mutation_tool_length = 0;
bool setup =
application && index_session && mutation_session && project &&
app_test_context_request(root, root, &context, &context_length) &&
app_test_tool_request("index_repository", index_args, &index_tool, &index_tool_length) &&
app_test_tool_request("manage_adr", mutation_args, &mutation_tool, &mutation_tool_length);
uint8_t *response = NULL;
uint32_t response_length = 0;
cbm_daemon_runtime_application_session_t *sessions[2] = {index_session, mutation_session};
for (size_t i = 0; setup && i < 2; i++) {
setup = app_test_request(&callbacks, sessions[i], context, context_length, &response,
&response_length) == CBM_DAEMON_RUNTIME_APPLICATION_OK;
free(response);
response = NULL;
}
app_request_thread_t index_request = {
.callbacks = callbacks,
.session = index_session,
.request = index_tool,
.request_length = index_tool_length,
};
app_request_thread_t mutation_request = {
.callbacks = callbacks,
.session = mutation_session,
.request = mutation_tool,
.request_length = mutation_tool_length,
};
cbm_thread_t index_thread;
cbm_thread_t mutation_thread;
bool index_started =
setup && cbm_thread_create(&index_thread, 0, app_request_thread, &index_request) == 0;
bool index_active = index_started && app_wait_for_subscribers(application, project, 1) &&
app_wait_for_atomic_int(&fake.starts, 1);
bool mutation_started =
index_active &&
cbm_thread_create(&mutation_thread, 0, app_request_thread, &mutation_request) == 0;
uint64_t blocked_deadline = cbm_now_ms() + 100;
while (mutation_started && !atomic_load(&mutation_request.done) &&
cbm_now_ms() < blocked_deadline) {
cbm_usleep(1000);
}
bool initially_blocked = mutation_started && !atomic_load(&mutation_request.done);
if (initially_blocked) {
callbacks.session_cancel(callbacks.context, mutation_session);
}
uint64_t cancel_deadline = cbm_now_ms() + APP_TEST_TIMEOUT_MS;
while (mutation_started && !atomic_load(&mutation_request.done) &&
cbm_now_ms() < cancel_deadline) {
cbm_usleep(1000);
}
bool cancelled_wait_finished = mutation_started && atomic_load(&mutation_request.done);
int physical_cancels_before_release = atomic_load(&fake.cancels);
atomic_store(&fake.allow_completion, true);
if (mutation_started) {
(void)cbm_thread_join(&mutation_thread);
}
if (index_started) {
(void)cbm_thread_join(&index_thread);
}
bool cancellation_reported =
mutation_request.status == CBM_DAEMON_RUNTIME_APPLICATION_CANCELLED &&
mutation_request.response == NULL && mutation_request.response_length == 0;
callbacks.session_close(callbacks.context, index_session);
callbacks.session_close(callbacks.context, mutation_session);
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
free(index_request.response);
free(mutation_request.response);
free(context);
free(index_tool);
free(mutation_tool);
free(project);
(void)cbm_rmdir(root);
ASSERT_TRUE(setup);
ASSERT_TRUE(index_active);
ASSERT_TRUE(mutation_started);
ASSERT_TRUE(initially_blocked);
ASSERT_TRUE(cancelled_wait_finished);
ASSERT_TRUE(cancellation_reported);
ASSERT_EQ(physical_cancels_before_release, 0);
ASSERT_EQ(index_request.status, CBM_DAEMON_RUNTIME_APPLICATION_OK);
ASSERT_TRUE(stopped);
PASS();
}
TEST(daemon_application_recovers_with_unique_per_job_quarantine_files) {
const char *old_restarts = getenv("CBM_INDEX_MAX_RESTARTS");
char *saved_restarts = old_restarts ? cbm_strdup(old_restarts) : NULL;
bool saved_ok = !old_restarts || saved_restarts;
const char *old_cache = getenv("CBM_CACHE_DIR");
bool had_cache = old_cache != NULL;
char *saved_cache = old_cache ? cbm_strdup(old_cache) : NULL;
bool saved_cache_ok = !had_cache || saved_cache;
app_fake_worker_context_t fake;
app_fake_worker_context_init(&fake);
atomic_store(&fake.scripted, true);
for (int base = 0; base <= 4; base += 4) {
fake.outcomes[base] = CBM_PROC_CRASH;
fake.outcomes[base + 1] = CBM_PROC_CRASH;
fake.outcomes[base + 2] = CBM_PROC_CRASH;
fake.outcomes[base + 3] = CBM_PROC_CLEAN;
fake.marker_payloads[base + 1] = "S crashing.c\n";
fake.marker_payloads[base + 2] = "S crashing.c\n";
}
cbm_daemon_application_worker_ops_t worker_ops = {
.context = &fake,
.start = app_fake_worker_start,
.poll = app_fake_worker_poll,
.cancel = app_fake_worker_cancel,
.log_path = app_fake_worker_log_path,
.destroy = app_fake_worker_destroy,
};
cbm_daemon_application_config_t config = {.worker_ops = &worker_ops};
cbm_daemon_application_t *application = cbm_daemon_application_new(&config);
char root[APP_TEST_PATH_CAP];
char cache[APP_TEST_PATH_CAP];
snprintf(root, sizeof(root), "%s/cbm-app-recovery-XXXXXX", cbm_tmpdir());
snprintf(cache, sizeof(cache), "%s/cbm-app-recovery-cache-XXXXXX", cbm_tmpdir());
bool root_ok = cbm_mkdtemp(root) != NULL;
bool cache_ok = cbm_mkdtemp(cache) != NULL;
bool env_ok = saved_ok && saved_cache_ok && cache_ok &&
cbm_setenv("CBM_INDEX_MAX_RESTARTS", "3", 1) == 0 &&
cbm_setenv("CBM_CACHE_DIR", cache, 1) == 0;
int first = application && root_ok && env_ok
? cbm_daemon_application_index(application, "recovery-one", root)
: -1;
int second = application && root_ok && env_ok
? cbm_daemon_application_index(application, "recovery-two", root)
: -1;
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
bool first_marker_removed = !cbm_file_exists(fake.marker_paths[1]);
bool first_quarantine_removed = !cbm_file_exists(fake.quarantine_paths[1]);
bool second_marker_removed = !cbm_file_exists(fake.marker_paths[5]);
bool second_quarantine_removed = !cbm_file_exists(fake.quarantine_paths[5]);
cbm_daemon_application_free(application);
if (saved_restarts) {
(void)cbm_setenv("CBM_INDEX_MAX_RESTARTS", saved_restarts, 1);
} else if (!old_restarts) {
(void)cbm_unsetenv("CBM_INDEX_MAX_RESTARTS");
}
free(saved_restarts);
if (saved_cache) {
(void)cbm_setenv("CBM_CACHE_DIR", saved_cache, 1);
} else if (!had_cache) {
(void)cbm_unsetenv("CBM_CACHE_DIR");
}
free(saved_cache);
ASSERT_TRUE(env_ok);
ASSERT_TRUE(root_ok);
ASSERT_TRUE(cache_ok);
ASSERT_EQ(first, 0);
ASSERT_EQ(second, 0);
ASSERT_TRUE(stopped);
ASSERT_EQ(atomic_load(&fake.starts), 8);
ASSERT_EQ(atomic_load(&fake.destroys), 8);
ASSERT_STR_EQ(fake.marker_paths[0], "");
ASSERT_STR_EQ(fake.quarantine_paths[0], "");
ASSERT_TRUE(fake.marker_paths[1][0] != '\0');
ASSERT_TRUE(fake.quarantine_paths[1][0] != '\0');
ASSERT_STR_EQ(fake.marker_paths[1], fake.marker_paths[2]);
ASSERT_STR_EQ(fake.marker_paths[2], fake.marker_paths[3]);
ASSERT_STR_EQ(fake.quarantine_paths[1], fake.quarantine_paths[3]);
ASSERT_STR_NEQ(fake.marker_paths[1], fake.quarantine_paths[1]);
ASSERT_STR_NEQ(fake.marker_paths[1], fake.marker_paths[5]);
ASSERT_STR_NEQ(fake.quarantine_paths[1], fake.quarantine_paths[5]);
ASSERT_TRUE(strstr(fake.quarantine_seen[3], "crashing.c\tcrash\n") != NULL);
ASSERT_TRUE(strstr(fake.quarantine_seen[7], "crashing.c\tcrash\n") != NULL);
ASSERT_TRUE(first_marker_removed);
ASSERT_TRUE(first_quarantine_removed);
ASSERT_TRUE(second_marker_removed);
ASSERT_TRUE(second_quarantine_removed);
(void)cbm_rmdir(root);
char logs[APP_TEST_PATH_CAP];
snprintf(logs, sizeof(logs), "%s/logs", cache);
(void)cbm_rmdir(logs);
(void)cbm_rmdir(cache);
PASS();
}
TEST(daemon_application_cancellation_between_recovery_attempts_stops_retry) {
app_fake_worker_context_t fake;
app_fake_worker_context_init(&fake);
atomic_store(&fake.scripted, true);
atomic_store(&fake.hold_destroy_attempt, 0);
fake.outcomes[0] = CBM_PROC_CRASH;
cbm_daemon_application_worker_ops_t worker_ops = {
.context = &fake,
.start = app_fake_worker_start,
.poll = app_fake_worker_poll,
.cancel = app_fake_worker_cancel,
.log_path = app_fake_worker_log_path,
.destroy = app_fake_worker_destroy,
};
cbm_daemon_application_config_t config = {.worker_ops = &worker_ops};
cbm_daemon_application_t *application = cbm_daemon_application_new(&config);
cbm_daemon_runtime_application_callbacks_t callbacks =
cbm_daemon_application_runtime_callbacks(application);
cbm_daemon_runtime_application_session_t *session = app_test_open(&callbacks, 41);
char root[APP_TEST_PATH_CAP];
snprintf(root, sizeof(root), "%s/cbm-app-retry-cancel-XXXXXX", cbm_tmpdir());
bool root_ok = cbm_mkdtemp(root) != NULL;
uint8_t *context = NULL;
uint32_t context_length = 0;
char args[APP_TEST_PATH_CAP + 32];
snprintf(args, sizeof(args), "{\"repo_path\":\"%s\"}", root);
uint8_t *tool = NULL;
uint32_t tool_length = 0;
bool setup = application && session && root_ok &&
app_test_context_request(root, root, &context, &context_length) &&
app_test_tool_request("index_repository", args, &tool, &tool_length);
uint8_t *empty = NULL;
uint32_t empty_length = 0;
if (setup) {
setup = app_test_request(&callbacks, session, context, context_length, &empty,
&empty_length) == CBM_DAEMON_RUNTIME_APPLICATION_OK;
}
free(empty);
app_request_thread_t request = {
.callbacks = callbacks,
.session = session,
.request = tool,
.request_length = tool_length,
};
cbm_thread_t thread;
bool started = setup && cbm_thread_create(&thread, 0, app_request_thread, &request) == 0;
bool between_attempts = started && app_wait_for_atomic_int(&fake.destroys, 1);
if (between_attempts) {
callbacks.session_cancel(callbacks.context, session);
}
atomic_store(&fake.release_destroy, true);
if (started) {
(void)cbm_thread_join(&thread);
}
callbacks.session_close(callbacks.context, session);
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
ASSERT_TRUE(setup);
ASSERT_TRUE(started);
ASSERT_TRUE(between_attempts);
ASSERT_TRUE(stopped);
ASSERT_EQ(atomic_load(&fake.starts), 1);
ASSERT_EQ(atomic_load(&fake.destroys), 1);
ASSERT_EQ(request.status, CBM_DAEMON_RUNTIME_APPLICATION_CANCELLED);
ASSERT_NULL(request.response);
ASSERT_EQ(request.response_length, 0);
free(request.response);
free(context);
free(tool);
(void)cbm_rmdir(root);
PASS();
}
TEST(daemon_application_thread_start_failure_rolls_back_job_reservation) {
app_fake_worker_context_t fake;
app_fake_worker_context_init(&fake);
atomic_store(&fake.allow_completion, true);
cbm_daemon_application_worker_ops_t worker_ops = {
.context = &fake,
.start = app_fake_worker_start,
.poll = app_fake_worker_poll,
.cancel = app_fake_worker_cancel,
.log_path = app_fake_worker_log_path,
.destroy = app_fake_worker_destroy,
};
cbm_daemon_application_config_t config = {.worker_ops = &worker_ops};
cbm_daemon_application_t *application = cbm_daemon_application_new(&config);
char root[APP_TEST_PATH_CAP];
(void)snprintf(root, sizeof(root), "%s/cbm-app-thread-start-XXXXXX", cbm_tmpdir());
bool root_ok = cbm_mkdtemp(root) != NULL;
int failed = 0;
if (application && root_ok) {
cbm_daemon_application_fail_next_job_thread_start_for_test();
failed = cbm_daemon_application_index(application, "thread-start", root);
}
size_t active_after_failure = application ? cbm_daemon_application_active_jobs(application) : 1;
size_t subscribers_after_failure =
application ? cbm_daemon_application_job_subscribers(application, "thread-start") : 1;
int retried = application && root_ok
? cbm_daemon_application_index(application, "thread-start", root)
: -1;
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
(void)cbm_rmdir(root);
ASSERT_TRUE(root_ok);
ASSERT_LT(failed, 0);
ASSERT_EQ(active_after_failure, 0);
ASSERT_EQ(subscribers_after_failure, 0);
ASSERT_EQ(retried, 0);
ASSERT_EQ(atomic_load(&fake.starts), 1);
ASSERT_EQ(atomic_load(&fake.destroys), 1);
ASSERT_TRUE(stopped);
PASS();
}
/* Session-request worker: issues one runtime request and records the raw
* response, so a request that legitimately BLOCKS (queued behind the
* physical job limit) can run off the test thread. */
typedef struct {
cbm_daemon_runtime_application_callbacks_t *callbacks;
cbm_daemon_runtime_application_session_t *session;
uint8_t *payload;
uint32_t payload_length;
uint8_t *response;
uint32_t response_length;
int status;
} app_session_request_thread_t;
static void *app_session_request_thread(void *opaque) {
app_session_request_thread_t *request = opaque;
request->status =
app_test_request(request->callbacks, request->session, request->payload,
request->payload_length, &request->response, &request->response_length);
return NULL;
}
/* An explicit index request that hits the physical job limit QUEUES (parked,
* observable via the busy-queue-wait counter) instead of erroring, and runs
* to completion once the occupying job finishes. The non-session coordinator
* API keeps its non-blocking busy answer. */
TEST(daemon_application_queues_explicit_index_behind_physical_job_limit) {
app_fake_worker_context_t fake;
app_fake_worker_context_init(&fake);
cbm_daemon_application_worker_ops_t worker_ops = {
.context = &fake,
.start = app_fake_worker_start,
.poll = app_fake_worker_poll,
.cancel = app_fake_worker_cancel,
.log_path = app_fake_worker_log_path,
.destroy = app_fake_worker_destroy,
};
cbm_daemon_application_config_t config = {
.worker_ops = &worker_ops,
.physical_job_limit = 1,
};
cbm_daemon_application_t *application = cbm_daemon_application_new(&config);
char first_root[APP_TEST_PATH_CAP];
char second_root[APP_TEST_PATH_CAP];
snprintf(first_root, sizeof(first_root), "%s/cbm-app-cap-first-XXXXXX", cbm_tmpdir());
snprintf(second_root, sizeof(second_root), "%s/cbm-app-cap-second-XXXXXX", cbm_tmpdir());
bool roots_ok = cbm_mkdtemp(first_root) != NULL && cbm_mkdtemp(second_root) != NULL;
app_index_thread_t first = {
.application = application,
.project = "cap-first",
.root = first_root,
.result = -1,
};
cbm_thread_t thread;
bool started =
application && roots_ok && cbm_thread_create(&thread, 0, app_index_thread, &first) == 0;
bool admitted = started && app_wait_for_subscribers(application, "cap-first", 1) &&
app_wait_for_atomic_int(&fake.starts, 1);
/* The programmatic coordinator API stays non-blocking: callers with their
* own retry policy (watcher, auto-index) still get the busy answer. */
int busy = admitted ? cbm_daemon_application_index(application, "cap-second", second_root) : -1;
cbm_daemon_runtime_application_callbacks_t callbacks =
cbm_daemon_application_runtime_callbacks(application);
cbm_daemon_runtime_application_session_t *session = app_test_open(&callbacks, 51);
uint8_t *session_context = NULL;
uint32_t session_context_length = 0;
char args[APP_TEST_PATH_CAP + 32];
snprintf(args, sizeof(args), "{\"repo_path\":\"%s\"}", second_root);
uint8_t *tool = NULL;
uint32_t tool_length = 0;
bool session_encoded = admitted && session &&
app_test_context_request(second_root, second_root, &session_context,
&session_context_length) &&
app_test_tool_request("index_repository", args, &tool, &tool_length);
uint8_t *context_response = NULL;
uint32_t context_response_length = 0;
bool context_ok = session_encoded &&
app_test_request(&callbacks, session, session_context, session_context_length,
&context_response, &context_response_length) ==
CBM_DAEMON_RUNTIME_APPLICATION_OK;
free(context_response);
int queue_waits_baseline = cbm_daemon_application_busy_queue_waits_for_test();
app_session_request_thread_t queued = {
.callbacks = &callbacks,
.session = session,
.payload = tool,
.payload_length = tool_length,
.status = -1,
};
cbm_thread_t request_thread;
bool request_started =
context_ok &&
cbm_thread_create(&request_thread, 0, app_session_request_thread, &queued) == 0;
/* Positive queue signal: the request entered the busy-wait loop instead
* of erroring — and no second worker was started while parked. */
bool queued_parked = false;
if (request_started) {
uint64_t deadline = cbm_now_ms() + APP_TEST_TIMEOUT_MS;
while (cbm_now_ms() < deadline) {
if (cbm_daemon_application_busy_queue_waits_for_test() > queue_waits_baseline) {
queued_parked = true;
break;
}
cbm_usleep(1000);
}
}
bool no_second_start_while_parked = queued_parked && atomic_load(&fake.starts) == 1;
/* Release the slot: the first job completes, the queued request must be
* admitted and complete on its own. */
atomic_store(&fake.allow_completion, true);
bool request_joined = request_started && cbm_thread_join(&request_thread) == 0;
bool queued_succeeded = request_joined && queued.status == CBM_DAEMON_RUNTIME_APPLICATION_OK &&
queued.response && strstr((char *)queued.response, "indexed") != NULL &&
strstr((char *)queued.response, "physical index job limit") == NULL;
callbacks.session_close(callbacks.context, session);
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
if (started) {
(void)cbm_thread_join(&thread);
}
cbm_daemon_application_free(application);
free(session_context);
free(tool);
free(queued.response);
ASSERT_TRUE(roots_ok);
ASSERT_TRUE(started);
ASSERT_TRUE(admitted);
ASSERT_EQ(busy, 1);
ASSERT_TRUE(context_ok);
ASSERT_TRUE(request_started);
ASSERT_TRUE(queued_parked);
ASSERT_TRUE(no_second_start_while_parked);
ASSERT_TRUE(queued_succeeded);
ASSERT_TRUE(stopped);
ASSERT_EQ(atomic_load(&fake.starts), 2);
ASSERT_EQ(atomic_load(&fake.destroys), 2);
(void)cbm_rmdir(first_root);
(void)cbm_rmdir(second_root);
PASS();
}
TEST(daemon_application_default_limit_admits_four_and_rejects_fifth) {
enum { DEFAULT_CAP_RUNNING = 4, DEFAULT_CAP_TOTAL = 5 };
const size_t aggregate_budget = 4099;
app_fake_worker_context_t fake;
app_fake_worker_context_init(&fake);
cbm_daemon_application_worker_ops_t worker_ops = {
.context = &fake,
.start = app_fake_worker_start,
.poll = app_fake_worker_poll,
.cancel = app_fake_worker_cancel,
.log_path = app_fake_worker_log_path,
.destroy = app_fake_worker_destroy,
};
cbm_daemon_application_config_t config = {
.worker_ops = &worker_ops,
.aggregate_memory_budget_bytes = aggregate_budget,
};
cbm_daemon_application_t *application = cbm_daemon_application_new(&config);
char roots[DEFAULT_CAP_TOTAL][APP_TEST_PATH_CAP];
char projects[DEFAULT_CAP_TOTAL][32];
bool roots_ok = true;
for (int i = 0; i < DEFAULT_CAP_TOTAL; i++) {
(void)snprintf(roots[i], sizeof(roots[i]), "%s/cbm-app-default-cap-%d-XXXXXX", cbm_tmpdir(),
i);
(void)snprintf(projects[i], sizeof(projects[i]), "default-cap-%d", i);
roots_ok = roots_ok && cbm_mkdtemp(roots[i]) != NULL;
}
app_index_thread_t requests[DEFAULT_CAP_RUNNING];
cbm_thread_t threads[DEFAULT_CAP_RUNNING];
bool started[DEFAULT_CAP_RUNNING] = {false};
for (int i = 0; application && roots_ok && i < DEFAULT_CAP_RUNNING; i++) {
requests[i] = (app_index_thread_t){
.application = application,
.project = projects[i],
.root = roots[i],
.result = -1,
};
started[i] = cbm_thread_create(&threads[i], 0, app_index_thread, &requests[i]) == 0;
}
bool all_started = true;
for (int i = 0; i < DEFAULT_CAP_RUNNING; i++) {
all_started = all_started && started[i];
}
bool four_admitted = all_started &&
app_wait_for_active_jobs(application, DEFAULT_CAP_RUNNING) &&
app_wait_for_atomic_int(&fake.starts, DEFAULT_CAP_RUNNING);
int fifth =
four_admitted ? cbm_daemon_application_index(application, projects[4], roots[4]) : -1;
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
for (int i = 0; i < DEFAULT_CAP_RUNNING; i++) {
if (started[i]) {
(void)cbm_thread_join(&threads[i]);
}
}
cbm_daemon_application_free(application);
for (int i = 0; i < DEFAULT_CAP_TOTAL; i++) {
(void)cbm_rmdir(roots[i]);
}
ASSERT_TRUE(roots_ok);
ASSERT_TRUE(all_started);
ASSERT_TRUE(four_admitted);
ASSERT_EQ(fifth, 1);
ASSERT_TRUE(stopped);
ASSERT_EQ(atomic_load(&fake.starts), DEFAULT_CAP_RUNNING);
ASSERT_EQ(atomic_load(&fake.cancels), DEFAULT_CAP_RUNNING);
ASSERT_EQ(atomic_load(&fake.destroys), DEFAULT_CAP_RUNNING);
size_t assigned_budget = 0;
for (int i = 0; i < DEFAULT_CAP_RUNNING; i++) {
ASSERT_EQ(fake.memory_budgets[i], aggregate_budget / DEFAULT_CAP_RUNNING);
assigned_budget += fake.memory_budgets[i];
}
ASSERT_TRUE(assigned_budget <= aggregate_budget);
PASS();
}
/* RED on the former void destruction contract: free could retain an
* application with live owners, but the daemon host had no way to observe that
* refusal and freed the application's borrowed dependencies anyway. */
TEST(daemon_application_free_reports_retained_live_ownership) {
cbm_daemon_application_t *application = cbm_daemon_application_new(NULL);
bool created = application != NULL;
bool mutation_held = application && cbm_daemon_application_project_mutation_try_begin(
application, "free-contract");
bool freed_while_busy =
application && mutation_held && cbm_daemon_application_free_with_timeout(application, 0);
bool retained_usable = application && mutation_held && !freed_while_busy &&
cbm_daemon_application_active_jobs(application) == 0;
if (retained_usable) {
cbm_daemon_application_project_mutation_end(application, "free-contract");
}
bool freed_after_release = retained_usable && cbm_daemon_application_free_with_timeout(
application, APP_TEST_TIMEOUT_MS);
ASSERT_TRUE(created);
ASSERT_TRUE(mutation_held);
ASSERT_FALSE(freed_while_busy);
ASSERT_TRUE(retained_usable);
ASSERT_TRUE(freed_after_release);
PASS();
}
TEST(daemon_application_rejects_clean_exit_when_process_tree_is_not_contained) {
app_fake_worker_context_t fake;
app_fake_worker_context_init(&fake);
atomic_store(&fake.allow_completion, true);
atomic_store(&fake.unsafe_clean, true);
cbm_daemon_application_worker_ops_t worker_ops = {
.context = &fake,
.start = app_fake_worker_start,
.poll = app_fake_worker_poll,
.cancel = app_fake_worker_cancel,
.log_path = app_fake_worker_log_path,
.destroy = app_fake_worker_destroy,
};
cbm_daemon_application_config_t config = {.worker_ops = &worker_ops};
cbm_daemon_application_t *application = cbm_daemon_application_new(&config);
int result = application ? cbm_daemon_application_index(application, "unsafe-clean-project",
cbm_tmpdir())
: -1;
bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS);
cbm_daemon_application_free(application);
ASSERT_NOT_NULL(application);
ASSERT_LT(result, 0);
ASSERT_TRUE(stopped);
ASSERT_EQ(atomic_load(&fake.starts), 1);
ASSERT_EQ(atomic_load(&fake.destroys), 1);
PASS();
}
/* #1375: a reply too large to frame must become a JSON-RPC ERROR, not a
* transport failure.
*
* Why this matters more than it looks: the frontend worker cannot tell a
* rejected oversized frame from a dead socket, and for a dead socket it
* deliberately _Exit()s the process (closing the kernel IPC handle is the only
* portable way to cancel daemon session ownership from a thread blocked in
* stdio). So passing an oversized reply DOWN to the transport killed the whole
* server — every tool gone for the rest of the session, exit=1, empty stderr.
* The substitution therefore has to happen here, at the layer that still holds
* the request id.
*
* Reproduced end-to-end before fixing, on a 20k-node fixture: LIMIT 10000 ->
* 8,529,990 bytes ok; LIMIT 20000 -> SERVER DIED exit=1. After: the same query
* returns this error and a follow-up query on the SAME session succeeds. */
TEST(daemon_application_oversized_reply_is_a_jsonrpc_error_not_a_death) {
cbm_jsonrpc_request_t request = {0};
request.has_id = true;
request.id = 42;
/* A reply that FITS must pass through byte-identical — the guard must not
* touch the overwhelming majority of replies. */
char *small = strdup("{\"jsonrpc\":\"2.0\",\"id\":42,\"result\":{}}");
ASSERT_NOT_NULL(small);
char *kept = cbm_daemon_application_framable_response_for_test(small, &request);
ASSERT_TRUE(kept == small); /* same pointer: untouched */
ASSERT_STR_EQ(kept, "{\"jsonrpc\":\"2.0\",\"id\":42,\"result\":{}}");
free(kept);
/* A reply that CANNOT be framed must come back as a JSON-RPC error instead.
* Passing it on is what killed the server: the frontend worker cannot tell a
* rejected oversized frame from a dead socket, and for a dead socket it
* deliberately _Exit()s the process — so every tool on that server was gone
* for the rest of the session, with exit=1 and an empty stderr (#1375).
* Reproduced end-to-end on a 20k-node fixture before fixing: LIMIT 10000 ->
* 8,529,990 bytes ok, LIMIT 20000 -> SERVER DIED. After: the same query
* returns this error and a follow-up query on the SAME session succeeds. */
size_t oversized = (size_t)CBM_DAEMON_RUNTIME_APPLICATION_PAYLOAD_MAX + 1U;
char *big = malloc(oversized + 1U);
ASSERT_NOT_NULL(big);
memset(big, 'x', oversized);
big[oversized] = '\0';
char *replaced = cbm_daemon_application_framable_response_for_test(big, &request);
ASSERT_NOT_NULL(replaced);
ASSERT_TRUE(replaced != big); /* substituted, not passed through */
/* The replacement must itself fit, or it reproduces the bug it replaces. */
ASSERT_TRUE(strlen(replaced) <= (size_t)CBM_DAEMON_RUNTIME_APPLICATION_PAYLOAD_MAX);
/* ...and carry the caller's id, or the client cannot match reply to request. */
ASSERT_NOT_NULL(strstr(replaced, "\"error\""));
ASSERT_NOT_NULL(strstr(replaced, "\"id\":42"));
ASSERT_NOT_NULL(strstr(replaced, "-32603"));
/* ...and say what happened and what to do: the failure it replaces was a
* SILENT exit, so an opaque "internal error" would be no improvement. */
ASSERT_NOT_NULL(strstr(replaced, "response too large"));
ASSERT_NOT_NULL(strstr(replaced, "LIMIT"));
free(replaced);
/* An unparseable message has no id to echo, but must still yield an error
* rather than NULL — NULL is turned back into a hard failure by the caller. */
char *big2 = malloc(oversized + 1U);
ASSERT_NOT_NULL(big2);
memset(big2, 'y', oversized);
big2[oversized] = '\0';
char *anonymous = cbm_daemon_application_framable_response_for_test(big2, NULL);
ASSERT_NOT_NULL(anonymous);
ASSERT_NOT_NULL(strstr(anonymous, "\"error\""));
free(anonymous);
PASS();
}
SUITE(daemon_application) {
RUN_TEST(daemon_application_oversized_reply_is_a_jsonrpc_error_not_a_death);
RUN_TEST(daemon_application_new_session_does_not_retain_initial_store);
RUN_TEST(daemon_application_request_cancel_is_scoped_to_exact_token);
RUN_TEST(daemon_application_requires_immutable_explicit_context);
RUN_TEST(daemon_application_ui_config_updates_are_masked_and_serialized);
RUN_TEST(daemon_application_ui_config_rejects_noncanonical_frames);
RUN_TEST(daemon_application_ui_readiness_proof_is_generation_bound_before_context);
RUN_TEST(daemon_application_restricted_profile_owns_no_background_surfaces);
RUN_TEST(daemon_application_hook_context_preserves_event_and_dialect);
RUN_TEST(daemon_application_mcp_notification_has_no_response);
RUN_TEST(daemon_application_reference_counts_one_shared_watch);
RUN_TEST(daemon_application_free_releases_live_watch_once);
RUN_TEST(daemon_application_prune_clears_logical_watch_for_reregistration);
RUN_TEST(daemon_application_initialize_coalesces_auto_index_for_full_sessions);
RUN_TEST(daemon_application_auto_index_honors_tracked_file_limit);
RUN_TEST(daemon_application_auto_index_file_count_handles_literal_metacharacter_path);
RUN_TEST(daemon_application_auto_index_file_count_supports_non_git_roots);
RUN_TEST(daemon_application_auto_index_retries_transient_busy_admission);
RUN_TEST(daemon_application_update_generation_notifies_initial_and_late_sessions_once);
RUN_TEST(daemon_application_update_generation_retries_worker_start_failure);
RUN_TEST(daemon_application_update_generation_retries_cancelled_check);
RUN_TEST(daemon_application_final_disconnect_cancels_and_joins_update_generation);
RUN_TEST(daemon_application_coalesces_semantically_identical_index_requests);
RUN_TEST(daemon_application_fresh_request_does_not_reuse_terminal_subscribed_job);
RUN_TEST(daemon_application_request_cancel_detaches_only_one_coalesced_subscriber);
RUN_TEST(daemon_application_cancels_physical_job_only_after_final_session);
RUN_TEST(daemon_application_disconnect_before_request_callback_is_sticky);
RUN_TEST(daemon_application_request_cancel_preserves_persistent_watch_and_session);
RUN_TEST(daemon_application_cancel_before_worker_start_skips_worker);
RUN_TEST(daemon_application_cancel_drops_watch_before_inflight_request_returns);
RUN_TEST(daemon_application_stale_watcher_callback_is_rejected_at_job_admission);
RUN_TEST(daemon_application_final_cancel_drains_admitted_watcher_job);
RUN_TEST(daemon_application_watcher_job_follows_exact_live_watch_owners);
RUN_TEST(daemon_application_late_watcher_session_owns_active_watcher_job);
RUN_TEST(daemon_application_serializes_adr_mutation_with_index_job);
RUN_TEST(daemon_application_reserved_mutation_delays_worker_start);
RUN_TEST(daemon_application_mutation_does_not_block_other_project);
RUN_TEST(daemon_application_mutation_still_owns_os_project_lock);
RUN_TEST(daemon_application_worker_boundary_is_sole_os_project_lock_owner);
RUN_TEST(daemon_application_worker_lock_serializes_external_mutation);
RUN_TEST(daemon_application_final_unsubscribe_cancels_external_lock_wait);
RUN_TEST(daemon_application_cancels_mutation_wait_without_cancelling_shared_job);
RUN_TEST(daemon_application_recovers_with_unique_per_job_quarantine_files);
RUN_TEST(daemon_application_cancellation_between_recovery_attempts_stops_retry);
RUN_TEST(daemon_application_thread_start_failure_rolls_back_job_reservation);
RUN_TEST(daemon_application_queues_explicit_index_behind_physical_job_limit);
RUN_TEST(daemon_application_default_limit_admits_four_and_rejects_fifth);
RUN_TEST(daemon_application_free_reports_retained_live_ownership);
RUN_TEST(daemon_application_rejects_clean_exit_when_process_tree_is_not_contained);
}
Reference in New Issue View Git Blame Copy Permalink