fix(mcp): resolve the root owner for unnamed index status polls

With #2134, an unnamed index_repository of a root that another indexed
project already owns adopts that owner's name, and the daemon keys the
job by it (application_index_project_key reads the forwarded "name").
The unnamed status poll that the #2144 async reply asks for ("the same
repo_path and status: true") still derived its key from the path alone
in index_status_project_key. For an owned root it never found the job
it had just started and answered "no index job or index is known for
project '<path-derived>'"; an ambiguous root got that same misleading
answer instead of the loud error the start returns.

index_status_project_key now runs the start path's
index_root_owner_resolve on the same canonical repo_path: an explicit
name still wins, a single other owner is the key, several owners fail
with the same error naming them, otherwise the path-derived key stays.
The status error buffer grows to hold that owner list.

The daemon's other readers of the key (job dedupe in
application_job_subscribe_locked, the synchronous cut-short marker set
and taken around application_index_run_sync) derive it from the
forwarded args, which already carry the owner name, so they agree with
the start; status was the only reader that diverged.

Refs #2134, #2144

Signed-off-by: Martin Vogel <martin.vogel.tech@gmail.com>

tests/test_daemon_application.c

@@ -6170,6 +6170,246 @@ TEST(daemon_application_async_refused_for_lone_one_shot_client_of_temporary_daem
     PASS();
 }
 
