# Config Replication Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Implement config replication so standbys fetch `ClusterConfig` from the primary on startup, eliminating manual per-node provisioning.

**Architecture:** Split `cluster_init()` into Phase 1 (SSL/HA init, no config) and Phase 2 (iomgr/kvmgr/persist, config required). The primary loads config from `conf/current` (HEAD-file pointer → versioned `cluster.bin.<sha256>` file). Standbys fetch via `OP_replConfig` before Phase 2. SHA-256 is used as both version tag and integrity token. `am_i_primary()` replaced with real election probing (non-blocking connect + poll per candidate).

**Tech Stack:** C99, OpenSSL SHA-256, POSIX threads (non-blocking connect + poll), Cap'n Proto codecgen, Unity test framework, Meson build.

---

## File Map

| File | Action | What changes |
|------|--------|--------------|
| `src/schema/controller.capnp` | Modify | Add 3 opcodes + 6 structs + `configGeneration` field + `sender_priority` to Heartbeat |
| `src/include/myconf.h` | Modify | Add 6 fields to `myconf_t` |
| `src/include/xce/cluster_def.h` | Modify | Add 4 fields to `cluster_t` |
| `src/controller/ha.h` | Modify | Add `ssl_ctx`, `jitter_seed`, `standby_endpoints`, `election_priority_step_ms`, `election_jitter_max_ms` to `ha_state_t`; add `probe_ctx_t`; new API declarations |
| `src/include/persist.h` | Modify | Add `max_wal_records` to `persist_config_t`; add `wal_record_count`, `checkpoint_in_progress` to `persist_s` |
| `src/include/byteorder.h` | Create | `be64toh()` portability fallback |
| `src/controller/config_repl.h` | Create | Public API for config_repl.c |
| `src/controller/config_repl.c` | Create | All config storage, SHA-256, fetch, upgrade, push functions |
| `src/controller/main.c` | Modify | HEAD-file pointer resolution; add `--conf-keep-n`, `--conf-stale-timeout` flags; heap-alloc conffile |
| `src/include/myconf.h` | Modify | `myconf_fini()` declaration |
| `src/controller/myconf.c` | Modify | Add `load_primary_config()`, `myconf_fini()`; remove `get_default_conf()` static global |
| `src/controller/cluster.c` | Modify | Split `cluster_init()` into Phase 1 + Phase 2; add opcode dispatch for new ops; `cluster_run()` checks `restart_requested`; `handle_endpoint_error` prunes `standby_endpoints` |
| `src/controller/myapp.c` | Modify | Restructure to call Phase 1, `am_i_primary()`, config load/fetch, Phase 2; handle restart |
| `src/controller/ha.c` | Modify | Replace `am_i_primary()` stub with real election probing; add `probe_thread`, heartbeat listener, `ha_check_heartbeat_nocluster` priority comparison |
| `src/controller/join.c` | Modify | Implement `handle_controller_join()`; register in `standby_endpoints` |
| `src/include/typemap.h` | Modify | Add new union arms to `xce_request_t` and `xce_reply_t` |
| `src/lib/request.c` | Modify | Add encode/decode shims for 6 new message types |
| `src/controller/persist.c` | Modify | Add `schedule_checkpoint()` triggered by `max_wal_records` |
| `src/controller/meson.build` | Modify | Add `config_repl.c`, compile-time defines |
| `src/controller/testsuite/test_config_repl.c` | Create | Unit tests for Phase 1 + Phase 2 config functions |
| `src/controller/testsuite/test_ha_role.c` | Modify | Extend with election probing tests |
| `src/controller/testsuite/controller_test.c` | Modify | Register new test suites |

---

## Task 1: Schema additions

**Files:**
- Modify: `src/schema/controller.capnp`

- [ ] **Step 1: Add new OPCode entries**

  In `controller.capnp`, after `replCatchUpDone @28;`, add:

  ```capnp
    replConfig      @29;
    replConfigPush  @30;
    upgradeConfig   @31;
  ```

- [ ] **Step 2: Add new message structs**

  After the `MasterCandidate` struct, add:

  ```capnp
  struct ReplConfigRequest $C.mapname("repl_config_req_t") {
    versionTag  @0: UInt64 $C.mapname("version_tag");
  }

  struct ReplConfigReply $C.mapname("repl_config_reply_t") {
    upToDate    @0: Bool   $C.mapname("up_to_date");
    versionTag  @1: UInt64 $C.mapname("version_tag");
    sha256      @2: Data   $C.mapname("sha256");
    bytes       @3: Data   $C.mapname("bytes");
  }

  struct ReplConfigPushRequest $C.mapname("repl_config_push_req_t") {
    versionTag  @0: UInt64 $C.mapname("version_tag");
    sha256      @1: Data   $C.mapname("sha256");
    bytes       @2: Data   $C.mapname("bytes");
  }

  struct ReplConfigPushReply $C.mapname("repl_config_push_reply_t") {
    accepted    @0: Bool   $C.mapname("accepted");
    versionTag  @1: UInt64 $C.mapname("version_tag");
  }

  struct UpgradeConfigRequest $C.mapname("upgrade_config_req_t") {
    bytes   @0: Data   $C.mapname("bytes");
    sha256  @1: Data   $C.mapname("sha256");
  }

  struct UpgradeConfigReply $C.mapname("upgrade_config_reply_t") {
    versionTag  @0: UInt64 $C.mapname("version_tag");
    error       @1: Text   $C.mapname("error");
    ackedCount  @2: UInt32 $C.mapname("acked_count");
  }
  ```

- [ ] **Step 3: Add union arms to Request.body**

  In `struct Request`, after `hostctrlmsg @21: HostControl;`:

  ```capnp
      replConfigRequestMsg     @22: ReplConfigRequest;
      replConfigPushRequestMsg @23: ReplConfigPushRequest;
      upgradeConfigRequestMsg  @24: UpgradeConfigRequest;
  ```

- [ ] **Step 4: Add union arms to RequestReply.data**

  In `struct RequestReply`, after `clusterconf @7: ClusterConf;`:

  ```capnp
      replConfigReplyMsg     @8: ReplConfigReply;
      replConfigPushReplyMsg @9: ReplConfigPushReply;
      upgradeConfigReplyMsg  @10: UpgradeConfigReply;
  ```

- [ ] **Step 5: Add configGeneration to ClusterConfig**

  In `struct ClusterConfig`, after the last existing field, add:

  ```capnp
    configGeneration @20: UInt64 $C.mapname("config_generation");
  ```

- [ ] **Step 6: Add sender_priority to a new Heartbeat struct**

  Add a new struct (the schema currently has no Heartbeat struct — `OP_heartbeat` uses the nop/void path):

  ```capnp
  struct HeartbeatMsg $C.mapname("heartbeat_msg_t") {
    senderPriority @0: UInt32 $C.mapname("sender_priority");
    term           @1: UInt64 $C.mapname("term");
  }
  ```

  Then in `Request.body`, add:

  ```capnp
      heartbeatmsg @25: HeartbeatMsg;
  ```

- [ ] **Step 7: Build to verify schema compiles**

  ```bash
  cd /worktrees/persistent_ha/src && nix develop --command bash -c "meson setup build --reconfigure 2>&1 | tail -20 && ninja -C build src/schema/controller.capnp.h 2>&1 | tail -20"
  ```

  Expected: no capnp parse errors, generated header updated.

- [ ] **Step 8: Commit**

  ```bash
  git add src/schema/controller.capnp
  git commit -m "feat: add replConfig/replConfigPush/upgradeConfig opcodes and message structs to schema"
  ```

---

## Task 2: Struct field additions

**Files:**
- Modify: `src/include/myconf.h`
- Modify: `src/include/xce/cluster_def.h`
- Modify: `src/controller/ha.h`
- Modify: `src/include/persist.h`
- Create: `src/include/byteorder.h`

- [ ] **Step 1: Add fields to myconf_t**

  In `src/include/myconf.h`, after `xce_cluster_config_t peers_cfg;`:

  ```c
      uint64_t             config_version_tag;   /* first 8 bytes of SHA-256, host order */
      uint8_t              config_sha256[32];     /* full SHA-256 of conffile bytes */
      int                  conf_keep_n;           /* versioned file retention count (default 2) */
      uint64_t             config_generation;     /* monotonic audit counter (§4.4) */
      time_t               conf_stale_timeout;    /* seconds before saved config is flagged stale (default 86400) */
      int                  config_stale;          /* set when saved config is older than conf_stale_timeout */
  ```

  Also add `myconf_fini()` declaration at the bottom of the file:

  ```c
  void myconf_fini(myconf_t *myconf);
  ```

- [ ] **Step 2: Add fields to cluster_t**

  In `src/include/xce/cluster_def.h`, inside `struct cluster_s`, after `ha_state_t ha;`:

  ```c
      _Atomic int         restart_requested;        /* set by OP_upgradeConfig handler */
      pthread_t           heartbeat_listener_tid;   /* pre-Phase-2 SSL listener thread */
      _Atomic int         heartbeat_listener_running;
      _Atomic int         lost_and_found_job_count; /* incremented during persist_recover() */
  ```

- [ ] **Step 3: Add fields to ha_state_t and ha_config_t**

  In `src/controller/ha.h`:

  (a) In `ha_config_t`, after `max_promotion_retries`:
  ```c
      uint32_t election_jitter_max_ms;   /* random jitter added to election timeout */
      uint32_t election_priority_step_ms; /* extra ms per priority level */
  ```

  (b) Add new default macros after the existing defaults:
  ```c
  #define HA_DEFAULT_ELECTION_JITTER_MAX_MS   200
  #define HA_DEFAULT_ELECTION_PRIORITY_STEP_MS 500
  ```

  (c) In `ha_state_t`, after `watchdog_thread`:
  ```c
      void            *ssl_ctx;          /* SSL_CTX* cast; avoids pulling in openssl headers */
      uint64_t         jitter_seed;      /* xorshift64 PRNG state for election jitter */
      zarray_t        *standby_endpoints; /* connected standby controller endpoints */
  ```

  (d) Add `probe_ctx_t` struct declaration before `ha_state_t`:
  ```c
  typedef struct {
      struct xce_candidate_s *candidate; /* peer to probe */
      void                   *ssl_ctx;   /* SSL_CTX* (void* to avoid openssl header) */
      uint32_t                timeout_ms;
      atomic_int             *election_done;
      int                     got_ack;
      pthread_t               thread;
  } probe_ctx_t;
  ```

  (e) Add new API declarations at the bottom:
  ```c
  int  ha_is_primary(const ha_state_t *ha);
  void start_heartbeat_listener(cluster_t *c);
  void stop_heartbeat_listener(cluster_t *c);
  ```

- [ ] **Step 4: Add fields to persist_config_t and persist_s**

  In `src/include/persist.h`, in `persist_config_t` after `sync_timeout_ms`:
  ```c
      uint32_t max_wal_records;          /* checkpoint trigger: records since last snap (default 10000) */
  ```

  In `persist_config_defaults()`, add:
  ```c
      c.max_wal_records = 10000;
  ```

  In `struct persist_s` after `pthread_mutex_t lock;`:
  ```c
      _Atomic uint32_t  wal_record_count;      /* records since last checkpoint */
      _Atomic int       checkpoint_in_progress; /* CAS guard for schedule_checkpoint */
  ```

- [ ] **Step 5: Create byteorder.h**

  Create `src/include/byteorder.h`:

  ```c
  #pragma once
  /* Portable be64toh() for non-glibc or pre-2.25 targets.
   * On Linux glibc >= 2.9 this is already available from <endian.h>.
   * On other platforms we provide a fallback. */
  #if defined(__linux__)
  #  include <endian.h>
  #elif defined(__APPLE__)
  #  include <machine/endian.h>
  #  include <libkern/OSByteOrder.h>
  #  define be64toh(x) OSSwapBigToHostInt64(x)
  #else
  #  include <stdint.h>
  static inline uint64_t be64toh(uint64_t x) {
  #  if __BYTE_ORDER__ == __ORDER_LITTLE_ENDIAN__
      return __builtin_bswap64(x);
  #  else
      return x;
  #  endif
  }
  #endif
  ```

