codebase-memory-mcp/src/cli/progress_sink.c

/*
 * progress_sink.c — Human-readable progress for one-shot CLI commands.
 *
 * Maps structured log events (msg=pass.timing, msg=pipeline.done, etc.)
 * to phase labels on stderr. Interactive terminals enable it automatically;
 * --progress forces it for redirected stderr. When installed, it replaces
 * default log output.
 *
 * Thread-safe: one sink mutex serializes shared summary state and each complete
 * output update, not merely the individual stdio calls.
 */
#include "progress_sink.h"
#include "foundation/compat_thread.h"
#include "foundation/constants.h"
#include "foundation/log.h"

#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

enum {
    PERCENT = 100,
    NOT_SET = -1,
    LOCK_UNINITIALIZED = 0,
    LOCK_INITIALIZING = 1,
    LOCK_READY = 2,
};

static FILE *s_out;
static atomic_int s_needs_newline;
static int s_gbuf_nodes = NOT_SET;
static int s_gbuf_edges = NOT_SET;
static CBMLogLevel s_previous_log_level = CBM_LOG_WARN;
static bool s_raised_log_level;
static bool s_preserve_details;
static cbm_mutex_t s_sink_mutex;
static atomic_int s_sink_mutex_state = ATOMIC_VAR_INIT(LOCK_UNINITIALIZED);

/* The CLI may install the sink more than once in one process (notably tests),
 * so keep one process-lifetime mutex rather than destroying it at fini. */
static void progress_sink_mutex_ensure(void) {
    int expected = LOCK_UNINITIALIZED;
    if (atomic_compare_exchange_strong_explicit(&s_sink_mutex_state, &expected, LOCK_INITIALIZING,
                                                memory_order_acq_rel, memory_order_acquire)) {
        cbm_mutex_init(&s_sink_mutex);
        atomic_store_explicit(&s_sink_mutex_state, LOCK_READY, memory_order_release);
        return;
    }
    while (atomic_load_explicit(&s_sink_mutex_state, memory_order_acquire) != LOCK_READY) {}
}

bool cbm_cli_output_flags_parse(int *argc, char **argv, cbm_cli_output_flags_t *flags, char *error,
                                size_t error_size) {
    if (!argc || !argv || !flags) {
        return false;
    }
    memset(flags, 0, sizeof(*flags));
    if (error && error_size > 0) {
        error[0] = '\0';
    }

    /* --quiet and --progress are reserved process controls wherever they
     * appear, matching the existing --progress behavior. --verbose remains
     * outer-only because tools such as index_status own a verbose input. */
    bool before_tool = true;
    int output_index = 0;
    for (int input_index = 0; input_index < *argc; input_index++) {
        char *arg = argv[input_index];
        bool quiet = arg && strcmp(arg, "--quiet") == 0;
        bool progress = arg && strcmp(arg, "--progress") == 0;
        bool verbose = before_tool && arg && strcmp(arg, "--verbose") == 0;
        if (quiet) {
            flags->quiet_requested = true;
        } else if (progress) {
            flags->progress_requested = true;
        } else if (verbose) {
            flags->verbose_requested = true;
        } else {
            argv[output_index++] = arg;
            if (before_tool && arg && arg[0] != '-') {
                before_tool = false;
            }
        }
    }
    *argc = output_index;

    if (flags->quiet_requested && (flags->progress_requested || flags->verbose_requested)) {
        if (error && error_size > 0) {
            (void)snprintf(error, error_size,
                           "--quiet cannot be combined with --progress or outer --verbose; "
                           "use --quiet for errors-only stderr, --progress for lifecycle "
                           "feedback, or --verbose for informational diagnostics");
        }
        return false;
    }
    return true;
}

void cbm_cli_diagnostics_configure(bool quiet_requested, bool verbose_requested) {
    if (quiet_requested) {
        /* Keep ERROR records visible while suppressing DEBUG/INFO/WARN. An
         * explicit CBM_LOG_LEVEL=none remains stronger than --quiet. */
        if (cbm_log_get_level() < CBM_LOG_ERROR) {
            cbm_log_set_level(CBM_LOG_ERROR);
        }
        return;
    }
    if (verbose_requested && cbm_log_get_level() > CBM_LOG_INFO) {
        cbm_log_set_level(CBM_LOG_INFO);
    }
}

bool cbm_cli_progress_enabled(bool explicitly_requested, bool quiet_requested, bool stderr_is_tty) {
    return !quiet_requested && (explicitly_requested || stderr_is_tty);
}

