cpp-scheduler-worker-dispatch-death

Answer

Three independent fixes: detached worker lifecycle, provider selection (inline key + flat-rate bucket), and chunk index time-ordering.

1. Detached worker launch (survives scheduler tick session end)

The session-scoped background process was SIGTERM'd when the scheduler tick session ended. Fix: launch through a /tmp launcher script using setsid nohup, write a pid file, and poll via kill -0 + log tail — never via the child of a session-bound fork.

// scheduler.cpp — detached worker dispatch
#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include <signal.h>
#include <cstdio>
#include <fstream>
#include <string>

struct WorkerHandle { std::string pid_file; std::string log_file; };

// 1) Materialize a launcher script in /tmp. The shell performs setsid+nohup,
//    so the worker is reparented to init and leaves the scheduler's session.
std::string write_launcher(const std::string& bin, const std::string& args,
                           const std::string& log_file, const std::string& pid_file) {
    std::string path = "/tmp/worker_launch_" + std::to_string(getpid()) + ".sh";
    std::ofstream out(path);
    out << "#!/bin/sh\n"
        << "exec setsid nohup '" << bin << "' " << args
        << " >'" << log_file << "' 2>&1 &\n"
        << "echo $! >'" << pid_file << "'\n";
    out.close();
    chmod(path.c_str(), 0755);
    return path;
}

// 2) Launch and forget. No kill on the tick session teardown can reach the
//    worker: its session id now differs from the scheduler's.
bool launch_detached(const std::string& launcher) {
    return system((launcher + " >/dev/null 2>&1").c_str()) == 0;
}

// 3) Poll: kill -0 for liveness, tail for output. Distinguish "died with
//    zero output" (crash) from "finished silently" (exit code 0, log nonempty).
enum class WorkerState { RUNNING, EXITED_OK, CRASHED };
WorkerState poll_worker(const WorkerHandle& h, std::string* tail) {
    pid_t pid = 0;
    std::ifstream pf(h.pid_file); pf >> pid;
    if (pid <= 0 || kill(pid, 0) != 0) {               // ESRCH or EPERM
        std::string log; { std::ifstream lf(h.log_file); lf >> log; }
        *tail = log;
        return log.empty() ? WorkerState::CRASHED : WorkerState::EXITED_OK;
    }
    char buf[4096]; size_t n = 0;
    FILE* f = popen(("tail -n 20 '" + h.log_file + "'").c_str(), "r");
    if (f) { n = fread(buf, 1, sizeof(buf) - 1, f); pclose(f); }
    buf[n] = '\0'; *tail = buf;
    return WorkerState::RUNNING;
}

The scheduler tick keeps a WorkerHandle map; each tick calls poll_worker and only relaunches when the worker is CRASHED or EXITED_OK with a non-empty completion marker in the log.

2. Provider selection: inline api_key, flat-rate worker bucket

Two fleet rules were violated: the worker ran on deepseek-foreman (foreman's own PAYG bucket) and on a provider block whose key resolves only via api_key_env — hermes resolves Bearer None at call time when the env var is absent in the detached session, yielding HTTP 401.

// provider.cpp — selection gate
struct Provider {
    std::string name;
    std::string api_key;       // INLINE key, non-empty
    std::string api_key_env;   // env indirection — unsafe for detached workers
    bool flat_rate_prepaid;    // fleet rule: never foreman's PAYG bucket
    std::string model;
};

bool usable_for_worker_dispatch(const Provider& p) {
    if (!p.flat_rate_prepaid)  return false;      // rule 1: prepaid bucket only
    if (p.api_key.empty())     return false;      // rule 2: INLINE key required
    return true;                                  // ollama-cloud / minimax-m3
}

Minimax provider block fix — add the inline key so resolution never falls back to Bearer None:

# providers.yaml
providers:
  minimax:
    base_url: https://api.minimax.chat/v1
    api_key: sk-minimax-inline-****      # INLINE — verified HTTP 200
    api_key_env: MINIMAX_API_KEY         # kept for compat, never sole source
    models: [minimax-m3]
  ollama-cloud:
    base_url: http://ollama-cloud:11434/v1
    api_key: sk-ollama-inline-****       # INLINE
    models: [minimax-m3]
    flat_rate_prepaid: true              # worker bucket

Dispatch now iterates usable_for_worker_dispatch providers in priority order and prefers ollama-cloud (prepaid, inline key, model pinned to minimax-m3).

3. Chunk index time-ordering under split batched inserts

Split batched inserts (60 rows → 54 chunks) dispatched per-row shuffled the chunk array, so expired_chunks prefix-scan assumed [oldest..newest] and downsample_candidates assumed newest-is-last — both broke. Fix: assign a monotonic seq before splitting, then sort by (bucket, end-time, seq) after insert.

// chunk_store.cpp — restore ordering invariants
struct Chunk {
    int64_t seq;        // monotonic insert sequence, assigned pre-split
    int64_t bucket;
    int64_t start_ns;   // min row timestamp
    int64_t end_ns;     // max row timestamp
};

// Called once per batched insert, after the split fan-out completes.
void restore_chunk_order(std::vector<Chunk>& chunks) {
    std::sort(chunks.begin(), chunks.end(),
              [](const Chunk& a, const Chunk& b) {
                  if (a.bucket != b.bucket) return a.bucket < b.bucket;
                  if (a.end_ns  != b.end_ns)  return a.end_ns  < b.end_ns;  // time-ordered
                  return a.seq   < b.seq;                                  // stable tiebreak
              });
    // Rebuild the prefix index in sorted order so:
    //   expired_chunks:      scan prefix while chunk.end_ns < cutoff
    //   downsample_candidates: newest chunk is chunks.back()
}

The single insert path now does: assign seq to all 60 rows → split into 54 chunks (chunks inherit min/max row ts) → restore_chunk_order → commit index. Any code that appends chunks individually (e.g. late-arriving rows) calls restore_chunk_order on the affected bucket too.