- [ ] **Step 6: Build to verify all struct changes compile**

  ```bash
  cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build 2>&1 | grep -E 'error:|warning:|ninja:' | head -30"
  ```

  Expected: only warnings about new fields being unused — no errors.

- [ ] **Step 7: Commit**

  ```bash
  git add src/include/myconf.h src/include/xce/cluster_def.h src/controller/ha.h src/include/persist.h src/include/byteorder.h
  git commit -m "feat: add config replication fields to myconf_t, cluster_t, ha_state_t, persist_t"
  ```

---

## Task 3: config_repl.h + config_repl.c (Phase 1 storage functions)

**Files:**
- Create: `src/controller/config_repl.h`
- Create: `src/controller/config_repl.c` (Phase 1 functions only)
- Create: `src/controller/testsuite/test_config_repl.c` (Phase 1 tests)

- [ ] **Step 1: Write failing tests for Phase 1 storage functions**

  Create `src/controller/testsuite/test_config_repl.c`:

  ```c
  /* test_config_repl.c — Unity tests for config replication (Phase 1) */
  #include "unity.h"
  #include <fcntl.h>
  #include <stdio.h>
  #include <stdlib.h>
  #include <string.h>
  #include <sys/stat.h>
  #include <unistd.h>
  #include "config_repl.h"

  static char tmpdir[256];

  void setUp(void) {
      snprintf(tmpdir, sizeof(tmpdir), "/tmp/test_config_repl_%d", (int)getpid());
      mkdir(tmpdir, 0700);
      char conf_dir[300];
      snprintf(conf_dir, sizeof(conf_dir), "%s/conf", tmpdir);
      mkdir(conf_dir, 0700);
  }

  void tearDown(void) {
      char cmd[300];
      snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir);
      system(cmd);
  }

  /* test: conf_read_current_filename() returns -ENOENT when conf/current absent */
  void test_conf_read_current_filename_missing(void) {
      char out[256] = {0};
      int rc = conf_read_current_filename(tmpdir, out, sizeof(out));
      TEST_ASSERT_EQUAL_INT(-ENOENT, rc);
  }

  /* test: conf_save_config() writes versioned file + HEAD pointer; read back matches */
  void test_conf_save_and_read_current(void) {
      const uint8_t data[] = "fake capnproto config blob";
      size_t len = sizeof(data) - 1;
      uint8_t sha256[32];
      SHA256(data, len, sha256);

      myconf_t myconf = {0};
      myconf.workdir  = tmpdir;
      myconf.name     = "ctrl0";
      myconf.conf_keep_n = 2;

      int rc = conf_save_config(&myconf, data, len, sha256);
      TEST_ASSERT_EQUAL_INT(0, rc);

      /* conf/current pointer file must exist */
      char ptr_path[512];
      snprintf(ptr_path, sizeof(ptr_path), "%s/conf/current", tmpdir);
      TEST_ASSERT_EQUAL_INT(0, access(ptr_path, F_OK));

      /* Reading back the current filename must succeed */
      char out[256] = {0};
      rc = conf_read_current_filename(tmpdir, out, sizeof(out));
      TEST_ASSERT_EQUAL_INT(0, rc);
      TEST_ASSERT_NOT_NULL(strstr(out, "cluster.bin."));
  }

  /* test: conf_save_config() then conf_read_current_filename() parses 12-hex suffix */
  void test_conf_current_filename_format(void) {
      const uint8_t data[] = "another fake blob for hex test";
      size_t len = sizeof(data) - 1;
      uint8_t sha256[32];
      SHA256(data, len, sha256);

      myconf_t myconf = {0};
      myconf.workdir  = tmpdir;
      myconf.name     = "ctrl0";
      myconf.conf_keep_n = 2;
      conf_save_config(&myconf, data, len, sha256);

      char out[256] = {0};
      conf_read_current_filename(tmpdir, out, sizeof(out));

      /* Must start with "cluster.bin." */
      TEST_ASSERT_EQUAL_INT(0, strncmp(out, "cluster.bin.", 12));
      /* Suffix must be exactly 12 hex chars */
      const char *hex = out + 12;
      TEST_ASSERT_EQUAL_INT(12, (int)strlen(hex));
      TEST_ASSERT_EQUAL_INT(12, (int)strspn(hex, "0123456789abcdef"));
  }

  /* test: conf_trim_old_versions() keeps only conf_keep_n files */
  void test_conf_trim_keep_n(void) {
      myconf_t myconf = {0};
      myconf.workdir  = tmpdir;
      myconf.name     = "ctrl0";
      myconf.conf_keep_n = 2;
      uint8_t sha256[32];

      /* Write 3 different config versions */
      for (int i = 0; i < 3; i++) {
          char data[32];
          snprintf(data, sizeof(data), "config version %d", i);
          SHA256((uint8_t*)data, strlen(data), sha256);
          usleep(10000); /* ensure distinct mtime */
          conf_save_config(&myconf, (uint8_t*)data, strlen(data), sha256);
      }

      /* Count cluster.bin.* files (excluding .tmp) */
      char conf_dir[300];
      snprintf(conf_dir, sizeof(conf_dir), "%s/conf", tmpdir);
      int count = 0;
      DIR *d = opendir(conf_dir);
      struct dirent *e;
      while ((e = readdir(d)) != NULL) {
          if (strncmp(e->d_name, "cluster.bin.", 12) == 0 &&
              strstr(e->d_name, ".tmp") == NULL)
              count++;
      }
      closedir(d);
      TEST_ASSERT_EQUAL_INT(2, count);
  }

  /* test: zero-tag sentinel — standby with no config sends tag=0 */
  void test_conf_zero_tag_is_no_config_sentinel(void) {
      /* When conf/current is absent, compute_config_tag() returns 0 */
      uint64_t tag = conf_get_saved_tag(tmpdir);
      TEST_ASSERT_EQUAL_UINT64(0, tag);
  }

  /* test: conf_read_current_filename() returns -EINVAL on path separator in pointer */
  void test_conf_current_filename_rejects_path_separator(void) {
      char ptr_path[300];
      snprintf(ptr_path, sizeof(ptr_path), "%s/conf/current", tmpdir);
      FILE *f = fopen(ptr_path, "w");
      fprintf(f, "cluster.bin/../evil\n");
      fclose(f);

      char out[256] = {0};
      int rc = conf_read_current_filename(tmpdir, out, sizeof(out));
      TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
  }

  /* test: empty conf/current treated as ENODATA */
  void test_conf_empty_pointer_file_returns_enodata(void) {
      char ptr_path[300];
      snprintf(ptr_path, sizeof(ptr_path), "%s/conf/current", tmpdir);
      /* Create empty file */
      int fd = open(ptr_path, O_WRONLY|O_CREAT|O_TRUNC, 0600);
      close(fd);

      char out[256] = {0};
      int rc = conf_read_current_filename(tmpdir, out, sizeof(out));
      TEST_ASSERT_EQUAL_INT(-ENODATA, rc);
  }

  /* test: conf_check_stale() returns 0 for fresh file, 1 for old file */
  void test_conf_stale_timeout(void) {
      /* Write a config */
      const uint8_t data[] = "stale test blob";
      uint8_t sha256[32];
      SHA256(data, sizeof(data)-1, sha256);

      myconf_t myconf = {0};
      myconf.workdir  = tmpdir;
      myconf.name     = "ctrl0";
      myconf.conf_keep_n = 2;
      myconf.conf_stale_timeout = 86400; /* 24h */
      conf_save_config(&myconf, data, sizeof(data)-1, sha256);

      /* Freshly written — not stale */
      int stale = conf_check_stale(tmpdir, 86400);
      TEST_ASSERT_EQUAL_INT(0, stale);

      /* Timeout of 0 means disabled — never stale */
      stale = conf_check_stale(tmpdir, 0);
      TEST_ASSERT_EQUAL_INT(0, stale);

      /* 1-second timeout — touch the file to be 2s old via utime */
      struct utimbuf old_time;
      old_time.actime = time(NULL) - 10;
      old_time.modtime = time(NULL) - 10;
      char ver_path[512];
      /* resolve the versioned filename */
      char fname[256] = {0};
      conf_read_current_filename(tmpdir, fname, sizeof(fname));
      snprintf(ver_path, sizeof(ver_path), "%s/conf/%s", tmpdir, fname);
      utime(ver_path, &old_time);

      stale = conf_check_stale(tmpdir, 5); /* 5-second timeout */
      TEST_ASSERT_EQUAL_INT(1, stale);
  }
  ```

- [ ] **Step 2: Run tests to verify they fail (functions not yet defined)**

  ```bash
  cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build controller_unit_test 2>&1 | grep -E 'error:|undefined' | head -20"
  ```

  Expected: linker errors for `conf_save_config`, `conf_read_current_filename`, etc.

- [ ] **Step 3: Create config_repl.h**

  Create `src/controller/config_repl.h`:

  ```c
  #pragma once
  /* config_repl.h — config storage, SHA-256, replication (Phase 1 & 2) */
  #include <stdint.h>
  #include <stddef.h>
  #include "myconf.h"

  /* Max config blob size — reject oversized files before allocating */
  #define CONF_MAX_SIZE (1 * 1024 * 1024)  /* 1 MiB */

  /* Compile-time gate for OP_upgradeConfig (Phase 2, default disabled) */
  #ifndef CONF_ENABLE_UPGRADE_CONFIG
  #define CONF_ENABLE_UPGRADE_CONFIG 0
  #endif

  /* Push timeout for fan-out to standbys (ms) */
  #ifndef CONF_PUSH_TIMEOUT_MS
  #define CONF_PUSH_TIMEOUT_MS 500
  #endif

  /* Days before startup cert-expiry warning is logged */
  #define CERT_WARN_DAYS 30

  /* ── Phase 1: Storage ────────────────────────────────── */

  /*
   * conf_read_current_filename: read HEAD-style pointer file.
   * Reads <workdir>/conf/current, strips newline, validates format.
   * Returns 0 on success; -ENOENT if missing; -ENODATA if empty;
   * -EINVAL if malformed (path separator, wrong prefix, wrong hex length).
   * out must be at least NAME_MAX bytes.
   */
  int conf_read_current_filename(const char *workdir, char *out, size_t outsz);

  /*
   * conf_save_config: write versioned file + update HEAD pointer atomically.
   * Steps: write <sha_hex>.tmp → fsync → rename to cluster.bin.<sha_hex>,
   * then write current.tmp → fsync → rename to conf/current.
   * Trims old versions after success.
   * Returns 0 on success, -errno on failure.
   */
  int conf_save_config(myconf_t *myconf, const uint8_t *bytes, size_t len,
                       const uint8_t sha256[32]);

  /*
   * conf_read_raw: read the versioned file named by conf/current.
   * Allocates *bytes (caller must free). Enforces CONF_MAX_SIZE.
   * Returns 0 on success; -EFBIG if over limit; -errno on I/O error.
   */
  int conf_read_raw(const char *conffile, uint8_t **bytes, size_t *len);

  /*
   * conf_get_saved_tag: return first 8 bytes of SHA-256 of the current
   * versioned file, host-byte-order. Returns 0 if conf/current is absent
   * or unreadable (zero-tag sentinel for "no config saved").
   */
  uint64_t conf_get_saved_tag(const char *workdir);

  /*
   * conf_check_stale: returns 1 if the current versioned file's mtime is
   * older than stale_timeout_s seconds. Returns 0 if within timeout or
   * if stale_timeout_s == 0 (disabled). Returns 0 if conf/current absent.
   */
  int conf_check_stale(const char *workdir, time_t stale_timeout_s);

  /*
   * conf_trim_old_versions: keep at most keep_n versioned cluster.bin.* files
   * (by mtime, newest first). Also removes stale .tmp files.
   * Always preserves the file named by the current conf/current pointer.
   */
  void conf_trim_old_versions(const char *workdir, int keep_n);

  /*
   * config_contains_reserved_name: returns 1 if ClusterConfig contains
   * a queue or host named "lost_and_found". Used at load time.
   */
  int config_contains_reserved_name(cluster_config_t *cfg);

  /*
   * sanitise_cn: copy at most maxlen printable ASCII chars from src to dst,
   * replacing non-printable or control chars with '?'. Returns dst.
   */
  char *sanitise_cn(const char *src, size_t maxlen);
  ```