static void progress_tool_name(const char *tool_name, char out[CBM_SZ_64]) {
    size_t offset = 0;
    if (tool_name) {
        for (; tool_name[offset] && offset + 1 < CBM_SZ_64; offset++) {
            unsigned char ch = (unsigned char)tool_name[offset];
            bool safe = (ch >= 'a' && ch <= 'z') || (ch >= 'A' && ch <= 'Z') ||
                        (ch >= '0' && ch <= '9') || ch == '_' || ch == '-';
            if (!safe) {
                offset = 0;
                break;
            }
            out[offset] = (char)ch;
        }
    }
    if (offset == 0) {
        (void)snprintf(out, CBM_SZ_64, "%s", "tool");
    } else {
        out[offset] = '\0';
    }
}

void cbm_cli_progress_start(FILE *out, const char *tool_name) {
    FILE *stream = out ? out : stderr;
    char safe_name[CBM_SZ_64] = {0};
    progress_tool_name(tool_name, safe_name);
    (void)fprintf(stream, "Running %s locally...\n", safe_name);
    (void)fflush(stream);
}

void cbm_cli_progress_finish(FILE *out, const char *tool_name, bool success, uint64_t elapsed_ms) {
    FILE *stream = out ? out : stderr;
    char safe_name[CBM_SZ_64] = {0};
    progress_tool_name(tool_name, safe_name);
    (void)fprintf(stream, "%s %s (%llu ms)\n", success ? "Completed" : "Failed", safe_name,
                  (unsigned long long)elapsed_ms);
    (void)fflush(stream);
}

/* Extract one string field from the logger's compact JSON format. Keys are
 * accepted only at object boundaries so worker-controlled values cannot spoof
 * a progress event by merely containing a key-shaped substring. */
static const char *extract_json_field(const char *line, const char *key, char *buf, int buf_len) {
    char needle[CBM_SZ_64];
    int needle_len = snprintf(needle, sizeof(needle), "\"%s\":", key);
    if (needle_len <= 0 || needle_len >= (int)sizeof(needle)) {
        return NULL;
    }
    const char *candidate = line;
    while ((candidate = strstr(candidate, needle)) != NULL) {
        const char *before = candidate;
        while (before > line && (before[-1] == ' ' || before[-1] == '\t')) {
            before--;
        }
        if (before == line || (before[-1] != '{' && before[-1] != ',')) {
            candidate += needle_len;
            continue;
        }
        const char *value = candidate + needle_len;
        while (*value == ' ' || *value == '\t') {
            value++;
        }
        if (*value++ != '\"') {
            return NULL;
        }
        int offset = 0;
        bool escaped = false;
        while (*value && offset < buf_len - 1) {
            if (!escaped && *value == '\"') {
                buf[offset] = '\0';
                return buf;
            }
            if (!escaped && *value == '\\') {
                escaped = true;
                value++;
                continue;
            }
            if (escaped) {
                switch (*value) {
                case 'n':
                    buf[offset++] = '\n';
                    break;
                case 'r':
                    buf[offset++] = '\r';
                    break;
                case 't':
                    buf[offset++] = '\t';
                    break;
                default:
                    buf[offset++] = *value;
                    break;
                }
                escaped = false;
            } else {
                buf[offset++] = *value;
            }
            value++;
        }
        return NULL;
    }
    return NULL;
}

/* Extract a field from either structured text or JSON worker logs. */
static const char *extract_kv(const char *line, const char *key, char *buf, int buf_len) {
    if (!line || !key || !buf || buf_len <= 0) {
        return NULL;
    }
    size_t klen = strlen(key);
    const char *p = line;
    while (*p) {
        if ((p == line || p[-SKIP_ONE] == ' ') && strncmp(p, key, klen) == 0 && p[klen] == '=') {
            const char *val = p + klen + SKIP_ONE;
            int i = 0;
            while (val[i] && val[i] != ' ' && i < buf_len - SKIP_ONE) {
                buf[i] = val[i];
                i++;
            }
            buf[i] = '\0';
            return buf;
        }
        p++;
    }
    const char *json_key = strcmp(key, "msg") == 0 ? "event" : key;
    return extract_json_field(line, json_key, buf, buf_len);
}

