feat(ha-term-safety): on_get_entry_term + on_truncate_from callback bodies (task 2.2)

on_get_entry_term → persist_entry_term (recorded term, 0=absent, legacy-0→1; caveat
documented per 1.3 review). on_truncate_from → persist_truncate_wal_above(p, index-1)
(delete LSN>=index), returns rc for library/leader retry; non-promotable fence intact;
replaces the committed-boundary heuristic on the library-driven path. New
truncate_term_index_locked keeps the in-memory per-LSN term index consistent on both
truncate paths (under lock, cutoff-aligned) → no stale-term read after truncation.
3 tests, reviewer APPROVED, per-group 0 failures.

.kiro/specs/ha-term-safety/tasks.md

@@ -38,7 +38,7 @@
   - _Requirements: 1.1, 1.2, 1.3, 1.4_
   - _Depends: 1.3_
 
-- [ ] 2.2 Implement `on_get_entry_term` and `on_truncate_from`
+- [x] 2.2 Implement `on_get_entry_term` and `on_truncate_from`
   - `on_get_entry_term` answers from the recovered/live per-LSN term index (legacy-0 rule from 1.3); `on_truncate_from` deletes entries at index..end via the existing LSN truncate, replacing the committed-boundary heuristic
   - Observable completion: the library reads back a stamped entry term via the callback; a truncate-from call removes the local suffix at and above the index and leaves the prefix intact
   - _Requirements: 1.5, 1.8, 3.1, 3.2_
@@ -153,3 +153,5 @@
   - **DEFERRED to task 3.1 (review):** primary path uses persist_wal_async's `current_lsn`, NOT the library `index` — holds only while `s->log_len` and `p->current_lsn` stay co-tracked. Task 3.1 MUST retire/disable the hand-rolled primary append so no WAL append bypasses raft_log_append, else current_lsn diverges from s->log_len. Also: persist_set_term + write are two lock regions (safe today, no concurrent appender) — 3.1 should re-verify when real concurrent appends exist.
   - **DEFERRED to task 3.2 (review):** a heartbeat / 0-entry AppendEntries must NOT call on_persist_entry (entry==NULL,len==0 currently falls through to WAL_NOOP) — verify the library only calls on_persist_entry when there is an entry.
   - **3.1/3.2 producer format:** must pack `raft_entry_blob_t {type,payload_len}+payload` to match the on_persist_entry unpack (single shared def in wal_types.h).
+
+- **2.2 done.** on_get_entry_term → persist_entry_term (legacy-0 caveat documented per 1.3 review); on_truncate_from → persist_truncate_wal_above(p,index-1) (delete LSN>=index), returns rc, non-promotable fence preserved. NEW `truncate_term_index_locked` (persist.c) shrinks the in-memory term index on BOTH truncate return paths (under p->lock, cutoff-aligned) so on_get_entry_term returns 0 for truncated LSNs — no stale-term hazard. 3 tests. Reviewer APPROVED.

src/controller/ha.c

@@ -1414,10 +1414,27 @@ static int ha_cb_persist_entry_stub(void *ctx, uint64_t index, uint64_t term,
     return persist_wal_write_replicated(p, index, type, payload, plen);
 }
 