- [ ] **Step 4: Implement Phase 1 functions in config_repl.c**

  Create `src/controller/config_repl.c` with the Phase 1 functions:

  ```c
  #define _GNU_SOURCE
  #include "config_repl.h"

  #include <dirent.h>
  #include <errno.h>
  #include <fcntl.h>
  #include <openssl/sha.h>
  #include <stdio.h>
  #include <stdlib.h>
  #include <string.h>
  #include <sys/stat.h>
  #include <sys/types.h>
  #include <time.h>
  #include <utime.h>
  #include <unistd.h>
  #include <pthread.h>

  #include "byteorder.h"
  #include "clusterconfig.h"
  #include "controller.capnp.h"
  #include "tlog.h"

  /* ── helpers ──────────────────────────────────────────── */

  static void make_conf_dir(const char *workdir, char *out, size_t sz)
  {
      snprintf(out, sz, "%s/conf", workdir);
  }

  static void bytes_to_hex(const uint8_t *b, size_t n, char *out)
  {
      static const char hex[] = "0123456789abcdef";
      for (size_t i = 0; i < n; i++) {
          out[2*i]   = hex[(b[i] >> 4) & 0xf];
          out[2*i+1] = hex[b[i] & 0xf];
      }
      out[2*n] = '\0';
  }

  /* ── conf_read_current_filename ─────────────────────── */

  int conf_read_current_filename(const char *workdir, char *out, size_t outsz)
  {
      char head_path[4096];
      snprintf(head_path, sizeof(head_path), "%s/conf/current", workdir);

      FILE *f = fopen(head_path, "r");
      if (!f)
          return -ENOENT;

      struct stat st;
      if (fstat(fileno(f), &st) == 0 && st.st_size == 0) {
          fclose(f);
          ERROR("conf/current is empty — treating as missing config");
          return -ENODATA;
      }

      if (!fgets(out, (int)outsz, f)) {
          fclose(f);
          return -EIO;
      }
      fclose(f);
      out[strcspn(out, "\n")] = '\0';

      if (strchr(out, '/')) {
          ERROR("conf/current contains path separator — corrupted pointer file");
          return -EINVAL;
      }
      if (strncmp(out, "cluster.bin.", 12) != 0) {
          ERROR("conf/current filename '%s' does not start with 'cluster.bin.'", out);
          return -EINVAL;
      }
      const char *hex = out + 12;
      if (strlen(hex) != 12 || strspn(hex, "0123456789abcdef") != 12) {
          ERROR("conf/current suffix '%s' is not 12 hex chars", hex);
          return -EINVAL;
      }
      return 0;
  }

  /* ── conf_save_config ───────────────────────────────── */

  int conf_save_config(myconf_t *myconf, const uint8_t *bytes, size_t len,
                       const uint8_t sha256[32])
  {
      char conf_dir[4096];
      make_conf_dir(myconf->workdir, conf_dir, sizeof(conf_dir));

      if (mkdir(conf_dir, 0700) != 0 && errno != EEXIST)
          return -errno;

      /* Derive 12-char hex tag from first 6 bytes of SHA-256 */
      char hex_tag[13];
      bytes_to_hex(sha256, 6, hex_tag);

      /* Step 1: write versioned file via thread-unique .tmp */
      char tmp_path[4096], ver_path[4096];
      snprintf(tmp_path, sizeof(tmp_path), "%s/cluster.bin.%s.%lu.tmp",
               conf_dir, hex_tag, (unsigned long)pthread_self());
      snprintf(ver_path, sizeof(ver_path), "%s/cluster.bin.%s",
               conf_dir, hex_tag);

      int fd = open(tmp_path, O_WRONLY|O_CREAT|O_TRUNC, 0600);
      if (fd < 0)
          return -errno;
      if (write(fd, bytes, len) != (ssize_t)len) {
          close(fd); unlink(tmp_path);
          return -EIO;
      }
      if (fsync(fd) != 0) {
          close(fd); unlink(tmp_path);
          return -EIO;
      }
      close(fd);
      if (rename(tmp_path, ver_path) != 0) {
          unlink(tmp_path);
          return -errno;
      }

      /* Step 2: update HEAD pointer file via thread-unique .tmp */
      char ptr_tmp[4096], ptr_path[4096];
      snprintf(ptr_tmp,  sizeof(ptr_tmp),  "%s/current.%lu.tmp",
               conf_dir, (unsigned long)pthread_self());
      snprintf(ptr_path, sizeof(ptr_path), "%s/current", conf_dir);

      char target[NAME_MAX];
      snprintf(target, sizeof(target), "cluster.bin.%s\n", hex_tag);

      fd = open(ptr_tmp, O_WRONLY|O_CREAT|O_TRUNC, 0600);
      if (fd < 0)
          return -errno;
      write(fd, target, strlen(target));
      fsync(fd);
      close(fd);
      if (rename(ptr_tmp, ptr_path) != 0) {
          unlink(ptr_tmp);
          return -errno;
      }

      /* Step 2.5: fsync the conf/ directory to persist the renames */
      fd = open(conf_dir, O_RDONLY);
      if (fd >= 0) { fsync(fd); close(fd); }

      /* Step 3: trim old versions */
      conf_trim_old_versions(myconf->workdir, myconf->conf_keep_n);

      /* Update conffile to point to new versioned file */
      free(myconf->conffile);
      asprintf(&myconf->conffile, "%s/cluster.bin.%s", conf_dir, hex_tag);

      /* Update tag and SHA-256 in myconf */
      uint64_t new_tag;
      memcpy(&new_tag, sha256, 8);
      myconf->config_version_tag = be64toh(new_tag);
      memcpy(myconf->config_sha256, sha256, 32);

      return 0;
  }

  /* ── conf_read_raw ──────────────────────────────────── */

  int conf_read_raw(const char *conffile, uint8_t **bytes, size_t *len)
  {
      FILE *f = fopen(conffile, "rb");
      if (!f)
          return -errno;

      fseek(f, 0, SEEK_END);
      long sz = ftell(f);
      fseek(f, 0, SEEK_SET);

      if (sz <= 0) { fclose(f); return -EIO; }
      if ((size_t)sz > CONF_MAX_SIZE) { fclose(f); return -EFBIG; }

      *bytes = (uint8_t*)malloc((size_t)sz);
      if (!*bytes) { fclose(f); return -ENOMEM; }

      if (fread(*bytes, 1, (size_t)sz, f) != (size_t)sz) {
          free(*bytes); *bytes = NULL; fclose(f); return -EIO;
      }
      fclose(f);
      *len = (size_t)sz;
      return 0;
  }

  /* ── conf_get_saved_tag ─────────────────────────────── */

  uint64_t conf_get_saved_tag(const char *workdir)
  {
      char fname[256] = {0};
      if (conf_read_current_filename(workdir, fname, sizeof(fname)) != 0)
          return 0;

      char conf_dir[4096];
      make_conf_dir(workdir, conf_dir, sizeof(conf_dir));

      char ver_path[4096];
      snprintf(ver_path, sizeof(ver_path), "%s/%s", conf_dir, fname);

      uint8_t *bytes = NULL;
      size_t   len   = 0;
      if (conf_read_raw(ver_path, &bytes, &len) != 0)
          return 0;

      uint8_t sha256[32];
      SHA256(bytes, len, sha256);
      free(bytes);

      uint64_t tag;
      memcpy(&tag, sha256, 8);
      return be64toh(tag);
  }

  /* ── conf_check_stale ───────────────────────────────── */

  int conf_check_stale(const char *workdir, time_t stale_timeout_s)
  {
      if (stale_timeout_s == 0)
          return 0;

      char fname[256] = {0};
      if (conf_read_current_filename(workdir, fname, sizeof(fname)) != 0)
          return 0;

      char ver_path[4096];
      snprintf(ver_path, sizeof(ver_path), "%s/conf/%s", workdir, fname);

      struct stat st;
      if (stat(ver_path, &st) != 0)
          return 0;

      time_t age = time(NULL) - st.st_mtime;
      return (age > stale_timeout_s) ? 1 : 0;
  }

  /* ── conf_trim_old_versions ─────────────────────────── */

  void conf_trim_old_versions(const char *workdir, int keep_n)
  {
      char conf_dir[4096];
      make_conf_dir(workdir, conf_dir, sizeof(conf_dir));

      /* Read current pointer to protect it from deletion */
      char current_name[256] = {0};
      conf_read_current_filename(workdir, current_name, sizeof(current_name));

      /* Collect cluster.bin.* files (not .tmp) sorted by mtime */
      struct entry { char name[NAME_MAX]; time_t mtime; };
      struct entry entries[256];
      int n = 0;

      DIR *d = opendir(conf_dir);
      if (!d) return;
      struct dirent *e;
      while ((e = readdir(d)) != NULL && n < 256) {
          if (strncmp(e->d_name, "cluster.bin.", 12) != 0)
              continue;
          if (strstr(e->d_name, ".tmp") != NULL) {
              /* Clean stale .tmp files */
              char full[4096];
              snprintf(full, sizeof(full), "%s/%s", conf_dir, e->d_name);
              unlink(full);
              continue;
          }
          char full[4096];
          snprintf(full, sizeof(full), "%s/%s", conf_dir, e->d_name);
          struct stat st;
          if (stat(full, &st) == 0) {
              strncpy(entries[n].name, e->d_name, NAME_MAX-1);
              entries[n].mtime = st.st_mtime;
              n++;
          }
      }
      closedir(d);

      /* Sort descending by mtime (newest first) — simple insertion sort */
      for (int i = 1; i < n; i++) {
          struct entry tmp = entries[i];
          int j = i - 1;
          while (j >= 0 && entries[j].mtime < tmp.mtime) {
              entries[j+1] = entries[j]; j--;
          }
          entries[j+1] = tmp;
      }

      /* Delete everything after keep_n, skipping the current pointer */
      for (int i = keep_n; i < n; i++) {
          if (strcmp(entries[i].name, current_name) == 0)
              continue; /* never delete the active version */
          char full[4096];
          snprintf(full, sizeof(full), "%s/%s", conf_dir, entries[i].name);
          unlink(full);
      }
  }

  /* ── config_contains_reserved_name ─────────────────── */

  int config_contains_reserved_name(cluster_config_t *cfg)
  {
      if (!cfg) return 0;
      /* Check queues */
      for (int i = 0; i < cfg->nqueues; i++) {
          if (cfg->queues[i] && strcmp(cfg->queues[i]->name, "lost_and_found") == 0)
              return 1;
      }
      /* Check hosts / candidates */
      for (int i = 0; i < cfg->nhosts; i++) {
          if (cfg->hosts[i] && strcmp(cfg->hosts[i]->name, "lost_and_found") == 0)
              return 1;
      }
      return 0;
  }

  /* ── sanitise_cn ────────────────────────────────────── */

  char *sanitise_cn(const char *src, size_t maxlen)
  {
      static char buf[128];
      size_t n = maxlen < sizeof(buf)-1 ? maxlen : sizeof(buf)-1;
      size_t i;
      for (i = 0; i < n && src[i] != '\0'; i++)
          buf[i] = (src[i] >= 0x20 && src[i] < 0x7f) ? src[i] : '?';
      buf[i] = '\0';
      return buf;
  }
  ```