void cbm_progress_sink_init(FILE *out) {
    progress_sink_mutex_ensure();
    cbm_mutex_lock(&s_sink_mutex);
    s_out = out ? out : stderr;
    atomic_store(&s_needs_newline, 0);
    s_gbuf_nodes = NOT_SET;
    s_gbuf_edges = NOT_SET;
    s_previous_log_level = cbm_log_get_level();
    s_preserve_details = s_previous_log_level <= CBM_LOG_INFO;
    s_raised_log_level = s_previous_log_level > CBM_LOG_INFO;
    if (s_raised_log_level) {
        /* Progress is an explicit INFO opt-in. The replacement sink below
         * keeps only recognized lifecycle events plus warnings/errors. */
        cbm_log_set_level(CBM_LOG_INFO);
    }
    cbm_log_set_sink(cbm_progress_sink_fn);
    cbm_mutex_unlock(&s_sink_mutex);
}

void cbm_progress_sink_fini(void) {
    progress_sink_mutex_ensure();
    cbm_mutex_lock(&s_sink_mutex);
    cbm_log_set_sink(NULL);
    if (atomic_load(&s_needs_newline) && s_out) {
        (void)fprintf(s_out, "\n");
        (void)fflush(s_out);
    }
    s_out = NULL;
    if (s_raised_log_level) {
        cbm_log_set_level(s_previous_log_level);
        s_raised_log_level = false;
    }
    s_preserve_details = false;
    cbm_mutex_unlock(&s_sink_mutex);
}

/* Phase label table: maps pass names to display labels. */
typedef struct {
    const char *pass;
    const char *label;
} phase_t;

static const phase_t phases[] = {
    {"parallel_extract", "[2/9] Extracting definitions"},
    {"registry_build", "[3/9] Building registry"},
    {"parallel_resolve", "[4/9] Resolving calls & edges"},
    {"tests", "[5/9] Detecting tests"},
    {"httplinks", "[6/9] Scanning HTTP links"},
    {"githistory_compute", "[7/9] Analyzing git history"},
    {"configlink", "[8/9] Linking config files"},
    {"dump", "[9/9] Writing database"},
};

enum { PHASE_COUNT = sizeof(phases) / sizeof(phases[0]) };

/* Flush pending \r line if needed. */
static void flush_carriage(void) {
    if (atomic_load(&s_needs_newline)) {
        (void)fprintf(s_out, "\n");
        atomic_store(&s_needs_newline, 0);
    }
}

/* Handle pipeline.discover event. */
static void on_discover(const char *line) {
    char files[CBM_SZ_32] = {0};
    if (extract_kv(line, "files", files, (int)sizeof(files))) {
        (void)fprintf(s_out, "  Discovering files (%s found)\n", files);
    } else {
        (void)fprintf(s_out, "  Discovering files...\n");
    }
    (void)fflush(s_out);
}

/* Handle pipeline.route event. */
static void on_route(const char *line) {
    char val[CBM_SZ_32] = {0};
    const char *path = extract_kv(line, "path", val, (int)sizeof(val));
    if (path && strcmp(path, "incremental") == 0) {
        (void)fprintf(s_out, "  Starting incremental index\n");
    } else {
        (void)fprintf(s_out, "  Starting full index\n");
    }
    (void)fflush(s_out);
}

/* Handle pass.start event. */
static void on_pass_start(const char *line) {
    char val[CBM_SZ_64] = {0};
    const char *pass = extract_kv(line, "pass", val, (int)sizeof(val));
    if (pass && strcmp(pass, "structure") == 0) {
        (void)fprintf(s_out, "[1/9] Building file structure\n");
        (void)fflush(s_out);
    }
}

/* Handle pass.timing event. */
static void on_pass_timing(const char *line) {
    char val[CBM_SZ_64] = {0};
    const char *pass = extract_kv(line, "pass", val, (int)sizeof(val));
    if (!pass) {
        return;
    }
    flush_carriage();
    for (int i = 0; i < PHASE_COUNT; i++) {
        if (strcmp(pass, phases[i].pass) == 0) {
            (void)fprintf(s_out, "%s\n", phases[i].label);
            (void)fflush(s_out);
            return;
        }
    }
}

/* Handle gbuf.dump event — capture node/edge counts. */
static void on_gbuf_dump(const char *line) {
    char n[CBM_SZ_32] = {0};
    char e[CBM_SZ_32] = {0};
    if (extract_kv(line, "nodes", n, (int)sizeof(n))) {
        s_gbuf_nodes = (int)strtol(n, NULL, CBM_DECIMAL_BASE);
    }
    if (extract_kv(line, "edges", e, (int)sizeof(e))) {
        s_gbuf_edges = (int)strtol(e, NULL, CBM_DECIMAL_BASE);
    }
}