+/* on_get_entry_term (ha-term-safety task 2.2, R1.5/R3.1): return the recorded
+ * Raft term for the entry at 1-based LSN `index`, sourced from the WAL term
+ * index via persist_entry_term. Returns 0 when the LSN is absent (no entry).
+ * The library calls this for its logOk / (term,lsn) reconcile gate.
+ *
+ * Legacy-0 caveat (from the task 1.3 review MEDIUM): persist_entry_term maps a
+ * present legacy-0 entry (hdr.term == 0, written before this feature) to
+ * WAL_LEGACY_TERM (== 1). In a mixed pre/post-feature cluster where a
+ * post-feature leader is genuinely in term 1, a (term,lsn) EQUALITY check
+ * cannot tell a legacy-0 entry (mapped to 1) apart from a real term-1 entry by
+ * term alone. This is safe because the Raft log-matching invariant (same term
+ * + same index ⇒ same entry, enforced across the prefix) plus the bounded
+ * one-time upgrade window (every entry written after upgrade carries a real
+ * term) mean the only collision is at the legacy/term-1 boundary, where the
+ * prefixes already agree. We note it explicitly rather than special-casing it. */
 static uint64_t ha_cb_get_entry_term_stub(void *ctx, uint64_t index)
 {
-    (void)ctx; (void)index;
-    return 0; /* TODO task 2.2: 0 == absent (logOk reads local last-log term as 0) */
+    ha_state_t *ha = (ha_state_t *)ctx;
+    if (!ha || !ha->raft_c)
+        return 0; /* test mode without a cluster: no log → absent */
+    return persist_entry_term(&ha->raft_c->persist, index);
 }
 
 static int ha_cb_send_append_entries_stub(void *ctx, int peer_idx,
@@ -1430,10 +1447,26 @@ static int ha_cb_send_append_entries_stub(void *ctx, int peer_idx,
     return 0; /* TODO task 2.3 */
 }
 
+/* on_truncate_from (ha-term-safety task 2.2, R1.8/R3.2): delete the local log
+ * entries at LSN `index..end` because they conflict with the leader's log at a
+ * (term,lsn) the library detected. persist_truncate_wal_above(p, cutoff) removes
+ * every entry with LSN > cutoff, so to drop LSN >= index we pass cutoff =
+ * index - 1. This REPLACES the committed-boundary truncate heuristic on the
+ * library-driven path: the library, not a watermark, decides the cut point.
+ *
+ * Returns the truncate rc. A non-zero rc propagates so the library/leader
+ * retries; the member stays non-promotable per R3.5 (that fence is existing
+ * behavior and is not touched here). persist_truncate_wal_above also shrinks
+ * the per-LSN term index for the removed LSNs, so a later on_get_entry_term on
+ * a truncated LSN reads back 0 (absent). */
 static int ha_cb_truncate_from_stub(void *ctx, uint64_t index)
 {
-    (void)ctx; (void)index;
-    return 0; /* TODO task 2.4 */
+    ha_state_t *ha = (ha_state_t *)ctx;
+    if (!ha || !ha->raft_c)
+        return -EINVAL; /* test mode without a cluster: cannot truncate */
+    if (index == 0)
+        return -EINVAL; /* LSN 0 is reserved — never an entry to truncate */
+    return persist_truncate_wal_above(&ha->raft_c->persist, index - 1);
 }
 
 static void ha_cb_commit_advanced_stub(void *ctx, uint64_t commit_index)
@@ -1633,6 +1666,20 @@ int ha_cb_persist_entry_for_test(ha_state_t *ha, uint64_t index, uint64_t term,
 {
     return s_real_cbs.on_persist_entry(ha, index, term, entry, len);
 }
+
+/* ha-term-safety task 2.2: drive the WIRED on_get_entry_term callback so a unit
+ * test exercises the real entry-term accessor. Exposed only under XCE_TESTING. */
+uint64_t ha_cb_get_entry_term_for_test(ha_state_t *ha, uint64_t index)
+{
+    return s_real_cbs.on_get_entry_term(ha, index);
+}
+
+/* ha-term-safety task 2.2: drive the WIRED on_truncate_from callback so a unit
+ * test exercises the real conflict truncation. Exposed only under XCE_TESTING. */
+int ha_cb_truncate_from_for_test(ha_state_t *ha, uint64_t index)
+{
+    return s_real_cbs.on_truncate_from(ha, index);
+}
 #endif
 
 /* ha_check_heartbeat_nocluster() / ha_watchdog_thread() / ha_start_watchdog() /

src/controller/ha.h

@@ -771,6 +771,24 @@ int handle_request_vote(ha_state_t *ha,
  */
 int ha_cb_persist_entry_for_test(ha_state_t *ha, uint64_t index, uint64_t term,
                                  const void *entry, size_t len);
+
+/*
+ * ha_cb_get_entry_term_for_test (ha-term-safety task 2.2): invoke the WIRED
+ * on_get_entry_term callback directly so a unit test exercises the real
+ * entry-term accessor (persist_entry_term at the given 1-based index == LSN).
+ * Returns the recorded term (legacy-0 mapped to WAL_LEGACY_TERM), or 0 when the
+ * LSN is absent. TEST-ONLY.
+ */
+uint64_t ha_cb_get_entry_term_for_test(ha_state_t *ha, uint64_t index);
+
+/*
+ * ha_cb_truncate_from_for_test (ha-term-safety task 2.2): invoke the WIRED
+ * on_truncate_from callback directly so a unit test exercises the real conflict
+ * truncation (delete local entries at LSN >= index). Returns the truncate rc
+ * (0 on success; non-zero propagates so the library/leader retries while the
+ * member stays non-promotable). TEST-ONLY.
+ */
+int ha_cb_truncate_from_for_test(ha_state_t *ha, uint64_t index);
 #endif
 
 /*

src/controller/persist.c

@@ -440,6 +440,21 @@ static void record_entry_term(persist_t* p, uint64_t lsn, uint64_t term)
         p->term_index_count = lsn;
 }
 
+/* Shrink the in-memory per-LSN term index so that every LSN > `keep_lsn` reads
+ * back as absent (ha-term-safety task 2.2). Called from the truncate path after
+ * the on-disk entries above keep_lsn are removed, keeping the term index
+ * consistent with the WAL: a later persist_entry_term on a truncated LSN must
+ * return 0. The backing allocation is left as-is (only the live count moves);
+ * the zeroed-on-grow contract means a later re-append at those LSNs overwrites
+ * cleanly. Caller must hold p->lock. */
+static void truncate_term_index_locked(persist_t* p, uint64_t keep_lsn)
+{
+    if (!p)
+        return;
+    if (keep_lsn < p->term_index_count)
+        p->term_index_count = keep_lsn;
+}
+
 void persist_set_term(persist_t* p, uint64_t term)
 {
     if (!p)
@@ -2659,6 +2674,8 @@ int persist_truncate_wal_above(persist_t* p, uint64_t cutoff_lsn)
         atomic_store_explicit(&p->committed_lsn, cutoff_lsn,
                               memory_order_release);
         clamp_durable_lsn_locked(p, cutoff_lsn);
+        /* Keep the per-LSN term index consistent: drop terms for removed LSNs. */
+        truncate_term_index_locked(p, cutoff_lsn);
         pthread_mutex_unlock(&p->lock);
         return 0;
     }
@@ -2738,6 +2755,8 @@ int persist_truncate_wal_above(persist_t* p, uint64_t cutoff_lsn)
     p->last_crc    = new_last_crc;
     atomic_store_explicit(&p->committed_lsn, cutoff_lsn, memory_order_release);
     clamp_durable_lsn_locked(p, cutoff_lsn);
+    /* Keep the per-LSN term index consistent: drop terms for removed LSNs. */
+    truncate_term_index_locked(p, cutoff_lsn);
 
     pthread_mutex_unlock(&p->lock);
     return 0;

src/controller/testsuite/test_term_safety_adapter.c

@@ -378,6 +378,122 @@ static void test_on_persist_entry_failure_propagates_no_ack(void)
     printf("test_on_persist_entry_failure_propagates_no_ack: PASS\n");
 }
 
+/* ================================================================== */
+/* Task 2.2 — on_get_entry_term + on_truncate_from callback bodies     */
+/* ================================================================== */
+
+/* Persist a couple of replicated entries (each at a 1-based index == LSN)
+ * through the WIRED on_persist_entry callback, stamping a leader term per
+ * entry. Returns the (started) cluster the caller tears down. */
+static cluster_t* persist_two_entries(const char* tmp, uint64_t t1, uint64_t t2)
+{
+    cluster_t* c = make_cluster(tmp);
+    if (!c) return NULL;
+    atomic_store(&c->ha.role, CTRL_STANDBY);
+
+    uint8_t blob[64];
+    size_t  n1 = pack_entry(blob, WAL_JOB_UPDATE, "a", 1);
+    if (ha_cb_persist_entry_for_test(&c->ha, /*index*/1, t1, blob, n1) != 0) {
+        teardown_cluster(c);
+        return NULL;
+    }
+    size_t n2 = pack_entry(blob, WAL_JOB_UPDATE, "b", 1);
+    if (ha_cb_persist_entry_for_test(&c->ha, /*index*/2, t2, blob, n2) != 0) {
+        teardown_cluster(c);
+        return NULL;
+    }
+    return c;
+}
+
+/* Test: on_get_entry_term returns the stamped per-LSN term for present entries,
+ * the lowest-known term for a legacy-0 entry, and 0 for an absent LSN
+ * (R1.5, R3.1). */
+static void test_on_get_entry_term_present_legacy_absent(void)
+{
+    char tmp[256];
+    strcpy(tmp, "/tmp/xce_ts_get_XXXXXX");
+    TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
+
+    cluster_t* c = make_cluster(tmp);
+    TEST_ASSERT_NOT_NULL(c);
+    atomic_store(&c->ha.role, CTRL_STANDBY);
+
+    uint8_t blob[64];
+
+    /* LSN 1: real term 6. */
+    size_t n = pack_entry(blob, WAL_JOB_UPDATE, "r", 1);
+    TEST_ASSERT_EQUAL_INT(0,
+        ha_cb_persist_entry_for_test(&c->ha, /*index*/1, /*term*/6, blob, n));
+
+    /* LSN 2: legacy-0 entry (leader term 0) → reads back as WAL_LEGACY_TERM. */
+    n = pack_entry(blob, WAL_JOB_UPDATE, "l", 1);
+    TEST_ASSERT_EQUAL_INT(0,
+        ha_cb_persist_entry_for_test(&c->ha, /*index*/2, /*term*/0, blob, n));
+
+    /* Present real-term entry returns its term verbatim. */
+    TEST_ASSERT_EQUAL_UINT64(6, ha_cb_get_entry_term_for_test(&c->ha, 1));
+    /* Present legacy-0 entry maps to the lowest-known real term. */
+    TEST_ASSERT_EQUAL_UINT64(LEGACY_LOWEST_TERM,
+                             ha_cb_get_entry_term_for_test(&c->ha, 2));
+    /* Absent LSNs (0 reserved, 3 never appended) return 0. */
+    TEST_ASSERT_EQUAL_UINT64(0, ha_cb_get_entry_term_for_test(&c->ha, 0));
+    TEST_ASSERT_EQUAL_UINT64(0, ha_cb_get_entry_term_for_test(&c->ha, 3));
+
+    teardown_cluster(c);
+    printf("test_on_get_entry_term_present_legacy_absent: PASS\n");
+}
+
+/* Test: on_truncate_from(k) deletes local entries at LSN >= k; the prefix < k
+ * stays intact and on_get_entry_term returns 0 for the truncated LSNs while the
+ * term index shrinks consistently (R1.8, R3.1, R3.2). */
+static void test_on_truncate_from_drops_suffix(void)
+{
+    char tmp[256];
+    strcpy(tmp, "/tmp/xce_ts_trunc_XXXXXX");
+    TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
+
+    /* Persist LSN 1 (term 3) and LSN 2 (term 5). */
+    cluster_t* c = persist_two_entries(tmp, /*t1*/3, /*t2*/5);
+    TEST_ASSERT_NOT_NULL(c);
+
+    /* Both present before truncation. */
+    TEST_ASSERT_EQUAL_UINT64(3, ha_cb_get_entry_term_for_test(&c->ha, 1));
+    TEST_ASSERT_EQUAL_UINT64(5, ha_cb_get_entry_term_for_test(&c->ha, 2));
+    TEST_ASSERT_EQUAL_UINT64(3, c->persist.current_lsn);
+
+    /* Truncate from LSN 2 → entries at LSN >= 2 are deleted. */
+    TEST_ASSERT_EQUAL_INT(0, ha_cb_truncate_from_for_test(&c->ha, 2));
+
+    /* Prefix < 2 intact, suffix >= 2 gone (term index returns 0). */
+    TEST_ASSERT_EQUAL_UINT64(3, ha_cb_get_entry_term_for_test(&c->ha, 1));
+    TEST_ASSERT_EQUAL_UINT64(0, ha_cb_get_entry_term_for_test(&c->ha, 2));
+    TEST_ASSERT_EQUAL_UINT64(2, c->persist.current_lsn);
+
+    teardown_cluster(c);
+    printf("test_on_truncate_from_drops_suffix: PASS\n");
+}
+
+/* Test: on_truncate_from(1) deletes the whole log (LSN >= 1); every entry's
+ * term reads back absent and the tip collapses to LSN 1. */
+static void test_on_truncate_from_removes_all(void)
+{
+    char tmp[256];
+    strcpy(tmp, "/tmp/xce_ts_trunc_all_XXXXXX");
+    TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
+
+    cluster_t* c = persist_two_entries(tmp, /*t1*/2, /*t2*/2);
+    TEST_ASSERT_NOT_NULL(c);
+
+    TEST_ASSERT_EQUAL_INT(0, ha_cb_truncate_from_for_test(&c->ha, 1));
+
+    TEST_ASSERT_EQUAL_UINT64(0, ha_cb_get_entry_term_for_test(&c->ha, 1));
+    TEST_ASSERT_EQUAL_UINT64(0, ha_cb_get_entry_term_for_test(&c->ha, 2));
+    TEST_ASSERT_EQUAL_UINT64(1, c->persist.current_lsn);
+
+    teardown_cluster(c);
+    printf("test_on_truncate_from_removes_all: PASS\n");
+}
+
 /* ------------------------------------------------------------------ */
 /* Suite entry point                                                   */
 /* ------------------------------------------------------------------ */
@@ -392,4 +508,7 @@ void test_term_safety_adapter(void)
     RUN_TEST(test_on_persist_entry_stamps_term_standby);
     RUN_TEST(test_on_persist_entry_stamps_term_primary);
     RUN_TEST(test_on_persist_entry_failure_propagates_no_ack);
+    RUN_TEST(test_on_get_entry_term_present_legacy_absent);
+    RUN_TEST(test_on_truncate_from_drops_suffix);
+    RUN_TEST(test_on_truncate_from_removes_all);
 }