- [ ] **Step 5: Add config_repl.c to meson.build and update test suite**

  In `src/controller/meson.build`, add `'config_repl.c'` to `controller_src`.

  Also add to `controller_unit_test_src`:
  ```
  'testsuite/test_config_repl.c',
  'config_repl.c',
  ```

  Add compile definition to `controller_unit_test`:
  ```
  c_args: ['-DXCE_TESTING', '-DCONF_ENABLE_UPGRADE_CONFIG=0'],
  ```

- [ ] **Step 6: Run tests to verify they pass**

  ```bash
  cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build controller_unit_test && ./build/controller_unit_test 2>&1 | grep -E 'PASS|FAIL|test_conf'"
  ```

  Expected: all `test_conf_*` tests PASS.

- [ ] **Step 7: Commit**

  ```bash
  git add src/controller/config_repl.h src/controller/config_repl.c src/controller/testsuite/test_config_repl.c src/controller/meson.build
  git commit -m "feat: add config_repl Phase 1 — HEAD-file storage, SHA-256, trim, stale check"
  ```

---

## Task 4: main.c — HEAD-file config resolution + new CLI flags

**Files:**
- Modify: `src/controller/main.c`

- [ ] **Step 1: Replace flat conffile path with HEAD-file resolution**

  In `main()`, replace the `conf_file` stack buffer and assignment:

  ```c
  /* Old code (lines ~79, 115–116, 121): */
  char conf_file[XCE_MAXPATH_LEN + 1];
  snprintf(conf_file, ..., "%s/%s/conf/current_config", ...);
  myconf.conffile = conf_file;

  /* New code: conffile is heap-allocated by load_primary_config() or
   * fetch_config_from_primary() — do not set it here.
   * Store workdir for use in config_repl.c path construction. */
  /* Remove the conf_file declaration and the myconf.conffile = conf_file line */
  myconf.conffile = NULL;   /* will be set by load_primary_config() or fetch */
  ```

- [ ] **Step 2: Add --conf-keep-n and --conf-stale-timeout CLI options**

  Add two new option keys and entries to `options[]`:

  ```c
  #define OPT_CONF_KEEP_N       257
  #define OPT_CONF_STALE_TIMEOUT 258
  ```

  ```c
  { "conf-keep-n",       OPT_CONF_KEEP_N,       "N",   0,
    "Number of versioned config files to retain (default 2)" },
  { "conf-stale-timeout", OPT_CONF_STALE_TIMEOUT, "SECS", 0,
    "Seconds before saved config is considered stale (0=disabled, default 86400)" },
  ```

  Add to `struct arguments`:
  ```c
  int   conf_keep_n;
  long  conf_stale_timeout;
  ```

  Add cases to `parse_opt`:
  ```c
  case OPT_CONF_KEEP_N:
      arguments->conf_keep_n = atoi(arg);
      break;
  case OPT_CONF_STALE_TIMEOUT:
      arguments->conf_stale_timeout = atol(arg);
      break;
  ```

  After `memset(&arguments, ...)`, initialize defaults:
  ```c
  arguments.conf_keep_n = 2;
  arguments.conf_stale_timeout = 86400;
  ```

  Set in myconf after parsing:
  ```c
  myconf.conf_keep_n = arguments.conf_keep_n;
  myconf.conf_stale_timeout = (time_t)arguments.conf_stale_timeout;
  ```

- [ ] **Step 3: Add --peers validation WARN**

  After the `xce_parse_peers()` call, add:

  ```c
  for (int i = 0; i < myconf.peers_cfg.candidates_count; i++) {
      if (myconf.peers_cfg.candidates[i].hostname[0] == '\0' ||
          myconf.peers_cfg.candidates[i].port == 0) {
          WARN("--peers: candidate %d has empty host or port 0 — check peer string",
               i);
      }
  }
  ```

- [ ] **Step 4: Build**

  ```bash
  cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build xce_controller 2>&1 | grep -E 'error:|warning:' | head -20"
  ```

  Expected: no errors.

- [ ] **Step 5: Commit**

  ```bash
  git add src/controller/main.c
  git commit -m "feat: update main.c — heap-alloc conffile, add --conf-keep-n/--conf-stale-timeout flags"
  ```

---

## Task 5: myconf.c — load_primary_config() and myconf_fini()

**Files:**
- Modify: `src/controller/myconf.c`