/* Handle pipeline.done event. */
static void on_done(const char *line) {
    flush_carriage();
    char ms[CBM_SZ_32] = {0};
    const char *elapsed = extract_kv(line, "elapsed_ms", ms, (int)sizeof(ms));
    if (s_gbuf_nodes >= 0 && s_gbuf_edges >= 0 && elapsed) {
        (void)fprintf(s_out, "Done: %d nodes, %d edges (%s ms)\n", s_gbuf_nodes, s_gbuf_edges,
                      elapsed);
    } else if (s_gbuf_nodes >= 0 && s_gbuf_edges >= 0) {
        (void)fprintf(s_out, "Done: %d nodes, %d edges\n", s_gbuf_nodes, s_gbuf_edges);
    } else {
        (void)fprintf(s_out, "Done.\n");
    }
    (void)fflush(s_out);
}

/* Handle parallel.extract.progress event — in-place counter. */
static void on_extract_progress(const char *line) {
    char done[CBM_SZ_32] = {0};
    char total[CBM_SZ_32] = {0};
    if (extract_kv(line, "done", done, (int)sizeof(done)) &&
        extract_kv(line, "total", total, (int)sizeof(total))) {
        long d = strtol(done, NULL, CBM_DECIMAL_BASE);
        long t = strtol(total, NULL, CBM_DECIMAL_BASE);
        int pct = (t > 0) ? (int)((d * PERCENT) / t) : 0;
        (void)fprintf(s_out, "\r  Extracting: %ld/%ld files (%d%%)", d, t, pct);
        (void)fflush(s_out);
        atomic_store(&s_needs_newline, SKIP_ONE);
    }
}

/* Event dispatch table. */
typedef struct {
    const char *msg;
    void (*handler)(const char *line);
} event_handler_t;

static const event_handler_t handlers[] = {
    {"pipeline.discover", on_discover},
    {"pipeline.route", on_route},
    {"pass.start", on_pass_start},
    {"pass.timing", on_pass_timing},
    {"gbuf.dump", on_gbuf_dump},
    {"pipeline.done", on_done},
    {"parallel.extract.progress", on_extract_progress},
};

enum { HANDLER_COUNT = sizeof(handlers) / sizeof(handlers[0]) };

/* The progress sink replaces normal logger output. Preserve actionable
 * failures verbatim and, when INFO/DEBUG was already selected before progress
 * initialization, preserve unrecognized diagnostic lines as well. Logger
 * callbacks do not include a trailing newline. */
static void passthrough_line(const char *line) {
    flush_carriage();
    (void)fputs(line, s_out);
    size_t line_len = strlen(line);
    if (line_len == 0 || line[line_len - 1] != '\n') {
        (void)fputc('\n', s_out);
    }
    (void)fflush(s_out);
}

void cbm_progress_sink_fn(const char *line) {
    progress_sink_mutex_ensure();
    cbm_mutex_lock(&s_sink_mutex);
    if (!line || !s_out) {
        cbm_mutex_unlock(&s_sink_mutex);
        return;
    }
    char level[CBM_SZ_32] = {0};
    if (!extract_kv(line, "level", level, (int)sizeof(level))) {
        cbm_mutex_unlock(&s_sink_mutex);
        return;
    }
    if (strcmp(level, "warn") == 0 || strcmp(level, "error") == 0) {
        passthrough_line(line);
        cbm_mutex_unlock(&s_sink_mutex);
        return;
    }
    if (strcmp(level, "debug") == 0) {
        if (s_preserve_details) {
            passthrough_line(line);
        }
        cbm_mutex_unlock(&s_sink_mutex);
        return;
    }
    if (strcmp(level, "info") != 0) {
        cbm_mutex_unlock(&s_sink_mutex);
        return;
    }

    char msg[CBM_SZ_64] = {0};
    if (!extract_kv(line, "msg", msg, (int)sizeof(msg))) {
        cbm_mutex_unlock(&s_sink_mutex);
        return;
    }
    for (int i = 0; i < HANDLER_COUNT; i++) {
        if (strcmp(msg, handlers[i].msg) == 0) {
            handlers[i].handler(line);
            cbm_mutex_unlock(&s_sink_mutex);
            return;
        }
    }
    if (s_preserve_details) {
        passthrough_line(line);
    }
    cbm_mutex_unlock(&s_sink_mutex);
}