trending/codebase-memory-mcp
1
0
Fork 0
mirror of https://github.com/DeusData/codebase-memory-mcp.git synced 2026-11-09 09:11:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(pipeline): defer containment semantic refresh

Browse Source 
Add opt-in incremental_derived_refresh=stale_on_incremental for full/moderate containment incremental publishes. The default remains eager; when configured, containment skips global semantic/similarity refresh and marks semantic_edges stale so existing MCP/query warnings remain accurate.

Validation: diff-check, source-safety, ASan/UBSan test-runner rebuild, pipeline, cli, mcp, pagerank, graph_buffer, incremental, and product build logs are recorded in the plan and forensics notes.
Signed-off-by: Andrew Hundt <ATHundt@gmail.com>
This commit is contained in:
Andrew Hundt
2026-07-05 06:48:13 -04:00
parent 9ab9fd0cb9
commit 43d100acaa
6 changed files with 125 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/cli/cli.c
+5 -3
View File
@@ -2927,12 +2927,14 @@ const cbm_config_entry_t CBM_CONFIG_REGISTRY[] = {
CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_EAGER,
NULL,
"Indexing",
"When exact incremental publishes may defer global semantic/similarity edge refresh",
"When incremental publishes may defer global semantic/similarity edge refresh",
CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_EAGER "|"
CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_STALE_ON_EXACT,
CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_STALE_ON_EXACT "|"
CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_STALE_ON_INCREMENTAL,
"'eager' preserves full/moderate publish freshness. 'stale_on_exact' lets small exact "
"incremental graph deltas publish after marking semantic_edges stale; semantic/similarity "
"queries warn until an eager index run or full reindex rebuilds those global edges."},
"queries warn until an eager index run or full reindex rebuilds those global edges. "
"'stale_on_incremental' also allows containment incremental publishes to defer."},
/* ── Search ── */
{"search_limit", "50", NULL, "Search",
"Default max results for search_graph/search_code",
src/pipeline/pipeline.c
+24 -8
View File
@@ -83,6 +83,7 @@ typedef enum {
typedef enum {
CBM_INCREMENTAL_DERIVED_REFRESH_EAGER = 0,
CBM_INCREMENTAL_DERIVED_REFRESH_STALE_ON_EXACT,
CBM_INCREMENTAL_DERIVED_REFRESH_STALE_ON_INCREMENTAL,
} cbm_incremental_derived_refresh_policy_t;
void cbm_pipeline_lock(void) {
@@ -363,12 +364,17 @@ void cbm_pipeline_apply_config(cbm_pipeline_t *p, cbm_config_t *cfg) {
const char *derived_refresh = cbm_config_get(cfg, CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH,
CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_EAGER);
p->incremental_derived_refresh =
derived_refresh &&
strcmp(derived_refresh,
CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_STALE_ON_EXACT) == 0
? CBM_INCREMENTAL_DERIVED_REFRESH_STALE_ON_EXACT
: CBM_INCREMENTAL_DERIVED_REFRESH_EAGER;
if (derived_refresh &&
strcmp(derived_refresh, CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_STALE_ON_INCREMENTAL) ==
0) {
p->incremental_derived_refresh = CBM_INCREMENTAL_DERIVED_REFRESH_STALE_ON_INCREMENTAL;
} else if (derived_refresh &&
strcmp(derived_refresh, CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_STALE_ON_EXACT) ==
0) {
p->incremental_derived_refresh = CBM_INCREMENTAL_DERIVED_REFRESH_STALE_ON_EXACT;
} else {
p->incremental_derived_refresh = CBM_INCREMENTAL_DERIVED_REFRESH_EAGER;
}
int max_changed = cbm_config_get_int(cfg, CBM_CONFIG_INCREMENTAL_EXACT_MAX_CHANGED_PATHS,
p->exact_delta_max_changed_paths);
@@ -637,7 +643,16 @@ bool cbm_pipeline_overlay_publish_small_deltas(const cbm_pipeline_t *p) {
}
bool cbm_pipeline_incremental_derived_refresh_stale_on_exact(const cbm_pipeline_t *p) {
return p && p->incremental_derived_refresh == CBM_INCREMENTAL_DERIVED_REFRESH_STALE_ON_EXACT;
return p &&
(p->incremental_derived_refresh == CBM_INCREMENTAL_DERIVED_REFRESH_STALE_ON_EXACT ||
p->incremental_derived_refresh ==
CBM_INCREMENTAL_DERIVED_REFRESH_STALE_ON_INCREMENTAL);
}
bool cbm_pipeline_incremental_derived_refresh_stale_on_incremental(const cbm_pipeline_t *p) {
return p &&
p->incremental_derived_refresh ==
CBM_INCREMENTAL_DERIVED_REFRESH_STALE_ON_INCREMENTAL;
}
cbm_pipeline_exact_delta_stats_t cbm_pipeline_exact_delta_stats(const cbm_pipeline_t *p) {
@@ -1532,7 +1547,8 @@ static int pipeline_persist_replacement_metadata(cbm_pipeline_t *p, cbm_store_t
return fts_rc;
}
}
int derived_rc = cbm_pipeline_mark_replacement_derived_views(store, p->project_name, p->mode);
int derived_rc =
cbm_pipeline_mark_replacement_derived_views(store, p->project_name, p->mode, true);
if (derived_rc != CBM_STORE_OK) {
cbm_log_error("pipeline.err", "phase", "mark_derived_views", "rc", itoa_buf(derived_rc));
return derived_rc;
src/pipeline/pipeline.h
+1
View File
@@ -125,6 +125,7 @@ void cbm_pipeline_set_flush_store(cbm_pipeline_t *p, cbm_store_t *store);
#define CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH "incremental_derived_refresh"
#define CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_EAGER "eager"
#define CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_STALE_ON_EXACT "stale_on_exact"
#define CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_STALE_ON_INCREMENTAL "stale_on_incremental"
/* Set the Jaccard similarity threshold for SIMILAR-edge creation (pass_similarity).
* <=0 (or unset) uses the CBM_MINHASH_JACCARD_THRESHOLD default. Before run(). */
src/pipeline/pipeline_incremental.c
+12 -4
View File
@@ -2371,7 +2371,8 @@ cleanup:
static int publish_and_persist(cbm_gbuf_t *gbuf, const char *db_path, const char *project,
cbm_file_info_t *files, int file_count,
const cbm_file_hash_t *mode_skipped, int mode_skipped_count,
const char *repo_path, const char *pass_fingerprint, int mode) {
const char *repo_path, const char *pass_fingerprint, int mode,
bool semantic_edges_refreshed) {
struct timespec t;
cbm_clock_gettime(CLOCK_MONOTONIC, &t);
@@ -2423,7 +2424,9 @@ static int publish_and_persist(cbm_gbuf_t *gbuf, const char *db_path, const char
cbm_store_close(hash_store);
return fts_rc;
}
int derived_rc = cbm_pipeline_mark_replacement_derived_views(hash_store, project, mode);
int derived_rc =
cbm_pipeline_mark_replacement_derived_views(hash_store, project, mode,
semantic_edges_refreshed);
if (derived_rc != CBM_STORE_OK) {
cbm_log_error("incremental.err", "phase", "mark_derived_views", "rc",
itoa_buf_incr(derived_rc));
@@ -2752,7 +2755,10 @@ int cbm_pipeline_run_incremental(cbm_pipeline_t *p, const char *db_path, cbm_fil
itoa_buf_incr(CBM_NOT_FOUND));
pipeline_rc = CBM_NOT_FOUND;
} else {
pipeline_rc = run_postpasses(&ctx, changed_files, ci, project, true);
bool refresh_global_semantic_edges =
!cbm_pipeline_incremental_derived_refresh_stale_on_incremental(p);
pipeline_rc = run_postpasses(&ctx, changed_files, ci, project,
refresh_global_semantic_edges);
}
}
@@ -2807,10 +2813,12 @@ int cbm_pipeline_run_incremental(cbm_pipeline_t *p, const char *db_path, cbm_fil
* covers incremental reindexes, not just full publishes. */
cbm_pipeline_set_committed_counts(p, cbm_gbuf_node_count(existing),
cbm_gbuf_edge_count(existing));
bool semantic_edges_refreshed =
!cbm_pipeline_incremental_derived_refresh_stale_on_incremental(p);
int persist_rc = publish_and_persist(existing, db_path, project, files, file_count,
cls.mode_skipped, cls.mode_skipped_count,
cbm_pipeline_repo_path(p), pass_fingerprint,
cbm_pipeline_get_mode(p));
cbm_pipeline_get_mode(p), semantic_edges_refreshed);
if (persist_rc == 0) {
cbm_pipeline_set_graph_changed(p, true);
cbm_pipeline_set_publish_kind(p, CBM_PIPELINE_PUBLISH_INCREMENTAL_CONTAINMENT);
src/pipeline/pipeline_internal.h
+4 -2
View File
@@ -438,7 +438,8 @@ static inline bool cbm_pipeline_mode_extracts_macro_nodes(int mode) {
}
static inline int cbm_pipeline_mark_replacement_derived_views(cbm_store_t *store,
const char *project, int mode) {
const char *project, int mode,
bool semantic_edges_refreshed) {
static const char *const complete_graph_views[] = {
CBM_STORE_DERIVED_VIEW_ROUTES,
CBM_STORE_DERIVED_VIEW_ARCHITECTURE,
@@ -459,7 +460,7 @@ static inline int cbm_pipeline_mark_replacement_derived_views(cbm_store_t *store
if (rc != CBM_STORE_OK) {
return rc;
}
if (cbm_pipeline_mode_builds_global_semantic_edges(mode)) {
if (cbm_pipeline_mode_builds_global_semantic_edges(mode) && semantic_edges_refreshed) {
return cbm_store_mark_derived_views_complete(
store, project, CBM_PIPELINE_COMPAT_GENERATION, semantic_views,
(int)(sizeof(semantic_views) / sizeof(semantic_views[0])));
@@ -1151,6 +1152,7 @@ void cbm_pipeline_set_publish_kind(cbm_pipeline_t *p, cbm_pipeline_publish_kind_
void cbm_pipeline_set_publish_reason(cbm_pipeline_t *p, const char *reason);
bool cbm_pipeline_overlay_publish_small_deltas(const cbm_pipeline_t *p);
bool cbm_pipeline_incremental_derived_refresh_stale_on_exact(const cbm_pipeline_t *p);
bool cbm_pipeline_incremental_derived_refresh_stale_on_incremental(const cbm_pipeline_t *p);
void cbm_pipeline_set_exact_delta_stats(cbm_pipeline_t *p, int changed_paths,
int affected_paths, int published_paths);
void cbm_pipeline_set_exact_delta_stats_with_limit(cbm_pipeline_t *p, int changed_paths,
tests/test_pipeline.c
+79 -1
View File
@@ -11343,6 +11343,81 @@ TEST(incremental_full_stale_on_exact_mixed_delete_upsert_marks_semantic_stale) {
PASS();
}
TEST(incremental_full_stale_on_incremental_defers_containment_semantic_refresh) {
if (setup_incremental_repo() != 0) {
FAIL("setup failed");
}
char path[CBM_PATH_MAX];
int n = snprintf(path, sizeof(path), "%s/leaf.go", g_incr_tmpdir);
ASSERT(n >= 0 && (size_t)n < sizeof(path));
ASSERT_EQ(th_write_file(path,
"package main\n\n"
"func Leaf() int {\n\treturn 1\n}\n"),
0);
cbm_config_t *cfg = incremental_test_config(g_incr_tmpdir);
ASSERT_NOT_NULL(cfg);
ASSERT_EQ(cbm_config_set(cfg, CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH,
CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_STALE_ON_INCREMENTAL),
0);
cbm_pipeline_t *p = cbm_pipeline_new(g_incr_tmpdir, g_incr_dbpath, CBM_MODE_FULL);
ASSERT_NOT_NULL(p);
cbm_pipeline_apply_config(p, cfg);
ASSERT_EQ(cbm_pipeline_run(p), 0);
char *project = cbm_strdup(cbm_pipeline_project_name(p));
cbm_pipeline_free(p);
ASSERT_NOT_NULL(project);
n = snprintf(path, sizeof(path), "%s/main.go", g_incr_tmpdir);
ASSERT(n >= 0 && (size_t)n < sizeof(path));
ASSERT_EQ(th_write_file(path,
"package main\n\n"
"func main() {\n\tHelper()\n\tNewHelper()\n\tLeaf()\n}\n\n"
"func NewMain() int {\n\treturn 11\n}\n"),
0);
n = snprintf(path, sizeof(path), "%s/helper.go", g_incr_tmpdir);
ASSERT(n >= 0 && (size_t)n < sizeof(path));
ASSERT_EQ(th_write_file(path,
"package main\n\n"
"func Helper() string {\n\treturn \"updated\"\n}\n\n"
"func NewHelper() int {\n\treturn 13\n}\n"),
0);
n = snprintf(path, sizeof(path), "%s/leaf.go", g_incr_tmpdir);
ASSERT(n >= 0 && (size_t)n < sizeof(path));
ASSERT_EQ(th_write_file(path,
"package main\n\n"
"func Leaf() int {\n\treturn 2\n}\n\n"
"func NewLeaf() int {\n\treturn 17\n}\n"),
0);
p = cbm_pipeline_new(g_incr_tmpdir, g_incr_dbpath, CBM_MODE_FULL);
ASSERT_NOT_NULL(p);
cbm_pipeline_apply_config(p, cfg);
pipeline_capture_logs_start();
int run_rc = cbm_pipeline_run(p);
const char *logs = pipeline_capture_logs_end();
ASSERT_EQ(run_rc, 0);
ASSERT(strstr(logs, "msg=incremental.exact.skip reason=changed_batch_too_large") != NULL);
ASSERT(strstr(logs, "pass=incr_semantic_edges") == NULL);
ASSERT_EQ(cbm_pipeline_publish_kind(p), CBM_PIPELINE_PUBLISH_INCREMENTAL_CONTAINMENT);
ASSERT_STR_EQ(cbm_pipeline_publish_reason(p), "changed_batch_too_large");
cbm_pipeline_free(p);
cbm_store_t *s = cbm_store_open_path(g_incr_dbpath);
ASSERT_NOT_NULL(s);
ASSERT_TRUE(pipeline_test_derived_status_is(s, project,
CBM_STORE_DERIVED_VIEW_SEMANTIC_EDGES,
CBM_STORE_DERIVED_STATUS_STALE));
cbm_store_close(s);
free(project);
cbm_config_close(cfg);
cleanup_incremental_repo();
PASS();
}
TEST(incremental_fast_mixed_unowned_edge_frontier_falls_back_to_full_rebuild) {
enum { PIPELINE_CONFIGURED_AFFECTED_CAP = CBM_SZ_8 };
if (setup_incremental_repo() != 0) {
@@ -14637,9 +14712,11 @@ TEST(config_registry_includes_incremental_derived_refresh_policy) {
ASSERT_NOT_NULL(entry);
ASSERT_STR_EQ(entry->default_val, CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_EAGER);
ASSERT_STR_EQ(entry->category, "Indexing");
ASSERT_STR_EQ(entry->range, "eager|stale_on_exact");
ASSERT_STR_EQ(entry->range, "eager|stale_on_exact|stale_on_incremental");
ASSERT_NOT_NULL(strstr(entry->guidance,
CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_STALE_ON_EXACT));
ASSERT_NOT_NULL(strstr(entry->guidance,
CBM_CONFIG_INCREMENTAL_DERIVED_REFRESH_STALE_ON_INCREMENTAL));
PASS();
}
@@ -15631,6 +15708,7 @@ SUITE(pipeline) {
RUN_TEST(incremental_fast_configured_frontier_cap_allows_bounded_exact);
RUN_TEST(incremental_full_stale_on_exact_defers_global_derived_refresh);
RUN_TEST(incremental_full_stale_on_exact_mixed_delete_upsert_marks_semantic_stale);
RUN_TEST(incremental_full_stale_on_incremental_defers_containment_semantic_refresh);
RUN_TEST(incremental_fast_mixed_unowned_edge_frontier_falls_back_to_full_rebuild);
RUN_TEST(incremental_fast_expands_small_inbound_frontier_and_matches_full);
RUN_TEST(incremental_fast_three_file_batch_falls_back_to_full_rebuild_parity);