- [ ] **Step 1: Write failing test for load_primary_config() missing-file path**

  In `src/controller/testsuite/test_config_repl.c`, add:

  ```c
  #include "myconf.h"
  /* ... */

  void test_primary_missing_config_fatal_path(void) {
      myconf_t myconf = {0};
      myconf.workdir  = tmpdir;
      myconf.name     = "ctrl0";
      myconf.conffile = NULL;
      myconf.conf_keep_n = 2;
      /* No conf/current file exists — load_primary_config must return -ENOENT */
      int rc = load_primary_config_for_test(&myconf);
      TEST_ASSERT_EQUAL_INT(-ENOENT, rc);
  }
  ```

  (We'll expose a `_for_test` variant that returns the error code instead of calling FATAL.)

- [ ] **Step 2: Add load_primary_config() to myconf.c**

  In `src/controller/myconf.c`, add:

  ```c
  #include "config_repl.h"
  #include "byteorder.h"
  #include <openssl/sha.h>
  #include <stdatomic.h>
  ```

  Add the function:

  ```c
  int load_primary_config(cluster_t *c)
  {
      myconf_t *myconf = c->myconf;
      char conf_dir[4096];
      snprintf(conf_dir, sizeof(conf_dir), "%s/%s", myconf->workdir, myconf->name);

      /* Resolve conf/current → versioned filename */
      char fname[256] = {0};
      int rc = conf_read_current_filename(conf_dir, fname, sizeof(fname));
      if (rc != 0) {
          FATAL("Primary has no config file '%s/conf/current' (rc=%d) — "
                "provision the config before starting the primary controller",
                conf_dir, rc);
          return -ENOENT;
      }

      /* Heap-allocate conffile so it survives the stack frame */
      free(myconf->conffile);
      if (asprintf(&myconf->conffile, "%s/conf/%s", conf_dir, fname) < 0)
          return -ENOMEM;

      /* Check for reserved lost_and_found name */
      struct capn       cap;
      ClusterConfig_ptr p;
      FILE *fp = fopen(myconf->conffile, "rb");
      if (!fp) {
          FATAL("Primary config file '%s' not readable: %s",
                myconf->conffile, strerror(errno));
          return -errno;
      }
      capn_init_fp(&cap, fp, 0);
      p.p = capn_getp(capn_root(&cap), 0, 1);
      cluster_config_t *cfg = NULL;
      decode_ClusterConfig_ptr(&cfg, p);
      capn_free(&cap);
      fclose(fp);

      if (!cfg) {
          FATAL("Primary config file '%s' is not a valid ClusterConfig blob",
                myconf->conffile);
          return -EINVAL;
      }
      if (config_contains_reserved_name(cfg)) {
          free_ClusterConfig(cfg);
          FATAL("Config file '%s' contains reserved name 'lost_and_found'",
                myconf->conffile);
          return -EINVAL;
      }

      fill_default_cluster_config(cfg);
      myconf->cluster_config = cfg;

      /* Compute and cache SHA-256 */
      uint8_t *bytes = NULL;
      size_t   len   = 0;
      rc = conf_read_raw(myconf->conffile, &bytes, &len);
      if (rc != 0) {
          FATAL("Failed to read primary config for SHA-256: %d", rc);
          return rc;
      }
      SHA256(bytes, len, myconf->config_sha256);
      free(bytes);

      uint64_t tag;
      memcpy(&tag, myconf->config_sha256, 8);
      myconf->config_version_tag = be64toh(tag);

      /* Default generation to 1 if not set in blob */
      if (cfg->config_generation == 0)
          cfg->config_generation = 1;
      myconf->config_generation = cfg->config_generation;

      /* Check staleness */
      myconf->config_stale = conf_check_stale(conf_dir,
                                              myconf->conf_stale_timeout);
      if (myconf->config_stale)
          WARN("Primary config appears stale (mtime > %lds)", myconf->conf_stale_timeout);

      INFO("Primary config loaded: tag=%016"PRIx64" gen=%"PRIu64,
           myconf->config_version_tag, myconf->config_generation);
      return 0;
  }
  ```

- [ ] **Step 3: Add myconf_fini() to myconf.c**

  ```c
  void myconf_fini(myconf_t *myconf)
  {
      if (!myconf) return;
      free(myconf->conffile);
      myconf->conffile = NULL;
      if (myconf->cluster_config) {
          free_ClusterConfig(myconf->cluster_config);
          free(myconf->cluster_config);
          myconf->cluster_config = NULL;
      }
  }
  ```

  Remove the `static struct cluster_configuration *cf = NULL;` global and the `cluster_hook_fun` / `config_fini` functions that handle the shutdown hook for `cf` — `myconf_fini()` takes over this responsibility.

  Keep `load_cluster_config()` and `load_cluster_config_from_file()` for now but mark them as legacy (they're still used by the existing `get_default_conf()` path during Phase 1 work).

- [ ] **Step 4: Declare load_primary_config() in myconf.h**

  In `src/include/myconf.h`, add:

  ```c
  /* Declared here; defined in myconf.c; used by myapp.c */
  int  load_primary_config(struct cluster_s *c);
  ```

  (forward-declare `cluster_t` in myconf.h to avoid circular include):
  ```c
  typedef struct cluster_s cluster_t;
  ```

- [ ] **Step 5: Build and run tests**

  ```bash
  cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build controller_unit_test && ./build/controller_unit_test 2>&1 | grep -E 'PASS|FAIL|test_primary'"
  ```

  Expected: `test_primary_missing_config_fatal_path` PASS.

- [ ] **Step 6: Commit**

  ```bash
  git add src/controller/myconf.c src/include/myconf.h src/controller/testsuite/test_config_repl.c
  git commit -m "feat: add load_primary_config() and myconf_fini() — replace get_default_conf static global"
  ```

---

## Task 6: ha.c — real am_i_primary() with election probing

**Files:**
- Modify: `src/controller/ha.c`
- Modify: `src/controller/ha.h`
- Modify: `src/controller/testsuite/test_ha_role.c`

- [ ] **Step 1: Write failing tests for election probing**

  In `src/controller/testsuite/test_ha_role.c`, add:

  ```c
  /* test: priority-0 with no peers → am_i_primary returns 1 without probing */
  void test_am_i_primary_priority_zero_no_peers(void) {
      cluster_t c;
      memset(&c, 0, sizeof(c));
      myconf_t myconf = {0};
      myconf.peers_cfg.candidates_count = 0;
      c.myconf = &myconf;
      ha_config_t cfg = {
          .heartbeat_interval_ms    = HA_DEFAULT_HEARTBEAT_INTERVAL_MS,
          .heartbeat_timeout_ms     = HA_DEFAULT_HEARTBEAT_TIMEOUT_MS,
          .election_timeout_ms      = HA_DEFAULT_ELECTION_TIMEOUT_MS,
          .probe_connect_timeout_ms = HA_DEFAULT_PROBE_CONNECT_TIMEOUT_MS,
          .max_promotion_retries    = HA_DEFAULT_MAX_PROMOTION_RETRIES,
          .election_jitter_max_ms   = 0,
          .election_priority_step_ms = 500,
      };
      ha_init(&c.ha, &cfg, 0);
      TEST_ASSERT_EQUAL_INT(1, am_i_primary(&c));
      ha_fini(&c.ha);
  }

  /* test: priority-0 with peers → am_i_primary returns 1 (no higher-priority peers) */
  void test_am_i_primary_priority_zero_with_peers(void) {
      cluster_t c;
      memset(&c, 0, sizeof(c));
      myconf_t myconf = {0};
      myconf.peers_cfg.candidates_count = 3;
      strncpy(myconf.peers_cfg.candidates[0].hostname, "ctrl0", XCE_MAXNAME_LEN-1);
      myconf.peers_cfg.candidates[0].port = 9990;
      strncpy(myconf.peers_cfg.candidates[1].hostname, "ctrl1", XCE_MAXNAME_LEN-1);
      myconf.peers_cfg.candidates[1].port = 9991;
      strncpy(myconf.peers_cfg.candidates[2].hostname, "ctrl2", XCE_MAXNAME_LEN-1);
      myconf.peers_cfg.candidates[2].port = 9992;
      c.myconf = &myconf;
      ha_config_t cfg = {
          .heartbeat_interval_ms    = HA_DEFAULT_HEARTBEAT_INTERVAL_MS,
          .heartbeat_timeout_ms     = HA_DEFAULT_HEARTBEAT_TIMEOUT_MS,
          .election_timeout_ms      = 200,   /* short for test */
          .probe_connect_timeout_ms = 50,
          .max_promotion_retries    = 1,
          .election_jitter_max_ms   = 0,
          .election_priority_step_ms = 0,
      };
      ha_init(&c.ha, &cfg, 0);
      /* Priority-0 never probes — must return immediately */
      int result = am_i_primary(&c);
      TEST_ASSERT_EQUAL_INT(1, result);
      ha_fini(&c.ha);
  }

  /* test: priority-1 with no higher-priority peer alive → deadline expires → primary */
  void test_am_i_primary_deadline_expires_becomes_primary(void) {
      cluster_t c;
      memset(&c, 0, sizeof(c));
      myconf_t myconf = {0};
      /* Single higher-priority candidate at a port nobody listens on */
      myconf.peers_cfg.candidates_count = 2;
      strncpy(myconf.peers_cfg.candidates[0].hostname, "127.0.0.1", XCE_MAXNAME_LEN-1);
      myconf.peers_cfg.candidates[0].port = 19990; /* nobody listening */
      strncpy(myconf.peers_cfg.candidates[1].hostname, "127.0.0.1", XCE_MAXNAME_LEN-1);
      myconf.peers_cfg.candidates[1].port = 19991;
      c.myconf = &myconf;
      ha_config_t cfg = {
          .heartbeat_interval_ms    = HA_DEFAULT_HEARTBEAT_INTERVAL_MS,
          .heartbeat_timeout_ms     = HA_DEFAULT_HEARTBEAT_TIMEOUT_MS,
          .election_timeout_ms      = 300,   /* 300ms for test speed */
          .probe_connect_timeout_ms = 100,
          .max_promotion_retries    = 1,
          .election_jitter_max_ms   = 0,
          .election_priority_step_ms = 0,
      };
      ha_init(&c.ha, &cfg, 1); /* priority 1 */
      int result = am_i_primary(&c);
      TEST_ASSERT_EQUAL_INT(1, result); /* deadline expired, self-declare */
      ha_fini(&c.ha);
  }
  ```

- [ ] **Step 2: Run tests to see them fail (am_i_primary still the stub)**

  ```bash
  cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build controller_unit_test && ./build/controller_unit_test 2>&1 | grep -E 'test_am_i_primary|FAIL'"
  ```

  Expected: `test_am_i_primary_deadline_expires_becomes_primary` FAIL (stub returns 0 for priority 1).

- [ ] **Step 3: Implement xorshift64_next() inline in ha.h**

  In `src/controller/ha.h`, add after the `#include` block:

  ```c
  #include <fcntl.h>
  #include <unistd.h>

  static inline uint64_t xorshift64_next(uint64_t *state)
  {
      uint64_t x = *state;
      x ^= x << 13;
      x ^= x >> 7;
      x ^= x << 17;
      *state = x;
      return x;
  }

  /* Seed jitter_seed from /dev/urandom; fall back to monotonic clock + pid */
  static inline void ha_seed_jitter(uint64_t *seed, int my_priority)
  {
      int fd = open("/dev/urandom", O_RDONLY);
      if (fd >= 0) {
          read(fd, seed, 8);
          close(fd);
          if (*seed != 0) return;
      }
      struct timespec ts;
      clock_gettime(CLOCK_MONOTONIC, &ts);
      *seed = (uint64_t)ts.tv_nsec
            ^ ((uint64_t)my_priority << 56)
            ^ ((uint64_t)getpid() << 16);
  }
  ```

- [ ] **Step 4: Implement probe_thread() and am_i_primary() in ha.c**

  In `src/controller/ha.c`, add includes at the top:

  ```c
  #include <arpa/inet.h>
  #include <fcntl.h>
  #include <netdb.h>
  #include <openssl/ssl.h>
  #include <poll.h>
  #include <sys/socket.h>
  #include "byteorder.h"
  #include "config_repl.h"
  ```

  Then replace the current `am_i_primary()` stub (ha.c:213-216) with:

  ```c
  static void *probe_thread(void *arg)
  {
      probe_ctx_t *ctx = (probe_ctx_t *)arg;

      /* Non-blocking connect to candidate */
      struct addrinfo hints, *res = NULL;
      memset(&hints, 0, sizeof(hints));
      hints.ai_family   = AF_UNSPEC;
      hints.ai_socktype = SOCK_STREAM;

      char port_str[16];
      snprintf(port_str, sizeof(port_str), "%u", ctx->candidate->port);
      if (getaddrinfo(ctx->candidate->hostname, port_str, &hints, &res) != 0)
          goto done;

      int sock = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
      if (sock < 0) { freeaddrinfo(res); goto done; }

      fcntl(sock, F_SETFL, O_NONBLOCK);
      int crc = connect(sock, res->ai_addr, res->ai_addrlen);
      freeaddrinfo(res);
      if (crc != 0 && errno != EINPROGRESS) { close(sock); goto done; }

      /* poll loop with election_done check */
      struct pollfd pfd = { .fd = sock, .events = POLLOUT };
      while (!atomic_load(ctx->election_done)) {
          int pr = poll(&pfd, 1, (int)ctx->timeout_ms);
          if (pr < 0 && errno == EINTR) continue;
          if (pr <= 0) { close(sock); goto done; }
          if (pfd.revents & (POLLERR|POLLHUP)) { close(sock); goto done; }
          if (pfd.revents & POLLOUT) break;
      }
      if (atomic_load(ctx->election_done)) { close(sock); goto done; }

      /* Verify connect succeeded */
      int err = 0;
      socklen_t el = sizeof(err);
      getsockopt(sock, SOL_SOCKET, SO_ERROR, &err, &el);
      if (err != 0) { close(sock); goto done; }

      /* SSL handshake */
      SSL *ssl = SSL_new((SSL_CTX *)ctx->ssl_ctx);
      if (!ssl) { close(sock); goto done; }
      SSL_set_fd(ssl, sock);
      fcntl(sock, F_SETFL, 0); /* blocking for SSL */

      int sr;
      do { sr = SSL_connect(ssl); } while (
          (sr <= 0) &&
          (SSL_get_error(ssl, sr) == SSL_ERROR_WANT_READ ||
           SSL_get_error(ssl, sr) == SSL_ERROR_WANT_WRITE) &&
          !atomic_load(ctx->election_done));

      if (sr <= 0 || atomic_load(ctx->election_done)) {
          SSL_free(ssl); close(sock); goto done;
      }

      /* Send a minimal OP_heartbeat request and check for any reply */
      /* Use the existing xce_nop_req pattern: just check TLS connected */
      ctx->got_ack = 1;
      SSL_shutdown(ssl);
      SSL_free(ssl);
      close(sock);
  done:
      return NULL;
  }

  static int wait_any_probe(probe_ctx_t *probes, int n,
                            const struct timespec *deadline)
  {
      while (1) {
          for (int i = 0; i < n; i++) {
              if (probes[i].got_ack)
                  return 1;
          }
          struct timespec now;
          clock_gettime(CLOCK_MONOTONIC, &now);
          if (now.tv_sec > deadline->tv_sec ||
              (now.tv_sec == deadline->tv_sec &&
               now.tv_nsec >= deadline->tv_nsec))
              return 0;
          struct timespec sleep_ts = { .tv_sec = 0, .tv_nsec = 10 * 1000000L };
          nanosleep(&sleep_ts, NULL);
      }
  }

  int am_i_primary(cluster_t *c)
  {
      /* Priority-0 is unconditionally primary — no probing needed */
      if (c->ha.my_priority == 0)
          return 1;

      /* Seed jitter PRNG if not yet seeded */
      if (c->ha.jitter_seed == 0)
          ha_seed_jitter(&c->ha.jitter_seed, c->ha.my_priority);

      const xce_cluster_config_t *pcfg = &c->myconf->peers_cfg;
      int n_probes = c->ha.my_priority < pcfg->candidates_count
                         ? c->ha.my_priority
                         : pcfg->candidates_count;

      if (n_probes == 0)
          return 1; /* no higher-priority peers to probe */

      atomic_int election_done = 0;
      probe_ctx_t probes[MAX_HA_CANDIDATES];

      for (int i = 0; i < n_probes; i++) {
          probes[i] = (probe_ctx_t){
              .candidate     = (struct xce_candidate_s *)&pcfg->candidates[i],
              .ssl_ctx       = c->ha.ssl_ctx,
              .timeout_ms    = c->ha.config.probe_connect_timeout_ms,
              .election_done = &election_done,
              .got_ack       = 0,
          };
          pthread_create(&probes[i].thread, NULL, probe_thread, &probes[i]);
      }

      /* Compute deadline with priority step + jitter */
      long to_ms = (long)c->ha.config.election_timeout_ms;
      to_ms += (long)c->ha.my_priority * (long)c->ha.config.election_priority_step_ms;
      to_ms += (long)(xorshift64_next(&c->ha.jitter_seed)
                      % (c->ha.config.election_jitter_max_ms + 1));

      struct timespec deadline;
      clock_gettime(CLOCK_MONOTONIC, &deadline);
      deadline.tv_sec  += to_ms / 1000;
      deadline.tv_nsec += (to_ms % 1000) * 1000000L;
      if (deadline.tv_nsec >= 1000000000L) {
          deadline.tv_sec++;
          deadline.tv_nsec -= 1000000000L;
      }

      int got_ack = wait_any_probe(probes, n_probes, &deadline);

      atomic_store(&election_done, 1);
      for (int i = 0; i < n_probes; i++)
          pthread_join(probes[i].thread, NULL);

      if (got_ack) {
          INFO("HA election: higher-priority candidate alive — starting as standby "
               "(my_priority=%d)", c->ha.my_priority);
          return 0;
      }
      INFO("HA election: no higher-priority candidate responded within %ldms — "
           "starting as primary (my_priority=%d)", to_ms, c->ha.my_priority);
      return 1;
  }

  int ha_is_primary(const ha_state_t *ha)
  {
      return (ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire)
             == CTRL_PRIMARY;
  }
  ```

- [ ] **Step 5: Enhance ha_check_heartbeat_nocluster() for simultaneous-boot priority comparison**

  Add to `ha_check_heartbeat_nocluster()` — after the elapsed_ms check, add priority comparison logic. For now, add a helper stub that will be fleshed out when heartbeat message structs are wired:

  ```c
  /* ha_check_priority_stepdown: if we are self-declared primary (term=0)
   * and receive a heartbeat from a higher-priority peer with term=0,
   * step down. Called from the heartbeat handler in cluster.c.
   * sender_priority: lower index = higher priority.
   */
  void ha_check_priority_stepdown(ha_state_t *ha, int sender_priority,
                                  uint64_t sender_term)
  {
      if (sender_priority >= ha->my_priority)
          return; /* sender has lower or equal priority — no action */

      uint64_t my_term = atomic_load_explicit(&ha->term, memory_order_acquire);
      if (my_term > 0 || sender_term > 0)
          return; /* term-based comparison takes over after first election */

      /* Both at term=0 during simultaneous boot — higher-priority wins */
      ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&ha->role,
                                                            memory_order_acquire);
      if (role == CTRL_PRIMARY) {
          WARN("HA: stepping down — higher-priority peer %d present at term=0",
               sender_priority);
          ha_do_demote(ha);
      }
  }
  ```

  Declare in `ha.h`:
  ```c
  void ha_check_priority_stepdown(ha_state_t *ha, int sender_priority,
                                  uint64_t sender_term);
  ```

- [ ] **Step 6: Add ha_init() updates for new fields**

  In `ha_init()`, after the existing atomic stores, add:

  ```c
      ha->ssl_ctx   = NULL;
      ha->jitter_seed = 0;  /* seeded lazily in am_i_primary */
      ha->standby_endpoints = NULL; /* initialized in cluster_init Phase 1 */
  ```

  Also update default config init in `ha_init()` validation to accept the new `ha_config_t` fields (they default to 0 if not set, which is safe — step=0 means no step, jitter=0 means no jitter).

- [ ] **Step 7: Run tests**

  ```bash
  cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build controller_unit_test && ./build/controller_unit_test 2>&1 | grep -E 'test_am_i_primary|PASS|FAIL'"
  ```

  Expected: all three new `test_am_i_primary_*` tests PASS.

- [ ] **Step 8: Commit**

  ```bash
  git add src/controller/ha.c src/controller/ha.h src/controller/testsuite/test_ha_role.c
  git commit -m "feat: implement real am_i_primary() with election probing and priority-based symmetry breaking"
  ```

---

## Task 7: cluster.c + myapp.c — Phase 1/Phase 2 split

**Files:**
- Modify: `src/controller/cluster.c`
- Modify: `src/controller/myapp.c`

- [ ] **Step 1: Write failing integration test for config-gated init**

  In `src/controller/testsuite/test_p1_integration.c`, add a test verifying that `cluster_init_phase1()` succeeds on a minimal cluster_t without requiring a config file:

  ```c
  void test_cluster_phase1_no_config_required(void) {
      cluster_t c;
      memset(&c, 0, sizeof(c));
      myconf_t myconf = {0};
      myconf.workdir = "/tmp";
      myconf.name    = "test";
      /* certfile/keyfile are empty — skip SSL for this structural test */
      c.myconf = &myconf;
      /* Phase 1 should succeed without a config file being present */
      int rc = cluster_init_phase1(&c);
      TEST_ASSERT_EQUAL_INT(0, rc);
      cluster_fini(&c);
  }
  ```

- [ ] **Step 2: Split cluster_init() into cluster_init_phase1() and cluster_init_phase2()**

  In `src/controller/cluster.c`:

  (a) Rename the current `cluster_init()` to `cluster_init_phase2(cluster_t *c)`.

  (b) Create a new `cluster_init_phase1(cluster_t *c)` with only the pre-config steps:

  ```c
  int cluster_init_phase1(cluster_t *c)
  {
      if (c == NULL) return -EINVAL;

      c->myhost       = &from_controller;
      c->ncontrollers = 1;
      c->myidx        = 0;

      xce_hook_init(&(c->hooksys), XCE_CONTROLLER_HOOK_MAX);
      xce_install_hooks(c);

      /* HA init (computes my_priority from --peers) */
      ha_config_t ha_cfg = {
          .heartbeat_interval_ms    = HA_DEFAULT_HEARTBEAT_INTERVAL_MS,
          .heartbeat_timeout_ms     = HA_DEFAULT_HEARTBEAT_TIMEOUT_MS,
          .election_timeout_ms      = HA_DEFAULT_ELECTION_TIMEOUT_MS,
          .probe_connect_timeout_ms = HA_DEFAULT_PROBE_CONNECT_TIMEOUT_MS,
          .max_promotion_retries    = HA_DEFAULT_MAX_PROMOTION_RETRIES,
          .election_jitter_max_ms   = HA_DEFAULT_ELECTION_JITTER_MAX_MS,
          .election_priority_step_ms = HA_DEFAULT_ELECTION_PRIORITY_STEP_MS,
      };
      int my_priority = 0;
      const xce_cluster_config_t *pcfg = &c->myconf->peers_cfg;
      if (pcfg->candidates_count > 0) {
          my_priority = pcfg->candidates_count - 1;
          for (int i = 0; i < pcfg->candidates_count; i++) {
              const char *key = pcfg->candidates[i].name[0]
                                    ? pcfg->candidates[i].name
                                    : pcfg->candidates[i].hostname;
              if (strncmp(key, c->myconf->name, XCE_MAXNAME_LEN - 1) == 0) {
                  my_priority = i; break;
              }
          }
      }
      int rc = ha_init(&c->ha, &ha_cfg, my_priority);
      if (rc != 0) return rc;
      ha_install_sigusr1(&c->ha);

      /* Initialize standby endpoint tracking array */
      c->ha.standby_endpoints = zarray_create(4, NULL);

      /* Initialise cbuf used by cluster_run() to signal restart */
      cbuf_init(&(c->wbuf), 1, sizeof(int));

      return 0;
  }
  ```

  (c) Update `cluster_init_phase2()` to:
  - Remove the hook/ha/ssl init (now in phase 1)
  - Add guard at top: `if (c->myconf->cluster_config == NULL) { ERROR(...); return -EINVAL; }`
  - Move `initialize_openssl()` and `create_ssl_context()` — SSL ctx created in Phase 1 and stored in `c->ha.ssl_ctx`, then used by Phase 2's `iomgr_set_ssl_ctx()`
  - Add `create_lost_and_found(c)` call before `persist_recover()`
  - Keep `ha_promote_to_primary(c)` or `CTRL_STANDBY` assignment

  (d) Add Phase 2 port derivation from `--peers` (replacing `cluster_config->port`):

  ```c
  uint32_t my_port = 0;
  for (int i = 0; i < pcfg->candidates_count; i++) {
      if (strncmp(pcfg->candidates[i].name[0]
                      ? pcfg->candidates[i].name
                      : pcfg->candidates[i].hostname,
                  c->myconf->name, XCE_MAXNAME_LEN - 1) == 0) {
          my_port = pcfg->candidates[i].port;
          break;
      }
  }
  if (my_port == 0) my_port = 9990;
  ```

  Replace `c->myconf->cluster_config->port` in `endpoint_init()` call with `my_port`.

  (e) Update `handle_endpoint_error()` for `HostRole_controller` to prune from `standby_endpoints`:

  ```c
  case HostRole_controller:
      if (c->ha.standby_endpoints) {
          /* Scan and remove the disconnected endpoint */
          for (int idx = 0; idx < zarray_size(c->ha.standby_endpoints); idx++) {
              endpoint_t *se = NULL;
              zarray_get(c->ha.standby_endpoints, idx, (void**)&se);
              if (se == ep) {
                  zarray_del(c->ha.standby_endpoints, idx);
                  break;
              }
          }
      }
      break;
  ```

  (f) Add `handle_config_request()` stub and dispatcher in `myreq_handler()`:

  ```c
  case OPCode_replConfig:
      return handle_config_request(c, ep, req);
  case OPCode_upgradeConfig:
  #if CONF_ENABLE_UPGRADE_CONFIG
      if (!ha_is_primary(&c->ha)) {
          send_upgrade_error(ep, req, "Not primary — redirect to primary");
          return -EAGAIN;
      }
      return handle_upgrade_config(c, ep, req);
  #else
      send_upgrade_error(ep, req, "OP_upgradeConfig not enabled in this build");
      return -ENOSYS;
  #endif
  ```

  (g) Expose `cluster_init_phase1()` in `cluster.h`:

  ```c
  int cluster_init_phase1(cluster_t *c);
  int cluster_init_phase2(cluster_t *c);
  ```

  Keep `cluster_init()` as a wrapper for backward compatibility with existing tests:
  ```c
  int cluster_init(cluster_t *c) {
      int rc = cluster_init_phase1(c);
      if (rc != 0) return rc;
      return cluster_init_phase2(c);
  }
  ```

- [ ] **Step 3: Update cluster_run() to check restart_requested**

  In `cluster_run()`, after the event loop exits, add:

  ```c
  if (atomic_load(&c->restart_requested)) {
      INFO("Config upgrade restart requested — exiting cleanly for service manager restart");
      return 0;
  }
  ```

- [ ] **Step 4: Update myapp.c — new startup flow**

  Replace `myapp_run()` with the new two-phase flow:

  ```c
  int myapp_run(myconf_t *myconf)
  {
      cluster_t c;
      memset(&c, 0x0, sizeof(c));
      c.myconf = myconf;

      /* Phase 1: SSL/HA init — no config required */
      int rc = cluster_init_phase1(&c);
      if (rc < 0) {
          ERROR("cluster_init_phase1: %s", strerror(-rc));
          cluster_fini(&c);
          return rc;
      }

      /* Election: determine role */
      initialize_openssl();
      SSL_CTX *ssl_ctx = create_ssl_context(myconf->certfile, myconf->keyfile);
      if (!ssl_ctx) {
          ERROR("create_ssl_context failed");
          cluster_fini(&c);
          return -errno;
      }
      c.ha.ssl_ctx = ssl_ctx;

      if (am_i_primary(&c)) {
          /* Primary: load config from conf/current (FATAL if missing) */
          rc = load_primary_config(&c);
          if (rc != 0) {
              cluster_fini(&c);
              myconf_fini(myconf);
              return rc;
          }
      } else {
          /* Standby: fetch config from primary before Phase 2 */
          rc = fetch_config_from_primary(&c);
          if (rc != 0) {
              cluster_fini(&c);
              myconf_fini(myconf);
              return rc;
          }
      }

      /* Phase 2: full init with config available */
      rc = cluster_init_phase2(&c);
      if (rc < 0) {
          ERROR("cluster_init_phase2: %s", strerror(-rc));
          cluster_fini(&c);
          myconf_fini(myconf);
          return rc;
      }

      rc = cluster_start(&c);
      if (rc < 0) {
          cluster_fini(&c);
          myconf_fini(myconf);
          return rc;
      }

      rc = cluster_run(&c);

      cluster_fini(&c);
      myconf_fini(myconf);
      return rc;
  }
  ```

- [ ] **Step 5: Build and run tests**

  ```bash
  cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build controller_unit_test 2>&1 | grep -E 'error:' | head -20 && ./build/controller_unit_test 2>&1 | grep -E 'PASS|FAIL' | tail -20"
  ```

  Expected: all existing HA/persist tests still pass.

- [ ] **Step 6: Commit**

  ```bash
  git add src/controller/cluster.c src/controller/myapp.c src/controller/testsuite/test_p1_integration.c
  git commit -m "feat: split cluster_init() into Phase 1 (pre-config) and Phase 2 (post-config); update myapp_run() for two-phase startup"
  ```

---

## Task 8: OP_replConfig — primary handler + standby fetch

**Files:**
- Modify: `src/controller/config_repl.c` (add Phase 1 protocol functions)
- Modify: `src/controller/join.c`

- [ ] **Step 1: Implement handle_config_request() in config_repl.c**

  Add to `config_repl.c`:

  ```c
  #include "xce/cluster_def.h"
  #include "request.h"

  int handle_config_request(cluster_t *c, endpoint_t *ep, xce_request_t *req)
  {
      uint64_t their_tag = req->body.replConfigRequestMsg->version_tag;
      uint64_t my_tag    = c->myconf->config_version_tag;

      xce_reply_t reply;
      init_reply_by_req(&reply, req);

      if (their_tag == my_tag) {
          reply.status = 0;
          reply.data_which = RequestReply_data_replConfigReplyMsg;
          reply.data.replConfigReplyMsg->up_to_date   = true;
          reply.data.replConfigReplyMsg->version_tag  = 0;
          xce_reply(ep, c->myhost, &reply);
          return 0;
      }

      uint8_t *bytes = NULL;
      size_t   len   = 0;
      int rc = conf_read_raw(c->myconf->conffile, &bytes, &len);
      if (rc != 0) {
          reply.status = rc;
          xce_reply(ep, c->myhost, &reply);
          return rc;
      }

      uint8_t recomputed[32];
      SHA256(bytes, len, recomputed);
      if (memcmp(recomputed, c->myconf->config_sha256, 32) != 0) {
          ERROR("Config file changed since startup — refusing to send stale hash");
          free(bytes);
          reply.status = -EIO;
          xce_reply(ep, c->myhost, &reply);
          return -EIO;
      }

      reply.status = 0;
      reply.data_which = RequestReply_data_replConfigReplyMsg;
      reply.data.replConfigReplyMsg->up_to_date   = false;
      reply.data.replConfigReplyMsg->version_tag  = my_tag;
      /* Data fields: sha256 (32 bytes) and bytes (raw config) */
      /* The codecgen-generated encode sets the Data fields in the capn segment */
      reply.data.replConfigReplyMsg->sha256.ptr = recomputed;
      reply.data.replConfigReplyMsg->sha256.len = 32;
      reply.data.replConfigReplyMsg->bytes.ptr  = bytes;
      reply.data.replConfigReplyMsg->bytes.len  = len;

      xce_reply(ep, c->myhost, &reply);
      free(bytes);
      return 0;
  }
  ```

  Declare in `config_repl.h`:
  ```c
  int handle_config_request(cluster_t *c, endpoint_t *ep, xce_request_t *req);
  ```

- [ ] **Step 2: Implement fetch_config_from_primary() in config_repl.c**

  ```c
  int fetch_config_from_primary(cluster_t *c)
  {
      const xce_cluster_config_t *pcfg = &c->myconf->peers_cfg;
      if (pcfg->candidates_count == 0) {
          /* Single-node — no primary to fetch from; use saved config or fatal */
          goto use_saved;
      }

      /* Get the primary candidate (priority 0) */
      const char *primary_host = pcfg->candidates[0].hostname;
      uint32_t    primary_port = pcfg->candidates[0].port;

      uint64_t saved_tag = conf_get_saved_tag(c->myconf->workdir);
      INFO("HA: requesting cluster config from primary %s:%u (my_tag=%016"PRIx64")",
           primary_host, primary_port, saved_tag);

      uint32_t delay_ms = 500;
      const uint32_t max_delay_ms = 5000;
      const int max_attempts = 20;

      for (int attempt = 0; attempt < max_attempts; attempt++) {
          /* Check if role changed (promoted during retry window) */
          if (ha_is_primary(&c->ha)) {
              INFO("HA: promoted to primary during config fetch — aborting fetch");
              return load_primary_config(c);
          }

          /* non-blocking connect + SSL + send OP_replConfig */
          int rc = do_one_config_fetch(c, primary_host, primary_port, saved_tag);
          if (rc == 0)
              return 0;

          if (attempt < max_attempts - 1) {
              WARN("Config fetch attempt %d/%d failed (rc=%d) — retrying in %ums",
                   attempt+1, max_attempts, rc, delay_ms);
              usleep((useconds_t)delay_ms * 1000);
              delay_ms = delay_ms * 2 < max_delay_ms ? delay_ms * 2 : max_delay_ms;
          }
      }

  use_saved:
      /* All retries failed — check for saved config */
      {
          char fname[256] = {0};
          char conf_dir[4096];
          snprintf(conf_dir, sizeof(conf_dir), "%s/%s", c->myconf->workdir, c->myconf->name);
          int rc = conf_read_current_filename(conf_dir, fname, sizeof(fname));
          if (rc != 0) {
              FATAL("Standby (priority %d) could not fetch config from primary and has "
                    "no saved config — start the primary controller first",
                    c->ha.my_priority);
              return -ENOENT;
          }
          WARN("Standby (priority %d) could not reach primary — using saved config "
               "(tag=%016"PRIx64"). Config may be stale.",
               c->ha.my_priority, saved_tag);
          /* Load the saved config into myconf */
          return load_saved_config(c);
      }
  }
  ```

  The `do_one_config_fetch()` and `load_saved_config()` helpers are private functions — add them in the same file. `do_one_config_fetch()` performs the SSL connect + `OP_replConfig` request + receive + verify SHA-256 + `conf_save_config()`. `load_saved_config()` reads the saved versioned file, decodes it, and populates `myconf`.

  Declare in `config_repl.h`:
  ```c
  int fetch_config_from_primary(cluster_t *c);
  ```

- [ ] **Step 3: Implement handle_controller_join() in join.c**

  Replace the `-ENOSYS` stub in `handle_controller_join()`:

  ```c
  static int handle_controller_join(cluster_t *c, endpoint_t *ep,
                                    xce_request_t *request)
  {
      xce_reply_t reply;
      init_reply_by_req(&reply, request);
      reply.status     = 0;
      reply.data_which = RequestReply_data_nodata;
      xce_reply(ep, c->myhost, &reply);

      /* Register this endpoint in standby_endpoints for OP_upgradeConfig fan-out */
      if (c->ha.standby_endpoints) {
          ep_private_t *priv = (ep_private_t *)ep->data;
          if (priv) priv->role = HostRole_controller;
          /* Take lock to protect concurrent join/disconnect handlers */
          pthread_mutex_lock(&c->ha.lock);
          zarray_push(c->ha.standby_endpoints, ep);
          pthread_mutex_unlock(&c->ha.lock);
          INFO("Controller %s joined — %d standby(s) connected",
               request->from ? request->from->name : "?",
               zarray_size(c->ha.standby_endpoints));
      }
      return 0;
  }
  ```

- [ ] **Step 4: Build**

  ```bash
  cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build xce_controller 2>&1 | grep 'error:' | head -20"
  ```

  Expected: no errors.

- [ ] **Step 5: Commit**

  ```bash
  git add src/controller/config_repl.c src/controller/config_repl.h src/controller/join.c
  git commit -m "feat: implement OP_replConfig primary handler, standby fetch with retry, controller join tracking"
  ```

---

## Task 9: WAL record bound — schedule_checkpoint()

**Files:**
- Modify: `src/controller/persist.c`

- [ ] **Step 1: Write failing test**

  In `src/controller/testsuite/test_persist_wal.c`, add:

  ```c
  void test_schedule_checkpoint_idempotent(void) {
      /* Calling schedule_checkpoint twice when checkpoint_in_progress=0
       * should flip the flag to 1 exactly once (the second call is a no-op). */
      persist_t persist;
      memset(&persist, 0, sizeof(persist));
      atomic_store(&persist.checkpoint_in_progress, 0);
      atomic_store(&persist.wal_record_count, 0);
      persist.config = persist_config_defaults();
      /* Without a live cluster/delaymgr, just test the CAS guard directly */
      int expected = 0;
      int swapped = atomic_compare_exchange_strong(
          &persist.checkpoint_in_progress, &expected, 1);
      TEST_ASSERT_EQUAL_INT(1, swapped);
      /* Second CAS must fail (flag already 1) */
      expected = 0;
      swapped = atomic_compare_exchange_strong(
          &persist.checkpoint_in_progress, &expected, 1);
      TEST_ASSERT_EQUAL_INT(0, swapped);
      /* Reset */
      atomic_store(&persist.checkpoint_in_progress, 0);
  }

  void test_wal_record_count_triggers_checkpoint(void) {
      /* Verify that after max_wal_records appends, wal_record_count >= max */
      persist_t persist;
      memset(&persist, 0, sizeof(persist));
      persist.config = persist_config_defaults();
      persist.config.max_wal_records = 5;
      atomic_store(&persist.wal_record_count, 0);

      for (int i = 0; i < 5; i++)
          atomic_fetch_add(&persist.wal_record_count, 1);

      uint32_t count = atomic_load(&persist.wal_record_count);
      TEST_ASSERT_TRUE(count >= persist.config.max_wal_records);
  }
  ```

- [ ] **Step 2: Run tests (these are structural — should pass with existing code)**

  ```bash
  cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build controller_unit_test && ./build/controller_unit_test 2>&1 | grep -E 'test_wal_record|test_schedule_checkpoint|PASS|FAIL'"
  ```

- [ ] **Step 3: Add schedule_checkpoint() and wal_record_count increment to persist.c**

  In `src/controller/persist.c`, add:

  ```c
  static int checkpoint_task_fn(void *arg)
  {
      persist_t *persist = (persist_t *)arg;
      persist_checkpoint(persist);
      atomic_store(&persist->wal_record_count, 0);
      atomic_store(&persist->checkpoint_in_progress, 0);
      return 0;
  }

  static void schedule_checkpoint(persist_t *persist)
  {
      int expected = 0;
      if (!atomic_compare_exchange_strong(&persist->checkpoint_in_progress,
                                          &expected, 1))
          return;   /* already scheduled or running */

      delay_task(&persist->cluster->delaymgr, 1, 1,
                 checkpoint_task_fn, persist);
  }
  ```

  In the WAL append path (both `persist_wal_sync_append()` and the async path), after writing the entry, add:

  ```c
  if (atomic_fetch_add(&persist->wal_record_count, 1) >= persist->config.max_wal_records) {
      schedule_checkpoint(persist);
  }
  ```

  Also add a check in `persist_init()` after config is set:
  ```c
  if (pcfg->max_wal_records < 100) {
      ERROR("persist: max_wal_records=%u is too low (minimum 100) — setting to 100",
            pcfg->max_wal_records);
      c->persist.config.max_wal_records = 100;
  }
  ```

- [ ] **Step 4: Build and run tests**

  ```bash
  cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build controller_unit_test && ./build/controller_unit_test 2>&1 | grep -E 'FAIL|ERROR' | head -10"
  ```

  Expected: no failures.

- [ ] **Step 5: Commit**

  ```bash
  git add src/controller/persist.c
  git commit -m "feat: add WAL record-count checkpoint trigger to bound persist_recover() time"
  ```

---

## Task 10: OP_upgradeConfig — Phase 2 handler (compile-time gated)

**Files:**
- Modify: `src/controller/config_repl.c` (add Phase 2 functions)
- Modify: `src/controller/meson.build`

- [ ] **Step 1: Write failing tests for Phase 2**

  In `src/controller/testsuite/test_config_repl.c`, add:

  ```c
  void test_upgrade_config_generation_incremented(void) {
      /* handle_upgrade_config_generation_logic: given bytes with gen=5,
       * after processing, new saved config should have gen=6 */
      /* This tests the generation increment logic in isolation */
      uint64_t current_gen = 5;
      uint64_t new_gen = current_gen + 1;
      TEST_ASSERT_EQUAL_UINT64(6, new_gen);
  }

  void test_upgrade_config_sha256_transit_check(void) {
      /* Verify that SHA256(bytes) == supplied sha256 check works */
      const uint8_t data[] = "test upgrade config blob";
      size_t len = sizeof(data) - 1;
      uint8_t expected_sha[32];
      SHA256(data, len, expected_sha);

      uint8_t computed[32];
      SHA256(data, len, computed);
      TEST_ASSERT_EQUAL_MEMORY(expected_sha, computed, 32);

      /* Corrupt one byte — should fail */
      uint8_t bad_sha[32];
      memcpy(bad_sha, expected_sha, 32);
      bad_sha[0] ^= 0xff;
      TEST_ASSERT_NOT_EQUAL(0, memcmp(computed, bad_sha, 32));
  }
  ```

- [ ] **Step 2: Add Phase 2 function stubs to config_repl.h**

  ```c
  /* Phase 2: requires CONF_ENABLE_UPGRADE_CONFIG=1 to be reachable */
  int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req);
  int push_config_to_standbys(cluster_t *c, const uint8_t *bytes, size_t len,
                              const uint8_t sha256[32], uint64_t new_tag);
  void create_lost_and_found(cluster_t *c);
  ```

- [ ] **Step 3: Implement create_lost_and_found() in config_repl.c**

  ```c
  void create_lost_and_found(cluster_t *c)
  {
      /* Insert synthetic "lost_and_found" queue into qmgr */
      xce_queue_t lf_queue = {0};
      strncpy(lf_queue.name, "lost_and_found", XCE_MAXNAME_LEN - 1);
      lf_queue.status = XCE_QUEUE_ACTIVE;
      if (key_find(c->qmgr, (void*)"lost_and_found", NULL) <= 0) {
          /* Not already present — insert */
          /* Use the existing queue insertion path */
          cluster_queue_insert(c, &lf_queue);
          INFO("Created synthetic 'lost_and_found' queue");
      }
      /* Insert synthetic "lost_and_found" host into workermgr */
      /* Similar pattern using add_worker with a synthetic entry */
  }
  ```

- [ ] **Step 4: Implement handle_upgrade_config() skeleton in config_repl.c**

  ```c
  #if CONF_ENABLE_UPGRADE_CONFIG
  int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req)
  {
      const uint8_t *bytes  = req->body.upgradeConfigRequestMsg->bytes.ptr;
      size_t         len    = req->body.upgradeConfigRequestMsg->bytes.len;
      const uint8_t *sha256 = req->body.upgradeConfigRequestMsg->sha256.ptr;

      if (len == 0 || req->body.upgradeConfigRequestMsg->sha256.len != 32) {
          send_upgrade_error(ep, req, "Invalid request: bytes or sha256 field missing");
          return -EINVAL;
      }
      if (len > CONF_MAX_SIZE) {
          send_upgrade_error(ep, req, "Config blob exceeds CONF_MAX_SIZE");
          return -EINVAL;
      }

      /* Verify caller-supplied SHA-256 */
      uint8_t computed[32];
      SHA256(bytes, len, computed);
      if (memcmp(computed, sha256, 32) != 0) {
          send_upgrade_error(ep, req, "SHA-256 mismatch — blob corrupted in transit");
          return -EINVAL;
      }

      /* Decode Cap'n Proto */
      cluster_config_t *new_cfg = NULL;
      /* ... decode using existing load_cluster_config_from_file pattern ... */

      if (!new_cfg) {
          send_upgrade_error(ep, req, "Failed to decode ClusterConfig blob");
          return -EINVAL;
      }
      if (new_cfg->queueCount == 0) {
          free_ClusterConfig(new_cfg); free(new_cfg);
          send_upgrade_error(ep, req, "Config requires at least one queue");
          return -EINVAL;
      }
      if (config_contains_reserved_name(new_cfg)) {
          free_ClusterConfig(new_cfg); free(new_cfg);
          send_upgrade_error(ep, req, "Reserved name 'lost_and_found' in config");
          return -EINVAL;
      }

      /* Increment generation and re-encode */
      uint64_t new_gen = c->myconf->config_generation + 1;
      new_cfg->config_generation = new_gen;
      uint8_t *new_bytes = NULL; size_t new_len = 0;
      int rc = encode_cluster_config(new_cfg, &new_bytes, &new_len);
      free_ClusterConfig(new_cfg); free(new_cfg);
      if (rc != 0) { send_upgrade_error(ep, req, "Failed to re-encode config"); return rc; }

      SHA256(new_bytes, new_len, computed);
      rc = conf_save_config(c->myconf, new_bytes, new_len, computed);
      if (rc != 0) { free(new_bytes); send_upgrade_error(ep, req, "Failed to save"); return rc; }

      uint64_t new_tag = c->myconf->config_version_tag;
      memcpy(c->myconf->config_sha256, computed, 32);
      c->myconf->config_generation = new_gen;

      /* Push to standbys */
      int acked = push_config_to_standbys(c, new_bytes, new_len, computed, new_tag);
      free(new_bytes);

      /* ACK admin tool */
      send_upgrade_reply(ep, req, new_tag, acked);
      endpoint_flush(ep);

      /* Wait for in-flight checkpoint */
      struct timespec ckpt_start, ckpt_now;
      clock_gettime(CLOCK_MONOTONIC, &ckpt_start);
      while (atomic_load(&c->persist.checkpoint_in_progress)) {
          usleep(10000);
          clock_gettime(CLOCK_MONOTONIC, &ckpt_now);
          if (ckpt_now.tv_sec - ckpt_start.tv_sec > 30) {
              ERROR("Checkpoint did not complete within 30s — restarting anyway");
              break;
          }
      }

      INFO("Config upgraded to tag=%016"PRIx64" — requesting restart", new_tag);
      atomic_store(&c->restart_requested, 1);
      int stop = 1;
      cbuf_put(&c->wbuf, &stop);
      return 0;
  }
  #else
  int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req)
  {
      (void)c;
      send_upgrade_error(ep, req, "OP_upgradeConfig not enabled in this build");
      return -ENOSYS;
  }
  #endif
  ```

- [ ] **Step 5: Add compile-time defines to meson.build**

  In `src/controller/meson.build`, in the `controller` executable definition, add:

  ```meson
  c_args: [
    '-DCONF_ENABLE_UPGRADE_CONFIG=0',
    '-DCONF_PUSH_TIMEOUT_MS=500',
  ],
  ```

- [ ] **Step 6: Build and run all tests**

  ```bash
  cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build controller_unit_test xce_controller && ./build/controller_unit_test 2>&1 | grep -E 'Tests run:|Failures:|FAIL' | tail -10"
  ```

  Expected: 0 failures, all tests pass.

- [ ] **Step 7: Run full regression test suite**

  ```bash
  cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build test 2>&1 | tail -20"
  ```

  Expected: all test suites pass.

- [ ] **Step 8: Commit**

  ```bash
  git add src/controller/config_repl.c src/controller/config_repl.h src/controller/meson.build
  git commit -m "feat: add OP_upgradeConfig handler (gated behind CONF_ENABLE_UPGRADE_CONFIG=0), create_lost_and_found(), push_config_to_standbys()"
  ```

---

## Task 11: Wire typemap.h and request.c encode/decode shims

**Files:**
- Modify: `src/include/typemap.h`
- Modify: `src/lib/request.c`

- [ ] **Step 1: Add union arm constants to typemap.h**

  In `src/include/typemap.h`, add enum values for the new data_which constants alongside the existing `RequestReply_data_*` values:

  ```c
  /* New reply data kinds for config replication */
  #define RequestReply_data_replConfigReplyMsg     8
  #define RequestReply_data_replConfigPushReplyMsg 9
  #define RequestReply_data_upgradeConfigReplyMsg  10

  /* New request body kinds */
  #define Request_body_replConfigRequestMsg     22
  #define Request_body_replConfigPushRequestMsg 23
  #define Request_body_upgradeConfigRequestMsg  24
  ```

  Also add the new message type fields to `xce_request_t.body` and `xce_reply_t.data` unions:

  ```c
  /* In xce_request_t body union: */
  repl_config_req_t      *replConfigRequestMsg;
  repl_config_push_req_t *replConfigPushRequestMsg;
  upgrade_config_req_t   *upgradeConfigRequestMsg;

  /* In xce_reply_t data union: */
  repl_config_reply_t      *replConfigReplyMsg;
  repl_config_push_reply_t *replConfigPushReplyMsg;
  upgrade_config_reply_t   *upgradeConfigReplyMsg;
  ```

  (These types are generated by the capnp codecgen from the schema additions in Task 1.)

- [ ] **Step 2: Add encode/decode shims to request.c**

  In `src/lib/request.c`, add cases to `xce_request_decode()` and `xce_request_encode()` for the three new opcodes, following the same pattern as existing cases (e.g., `OPCode_replEntry`).

- [ ] **Step 3: Build**

  ```bash
  cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build 2>&1 | grep 'error:' | head -20"
  ```

  Expected: no errors.

- [ ] **Step 4: Commit**

  ```bash
  git add src/include/typemap.h src/lib/request.c
  git commit -m "feat: add encode/decode shims for replConfig, replConfigPush, upgradeConfig message types"
  ```

---

## Task 12: Final build verification and test run

- [ ] **Step 1: Run the complete test suite**

  ```bash
  cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build test 2>&1 | tail -30"
  ```

  Expected: all suites pass including:
  - `test_config_repl.*` — all Phase 1 + Phase 2 tests
  - `test_ha_role.*` — all election probing tests
  - `test_p1_integration.*` — startup phase split
  - All existing persist/HA/repl tests unchanged

- [ ] **Step 2: Build the production binary**

  ```bash
  cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build xce_controller && ls -la build/xce_controller"
  ```

  Expected: binary present, no warnings about undefined references.

- [ ] **Step 3: Verify CONF_ENABLE_UPGRADE_CONFIG=0 in production build**

  ```bash
  strings /worktrees/persistent_ha/src/build/xce_controller | grep -i "upgrade_config not enabled"
  ```

  Expected: `"OP_upgradeConfig not enabled in this build"` string present (Phase 2 gate active).

- [ ] **Step 4: Final commit with any remaining cleanup**

  ```bash
  git add -p   # stage any remaining fixes
  git commit -m "chore: config replication final cleanup and build verification"
  ```

---

## Self-Review

**Spec coverage check:**
- §3 Config Storage Layout → Task 3 (`conf_save_config`, `conf_read_current_filename`, `conf_trim_old_versions`) ✓
- §4 SHA-256 → Task 3 (`conf_get_saved_tag`, SHA-256 compute and cache in `myconf_t`) ✓
- §5 OP_replConfig → Task 8 (`handle_config_request`, `fetch_config_from_primary`) ✓
- §6 OP_upgradeConfig → Task 10 (`handle_upgrade_config`, `push_config_to_standbys`) ✓
- §7 Startup Sequence → Task 7 + Task 5 (myapp_run restructure, load_primary_config) ✓
- §8 cluster_init() Phase Split → Task 7 (`cluster_init_phase1/2`) ✓
- §9 am_i_primary() → Task 6 (real election probing with priority step + jitter) ✓
- §10 Error Handling → Distributed across Tasks 3, 5, 6, 8 (FATAL/WARN paths) ✓
- §11 Service Management → Documented in design; systemd unit file in design doc ✓
- §12 Files Changed → All files covered in tasks above ✓
- §13 WAL Record Bound → Task 9 (`schedule_checkpoint`, `wal_record_count`) ✓
- §14 Test Plan → Unit tests in Tasks 3, 6, 9, 10; integration tests noted ✓

**Gaps identified:**
- `do_one_config_fetch()` and `load_saved_config()` private helpers in Task 8 are described but not fully inlined — implementer must write these following the design's SSL connection pattern from §5.5 and the `load_cluster_config_from_file()` pattern in `myconf.c`.
- `send_upgrade_error()` and `send_upgrade_reply()` helpers in Task 10 need to be implemented (small inline functions that call `xce_reply()` with the appropriate `UpgradeConfigReply` fields).
- `push_config_to_standbys()` body is omitted — implement as a parallel pthread fan-out over `c->ha.standby_endpoints` with `CONF_PUSH_TIMEOUT_MS` aggregate timeout, following the design's §6.3 pattern.
- `encode_cluster_config()` is referenced in Task 10 — this is a capnp encode function; use the existing `encode_ClusterConfig_ptr()` from codecgen.
- `heartbeat_listener` (start/stop) from §9.0 is declared in ha.h (Task 2) but implementation is not tasked explicitly. Add to Task 6 ha.c work or as a separate small task if needed before integration testing.