+/* ── #2134 x #2144: the unnamed status poll resolves the start's key ────────
+ *
+ * An unnamed index_repository of a root that another indexed project already
+ * owns adopts that owner's name (#2134), so the daemon keys the async job by
+ * it. The async reply tells the client to poll "with the same repo_path and
+ * status: true"; that unnamed poll must resolve the same key (owner first,
+ * then path-derived; several owners fail loudly), or it never finds the job
+ * it just started. Owners are real project databases in a private cache; the
+ * held fake worker never writes one, so the seeded state is all there is. */
+
+enum { APP_OWNER_CANONICAL_CAP = 4096 }; /* cbm_canonical_path needs >= 4096 */
+
+typedef struct {
+    char cache[APP_TEST_PATH_CAP];
+    char canonical_root[APP_OWNER_CANONICAL_CAP];
+    char *saved_cache;
+    bool had_cache;
+    bool cache_set;
+    app_async_fixture_t fixture;
+    cbm_daemon_runtime_application_session_t *session;
+} app_owner_fixture_t;
+
+typedef struct {
+    char *started;
+    char *running;
+    char *finished;
+    bool worker_running;
+    bool drained;
+} app_owner_run_t;
+
+static bool app_owner_fixture_init(app_owner_fixture_t *owner, const char *label) {
+    memset(owner, 0, sizeof(*owner));
+    const char *old_cache = getenv("CBM_CACHE_DIR");
+    owner->had_cache = old_cache != NULL;
+    owner->saved_cache = old_cache ? strdup(old_cache) : NULL;
+    (void)snprintf(owner->cache, sizeof(owner->cache), "%s/cbm-app-owner-cache-XXXXXX",
+                   cbm_tmpdir());
+    if ((owner->had_cache && !owner->saved_cache) || !cbm_mkdtemp(owner->cache)) {
+        owner->cache[0] = '\0';
+        return false;
+    }
+    owner->cache_set = cbm_setenv("CBM_CACHE_DIR", owner->cache, 1) == 0;
+    if (!owner->cache_set || !app_async_fixture_init(&owner->fixture, label, true) ||
+        !cbm_canonical_path(owner->fixture.root, owner->canonical_root,
+                            sizeof(owner->canonical_root))) {
+        return false;
+    }
+    /* Stored root_path values use forward slashes on every platform. */
+    cbm_normalize_path_sep(owner->canonical_root);
+    return true;
+}
+
+/* One indexed project whose stored root is this fixture's root. */
+static bool app_owner_seed(const app_owner_fixture_t *owner, const char *project) {
+    char path[APP_TEST_PATH_CAP];
+    int written = snprintf(path, sizeof(path), "%s/%s.db", owner->cache, project);
+    if (written <= 0 || (size_t)written >= sizeof(path)) {
+        return false;
+    }
+    cbm_store_t *store = cbm_store_open_path(path);
+    bool seeded =
+        store && cbm_store_upsert_project(store, project, owner->canonical_root) == CBM_STORE_OK;
+    cbm_store_close(store);
+    return seeded;
+}
+
+/* An unnamed async start, then unnamed status polls naming the same
+ * repo_path: one while the held worker runs, one after it drained. */
+static void app_owner_run(app_owner_fixture_t *owner, app_owner_run_t *run) {
+    memset(run, 0, sizeof(*run));
+    app_async_fixture_t *fixture = &owner->fixture;
+    run->started = app_async_call_now(fixture, owner->session, ",\"async\":true");
+    run->worker_running = run->started && !app_async_contains(run->started, "\"isError\":true") &&
+                          app_wait_for_atomic_int(&fixture->fake.starts, 1);
+    run->running = app_async_call_now(fixture, owner->session, ",\"status\":true");
+    atomic_store(&fixture->fake.allow_completion, true);
+    run->drained = run->worker_running && app_wait_for_active_jobs(fixture->application, 0);
+    run->finished =
+        run->drained ? app_async_call_now(fixture, owner->session, ",\"status\":true") : NULL;
+}
+
+static void app_owner_run_free(app_owner_run_t *run) {
+    free(run->started);
+    free(run->running);
+    free(run->finished);
+    memset(run, 0, sizeof(*run));
+}
+
+static bool app_owner_fixture_finish(app_owner_fixture_t *owner) {
+    if (owner->session) {
+        owner->fixture.callbacks.session_close(owner->fixture.callbacks.context, owner->session);
+        owner->session = NULL;
+    }
+    bool stopped = app_async_fixture_finish(&owner->fixture);
+    if (owner->cache_set) {
+        if (owner->had_cache) {
+            (void)cbm_setenv("CBM_CACHE_DIR", owner->saved_cache, 1);
+        } else {
+            (void)cbm_unsetenv("CBM_CACHE_DIR");
+        }
+    }
+    free(owner->saved_cache);
+    owner->saved_cache = NULL;
+    if (owner->cache[0]) {
+        (void)th_rmtree(owner->cache);
+    }
+    return stopped;
+}
+
+static bool app_owner_names_project(const char *response, const char *project) {
+    char needle[APP_TEST_PATH_CAP];
+    int written = snprintf(needle, sizeof(needle), "\"project\":\"%s\"", project);
+    return written > 0 && (size_t)written < sizeof(needle) && app_async_contains(response, needle);
+}
+
+TEST(daemon_application_unnamed_status_finds_async_job_of_root_owner_issue2134) {
+    app_owner_fixture_t owner;
+    bool setup =
+        app_owner_fixture_init(&owner, "owner-poll") && app_owner_seed(&owner, "owned-2134s");
+    owner.session = setup ? app_async_session(&owner.fixture, 7801) : NULL;
+    bool session_opened = owner.session != NULL;
+    app_owner_run_t run;
+    memset(&run, 0, sizeof(run));
+    if (session_opened) {
+        app_owner_run(&owner, &run);
+    }
+    int cancels = atomic_load(&owner.fixture.fake.cancels);
+    bool stopped = app_owner_fixture_finish(&owner);
+    if (!app_async_contains(run.running, "\"state\":\"running\"")) {
+        fprintf(stderr, "  [2134x2144] unnamed status missed the owner's job: %.400s\n",
+                run.running ? run.running : "(null)");
+    }
+
+    ASSERT_TRUE(setup);
+    ASSERT_TRUE(session_opened);
+    /* #2134: the unnamed start adopted the owner; the daemon keyed its job so. */
+    ASSERT_TRUE(app_owner_names_project(run.started, "owned-2134s"));
+    ASSERT_TRUE(run.worker_running);
+    /* The unnamed poll with the same repo_path reports that live job ... */
+    ASSERT_FALSE(app_async_contains(run.running, "\"isError\":true"));
+    ASSERT_TRUE(app_owner_names_project(run.running, "owned-2134s"));
+    ASSERT_TRUE(app_async_contains(run.running, "\"state\":\"running\""));
+    /* ... and, once drained, the outcome recorded under the same key. */
+    ASSERT_TRUE(run.drained);
+    ASSERT_TRUE(app_owner_names_project(run.finished, "owned-2134s"));
+    ASSERT_TRUE(app_async_contains(run.finished, "\"state\":\"succeeded\""));
+    ASSERT_EQ(cancels, 0);
+    ASSERT_TRUE(stopped);
+    app_owner_run_free(&run);
+    PASS();
+}
+
+/* Control: the path-derived key is untouched where #2134 keeps it - a root no
+ * other project owns, and a root whose path-derived project exists even though
+ * a named twin shares it. Start and poll name the same project either way. */
+TEST(daemon_application_unnamed_status_keeps_path_derived_key_issue2134) {
+    app_owner_fixture_t fresh;
+    bool fresh_setup = app_owner_fixture_init(&fresh, "owner-fresh");
+    fresh.session = fresh_setup ? app_async_session(&fresh.fixture, 7811) : NULL;
+    char *fresh_project = fresh.fixture.project ? strdup(fresh.fixture.project) : NULL;
+    app_owner_run_t fresh_run;
+    memset(&fresh_run, 0, sizeof(fresh_run));
+    if (fresh.session) {
+        app_owner_run(&fresh, &fresh_run);
+    }
+    bool fresh_stopped = app_owner_fixture_finish(&fresh);
+
+    app_owner_fixture_t twin;
+    bool twin_setup = app_owner_fixture_init(&twin, "owner-twin") && twin.fixture.project &&
+                      app_owner_seed(&twin, twin.fixture.project) &&
+                      app_owner_seed(&twin, "twin-2134s");
+    twin.session = twin_setup ? app_async_session(&twin.fixture, 7812) : NULL;
+    char *twin_project = twin.fixture.project ? strdup(twin.fixture.project) : NULL;
+    app_owner_run_t twin_run;
+    memset(&twin_run, 0, sizeof(twin_run));
+    if (twin.session) {
+        app_owner_run(&twin, &twin_run);
+    }
+    bool twin_stopped = app_owner_fixture_finish(&twin);
+
+    ASSERT_TRUE(fresh_setup);
+    ASSERT_NOT_NULL(fresh_project);
+    ASSERT_TRUE(app_owner_names_project(fresh_run.started, fresh_project));
+    ASSERT_TRUE(fresh_run.worker_running);
+    ASSERT_TRUE(app_owner_names_project(fresh_run.running, fresh_project));
+    ASSERT_TRUE(app_async_contains(fresh_run.running, "\"state\":\"running\""));
+    ASSERT_TRUE(app_async_contains(fresh_run.finished, "\"state\":\"succeeded\""));
+    ASSERT_TRUE(fresh_stopped);
+
+    ASSERT_TRUE(twin_setup);
+    ASSERT_NOT_NULL(twin_project);
+    ASSERT_TRUE(app_owner_names_project(twin_run.started, twin_project));
+    ASSERT_FALSE(app_owner_names_project(twin_run.started, "twin-2134s"));
+    ASSERT_TRUE(twin_run.worker_running);
+    ASSERT_TRUE(app_owner_names_project(twin_run.running, twin_project));
+    ASSERT_TRUE(app_async_contains(twin_run.running, "\"state\":\"running\""));
+    ASSERT_TRUE(app_async_contains(twin_run.finished, "\"state\":\"succeeded\""));
+    ASSERT_TRUE(twin_stopped);
+    free(fresh_project);
+    free(twin_project);
+    app_owner_run_free(&fresh_run);
+    app_owner_run_free(&twin_run);
+    PASS();
+}
+
+/* Control: several owners and no path-derived project is ambiguous. The start
+ * still refuses and names them without spawning a worker, and the poll gives
+ * the same loud error instead of guessing (or pointing at a path-derived
+ * project nobody indexes). */
+TEST(daemon_application_unnamed_async_and_status_refuse_ambiguous_root_issue2134) {
+    app_owner_fixture_t owner;
+    bool setup = app_owner_fixture_init(&owner, "owner-ambiguous") &&
+                 app_owner_seed(&owner, "first-2134s") && app_owner_seed(&owner, "second-2134s");
+    owner.session = setup ? app_async_session(&owner.fixture, 7821) : NULL;
+    bool session_opened = owner.session != NULL;
+    app_owner_run_t run;
+    memset(&run, 0, sizeof(run));
+    if (session_opened) {
+        app_owner_run(&owner, &run);
+    }
+    int starts = atomic_load(&owner.fixture.fake.starts);
+    bool stopped = app_owner_fixture_finish(&owner);
+
+    ASSERT_TRUE(setup);
+    ASSERT_TRUE(session_opened);
+    ASSERT_TRUE(app_async_contains(run.started, "\"isError\":true"));
+    ASSERT_TRUE(app_async_contains(run.started, "several indexed projects share root_path"));
+    ASSERT_TRUE(app_async_contains(run.started, "first-2134s"));
+    ASSERT_TRUE(app_async_contains(run.started, "second-2134s"));
+    ASSERT_FALSE(run.worker_running);
+    ASSERT_EQ(starts, 0);
+    ASSERT_TRUE(app_async_contains(run.running, "\"isError\":true"));
+    ASSERT_TRUE(app_async_contains(run.running, "several indexed projects share root_path"));
+    ASSERT_TRUE(app_async_contains(run.running, "first-2134s"));
+    ASSERT_TRUE(app_async_contains(run.running, "second-2134s"));
+    ASSERT_TRUE(stopped);
+    app_owner_run_free(&run);
+    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);
@@ -6197,6 +6437,9 @@ SUITE(daemon_application) {
     RUN_TEST(daemon_application_cancelled_sync_index_reply_offers_async);
     RUN_TEST(daemon_application_cut_short_sync_index_advice_reaches_next_call);
     RUN_TEST(daemon_application_async_refused_for_lone_one_shot_client_of_temporary_daemon);
+    RUN_TEST(daemon_application_unnamed_status_finds_async_job_of_root_owner_issue2134);
+    RUN_TEST(daemon_application_unnamed_status_keeps_path_derived_key_issue2134);
+    RUN_TEST(daemon_application_unnamed_async_and_status_refuse_ambiguous_root_issue2134);
     RUN_TEST(daemon_application_programmatic_index_injects_resource_policy);
     RUN_TEST(daemon_application_auto_index_honors_tracked_file_limit);
     RUN_TEST(daemon_application_auto_index_file_count_handles_literal_metacharacter_path);