#define _GNU_SOURCE

#include <ctype.h>
#include <dirent.h>
#include <errno.h>
#include <fcntl.h>
#include <limits.h>
#include <signal.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <stdarg.h>
#include <string.h>
#include <sys/file.h>
#include <sys/resource.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <time.h>
#include <unistd.h>

#ifndef PATH_MAX
#define PATH_MAX 4096
#endif

#define SIS_PROTO_NOME "siscore-protocolo-objetos"
#define SIS_PROTO_VERSAO 1
#define SIS_MAX_ID 128
#define SIS_MAX_ENTITY 160
#define SIS_DEFAULT_INTERVALO 1
#define SIS_DEFAULT_LOTE 20000
#define SIS_DEFAULT_PARALELO 512
#define SIS_DEFAULT_TIMEOUT_ATOR 600
#define SIS_DEFAULT_TIMEOUT_PROCESSAMENTO 1800
#define SIS_DEFAULT_MAX_TENTATIVAS 3
#define SIS_DEFAULT_MAX_BYTES (10 * 1024 * 1024)
#define SIS_DEFAULT_RETER_EVENTOS 30
#define SIS_DEFAULT_RETER_ESTADO 90
#define SIS_DEFAULT_RETER_IDEMPOTENCIA 60
#define SIS_DEFAULT_INTERVALO_LIMPEZA 300
#define SIS_DEFAULT_LOTE_ATOR 100
#define SIS_MAX_LOTE 200000
#define SIS_MAX_PARALELO 10000
#define SIS_MAX_LOTE_ATOR 1000
#define SIS_ATOR_CACHE_MAX 2048
#define SIS_LIMPEZA_MAX_MARCADORES 2000

typedef enum {
    JSON_NULL,
    JSON_BOOL,
    JSON_NUMBER,
    JSON_STRING,
    JSON_ARRAY,
    JSON_OBJECT
} JsonType;

typedef struct Json Json;

typedef struct {
    char *key;
    Json *value;
} JsonPair;

struct Json {
    JsonType type;
    union {
        bool b;
        double n;
        char *s;
        struct { Json **items; size_t len; } array;
        struct { JsonPair *pairs; size_t len; } object;
    } u;
};

typedef struct {
    const char *p;
    const char *end;
    char erro[256];
} JsonParser;

typedef struct {
    char base[PATH_MAX];
    char espaco[PATH_MAX];
    int once;
    int init;
    int status;
    int intervalo;
    int lote;
    int paralelo;
    int timeout_ator;
    int timeout_processamento;
    int max_tentativas;
    size_t max_bytes;
    int reter_eventos;
    int reter_estado;
    int reter_idempotencia;
    int intervalo_limpeza;
    int lote_ator;
    int log_detalhado;
    int validar_marcador_api;
} Config;

typedef struct {
    bool valido;
    char mensagem_id[SIS_MAX_ID + 1];
    char processo_id[SIS_MAX_ID + 1];
    char origem[SIS_MAX_ENTITY + 1];
    char destino[SIS_MAX_ENTITY + 1];
    char tipo[32];
    char prioridade[16];
    int prioridade_peso;
    int numero_mensagem;
    bool sequencial_no_processo;
    time_t criado_ts;
    time_t nao_processar_antes_ts;
    time_t expira_ts;
    char resposta_a[SIS_MAX_ID + 1];
    char idempotencia[512];
    char idempotencia_escopo[SIS_MAX_ID + 1];
    char depende_de[64][SIS_MAX_ID + 1];
    int depende_count;
    char erro[512];
} MensagemInfo;

typedef struct {
    bool existe;
    bool ativo;
    bool tem_comando;
    char entidade[SIS_MAX_ENTITY + 1];
    char id[SIS_MAX_ID + 1];
    char tipo[16];
    char arquivo[PATH_MAX];
    char comando[2048];
    bool tem_comando_lote;
    char comando_lote[2048];
    int lote_max;
} Ator;

typedef struct {
    char arquivo[PATH_MAX];
    char destino[SIS_MAX_ENTITY + 1];
    char origem[SIS_MAX_ENTITY + 1];
    char processo_id[SIS_MAX_ID + 1];
    char mensagem_id[SIS_MAX_ID + 1];
} JobMensagem;

typedef struct {
    pid_t pid;
    time_t iniciado;
    char destino[SIS_MAX_ENTITY + 1];
    char origem[SIS_MAX_ENTITY + 1];
    char processo_id[SIS_MAX_ID + 1];
    char arquivo[PATH_MAX];
    char mensagem_id[SIS_MAX_ID + 1];
    char comando[4096];
    char lista_lote[PATH_MAX];
    int nmsgs;
    JobMensagem *msgs;
    bool timeout_enviado;
} Job;

typedef struct {
    const char *arquivo;
    char destino[SIS_MAX_ENTITY + 1];
    char origem[SIS_MAX_ENTITY + 1];
    char processo_id[SIS_MAX_ID + 1];
    char mensagem_id[SIS_MAX_ID + 1];
    int prioridade_peso;
    int numero_mensagem;
    time_t criado_ts;
} ItemAgendavel;

typedef struct {
    char **items;
    int len;
    int cap;
} ListaArquivos;

typedef struct {
    bool usado;
    char entidade[SIS_MAX_ENTITY + 1];
    Ator ator;
    time_t mtime;
    time_t checado;
} AtorCacheEntry;

static void die(const char *fmt, ...) {
    va_list ap;
    va_start(ap, fmt);
    vfprintf(stderr, fmt, ap);
    va_end(ap);
    fputc('\n', stderr);
    exit(1);
}

static void fmt_checked(char *out, size_t outsz, const char *fmt, ...) {
    va_list ap;
    va_start(ap, fmt);
    int n = vsnprintf(out, outsz, fmt, ap);
    va_end(ap);
    if (n < 0 || (size_t)n >= outsz) die("Texto/caminho excede limite interno.");
}

static void path_join(char *out, size_t outsz, const char *a, const char *b) {
    fmt_checked(out, outsz, "%s/%s", a, b);
}

static bool ends_with(const char *s, const char *suffix) {
    size_t ls = strlen(s), lf = strlen(suffix);
    return ls >= lf && strcmp(s + ls - lf, suffix) == 0;
}

static const char *base_name(const char *p) {
    const char *s = strrchr(p, '/');
    return s ? s + 1 : p;
}

static int raiz_sisc_de_base(const char *base, char *out, size_t outsz) {
    if (!base || !*base || !out || outsz == 0) return -1;
    char tmp[PATH_MAX];
    if (!realpath(base, tmp)) fmt_checked(tmp, sizeof tmp, "%s", base);
    char *slash = strrchr(tmp, '/');
    if (!slash || slash == tmp) return -1;
    *slash = '\0';
    fmt_checked(out, outsz, "%s", tmp);
    return 0;
}

static bool mkdir_p(const char *path) {
    char tmp[PATH_MAX];
    size_t len = strlen(path);
    if (len == 0 || len >= sizeof(tmp)) return false;
    memcpy(tmp, path, len + 1);
    if (tmp[len - 1] == '/') tmp[len - 1] = 0;
    for (char *p = tmp + 1; *p; p++) {
        if (*p == '/') {
            *p = 0;
            if (mkdir(tmp, 02775) == 0) {
                if (chmod(tmp, 02775) != 0 && errno != EPERM && errno != EACCES) return false;
            } else if (errno != EEXIST) return false;
            *p = '/';
        }
    }
    if (mkdir(tmp, 02775) == 0) {
        if (chmod(tmp, 02775) != 0 && errno != EPERM && errno != EACCES) return false;
    } else if (errno != EEXIST) return false;
    return true;
}

static bool is_regular_file(const char *path) {
    struct stat st;
    return stat(path, &st) == 0 && S_ISREG(st.st_mode);
}

static time_t file_mtime(const char *path) {
    struct stat st;
    if (stat(path, &st) != 0) return 0;
    return st.st_mtime;
}

static void now_iso(char *out, size_t outsz) {
    struct timespec ts;
    clock_gettime(CLOCK_REALTIME, &ts);
    struct tm tm;
    gmtime_r(&ts.tv_sec, &tm);
    snprintf(out, outsz, "%04d-%02d-%02dT%02d:%02d:%02d.%03ldZ",
             tm.tm_year + 1900, tm.tm_mon + 1, tm.tm_mday,
             tm.tm_hour, tm.tm_min, tm.tm_sec, ts.tv_nsec / 1000000L);
}

static void today_yyyymmdd(char *out, size_t outsz) {
    time_t t = time(NULL);
    struct tm tm;
    gmtime_r(&t, &tm);
    if (strftime(out, outsz, "%Y%m%d", &tm) == 0) die("Falha ao formatar data.");
}

static time_t parse_iso_utc(const char *s) {
    if (!s || !*s) return 0;
    int y = 0, mo = 0, d = 0, h = 0, mi = 0, se = 0;
    if (sscanf(s, "%d-%d-%dT%d:%d:%d", &y, &mo, &d, &h, &mi, &se) != 6) return 0;
    if (y < 1970 || mo < 1 || mo > 12 || d < 1 || d > 31 || h < 0 || h > 23 || mi < 0 || mi > 59 || se < 0 || se > 60) return 0;
    struct tm tm;
    memset(&tm, 0, sizeof tm);
    tm.tm_year = y - 1900;
    tm.tm_mon = mo - 1;
    tm.tm_mday = d;
    tm.tm_hour = h;
    tm.tm_min = mi;
    tm.tm_sec = se;
    tm.tm_isdst = 0;
    return timegm(&tm);
}

static int prioridade_peso(const char *s) {
    if (!s || !*s || strcmp(s, "normal") == 0) return 2;
    if (strcmp(s, "critica") == 0) return 0;
    if (strcmp(s, "alta") == 0) return 1;
    if (strcmp(s, "baixa") == 0) return 3;
    return 2;
}

static void json_escape(FILE *f, const char *s) {
    fputc('"', f);
    for (const unsigned char *p = (const unsigned char *)s; *p; p++) {
        switch (*p) {
            case '"': fputs("\\\"", f); break;
            case '\\': fputs("\\\\", f); break;
            case '\b': fputs("\\b", f); break;
            case '\f': fputs("\\f", f); break;
            case '\n': fputs("\\n", f); break;
            case '\r': fputs("\\r", f); break;
            case '\t': fputs("\\t", f); break;
            default:
                if (*p < 0x20) fprintf(f, "\\u%04x", *p);
                else fputc(*p, f);
        }
    }
    fputc('"', f);
}

static FILE *g_eventos_fp = NULL;
static char g_eventos_path[PATH_MAX];
static unsigned g_eventos_pendentes = 0;

static bool log_evento_suprimivel(const char *evento) {
    if (!evento) return false;
    const char *ruidosos[] = {
        "entregue", "ator_iniciado", "ator_lote_iniciado", "concluidas", "api_publicada_limpa",
        "api_publicada_validada_limpa", "tentativa_limpa", "indice_limpo",
        "mensagem_concluida_limpa", "marcador_transitorio_limpo"
    };
    for (size_t i = 0; i < sizeof(ruidosos)/sizeof(ruidosos[0]); i++) {
        if (strcmp(evento, ruidosos[i]) == 0) return true;
    }
    return false;
}

static FILE *log_event_fp(const Config *cfg) {
    char dia[16], dir[PATH_MAX], path[PATH_MAX];
    today_yyyymmdd(dia, sizeof dia);
    path_join(dir, sizeof dir, cfg->espaco, "eventos");
    mkdir_p(dir);
    fmt_checked(path, sizeof path, "%s/%s.jsonl", dir, dia);
    if (g_eventos_fp && strcmp(g_eventos_path, path) == 0) return g_eventos_fp;
    if (g_eventos_fp) {
        fflush(g_eventos_fp);
        fclose(g_eventos_fp);
        g_eventos_fp = NULL;
    }
    int fd = open(path, O_CREAT | O_WRONLY | O_APPEND, 0664);
    if (fd < 0) return NULL;
    fcntl(fd, F_SETFD, FD_CLOEXEC);
    g_eventos_fp = fdopen(fd, "a");
    if (!g_eventos_fp) { close(fd); return NULL; }
    setvbuf(g_eventos_fp, NULL, _IOFBF, 1024 * 1024);
    snprintf(g_eventos_path, sizeof g_eventos_path, "%s", path);
    g_eventos_pendentes = 0;
    return g_eventos_fp;
}

static void log_event_flush(void) {
    if (!g_eventos_fp) return;
    fflush(g_eventos_fp);
    g_eventos_pendentes = 0;
}

static void log_event_close(void) {
    if (!g_eventos_fp) return;
    fflush(g_eventos_fp);
    fclose(g_eventos_fp);
    g_eventos_fp = NULL;
    g_eventos_path[0] = 0;
    g_eventos_pendentes = 0;
}

static void log_event(const Config *cfg, const char *evento, const char *arquivo, const MensagemInfo *mi, const char *detalhe) {
    if (cfg && !cfg->log_detalhado && log_evento_suprimivel(evento)) return;
    char agora[64];
    FILE *f = log_event_fp(cfg);
    if (!f) return;
    now_iso(agora, sizeof agora);
    fputc('{', f);
    fputs("\"em\":", f); json_escape(f, agora);
    fputs(",\"evento\":", f); json_escape(f, evento ? evento : "");
    if (arquivo && *arquivo) { fputs(",\"arquivo\":", f); json_escape(f, arquivo); }
    if (mi) {
        if (mi->mensagem_id[0]) { fputs(",\"mensagemId\":", f); json_escape(f, mi->mensagem_id); }
        if (mi->processo_id[0]) { fputs(",\"processoId\":", f); json_escape(f, mi->processo_id); }
        if (mi->origem[0]) { fputs(",\"origem\":", f); json_escape(f, mi->origem); }
        if (mi->destino[0]) { fputs(",\"destino\":", f); json_escape(f, mi->destino); }
    }
    if (detalhe && *detalhe) { fputs(",\"detalhe\":", f); json_escape(f, detalhe); }
    fputs("}\n", f);
    g_eventos_pendentes++;
    if ((evento && (!strcmp(evento, "escuta_iniciada") || !strcmp(evento, "escuta_finalizada") || !strcmp(evento, "ciclo_concluido"))) || g_eventos_pendentes >= 256) {
        log_event_flush();
    }
}

static void *xcalloc(size_t n, size_t s) {
    void *p = calloc(n, s);
    if (!p) die("Memoria insuficiente.");
    return p;
}

static char *xstrdup(const char *s) {
    char *p = strdup(s ? s : "");
    if (!p) die("Memoria insuficiente.");
    return p;
}

static AtorCacheEntry g_ator_cache[SIS_ATOR_CACHE_MAX];
static volatile sig_atomic_t g_parar = 0;

static void sinalizar_parada(int sig) {
    (void)sig;
    g_parar = 1;
}

static Json *json_new(JsonType t) {
    Json *j = xcalloc(1, sizeof *j);
    j->type = t;
    return j;
}

static void json_free(Json *j) {
    if (!j) return;
    if (j->type == JSON_STRING) free(j->u.s);
    else if (j->type == JSON_ARRAY) {
        for (size_t i = 0; i < j->u.array.len; i++) json_free(j->u.array.items[i]);
        free(j->u.array.items);
    } else if (j->type == JSON_OBJECT) {
        for (size_t i = 0; i < j->u.object.len; i++) {
            free(j->u.object.pairs[i].key);
            json_free(j->u.object.pairs[i].value);
        }
        free(j->u.object.pairs);
    }
    free(j);
}

static void jp_error(JsonParser *p, const char *msg) {
    if (p->erro[0] == 0) snprintf(p->erro, sizeof p->erro, "%s", msg);
}

static void jp_ws(JsonParser *p) {
    while (p->p < p->end && isspace((unsigned char)*p->p)) p->p++;
}

static bool jp_lit(JsonParser *p, const char *lit) {
    size_t n = strlen(lit);
    if ((size_t)(p->end - p->p) >= n && strncmp(p->p, lit, n) == 0) { p->p += n; return true; }
    return false;
}

static char hex_digit_val(char c) {
    if (c >= '0' && c <= '9') return (char)(c - '0');
    if (c >= 'a' && c <= 'f') return (char)(10 + c - 'a');
    if (c >= 'A' && c <= 'F') return (char)(10 + c - 'A');
    return -1;
}

static char *jp_string(JsonParser *p) {
    if (p->p >= p->end || *p->p != '"') { jp_error(p, "string esperada"); return NULL; }
    p->p++;
    size_t cap = 64, len = 0;
    char *buf = xcalloc(cap, 1);
    while (p->p < p->end) {
        unsigned char c = (unsigned char)*p->p++;
        if (c == '"') { buf[len] = 0; return buf; }
        if (c < 0x20) { free(buf); jp_error(p, "controle invalido em string"); return NULL; }
        if (c == '\\') {
            if (p->p >= p->end) { free(buf); jp_error(p, "escape incompleto"); return NULL; }
            char e = *p->p++;
            switch (e) {
                case '"': c = '"'; break;
                case '\\': c = '\\'; break;
                case '/': c = '/'; break;
                case 'b': c = '\b'; break;
                case 'f': c = '\f'; break;
                case 'n': c = '\n'; break;
                case 'r': c = '\r'; break;
                case 't': c = '\t'; break;
                case 'u': {
                    if (p->end - p->p < 4) { free(buf); jp_error(p, "unicode escape incompleto"); return NULL; }
                    int ok = 1;
                    for (int i = 0; i < 4; i++) if (hex_digit_val(p->p[i]) < 0) ok = 0;
                    if (!ok) { free(buf); jp_error(p, "unicode escape invalido"); return NULL; }
                    p->p += 4;
                    c = '?';
                    break;
                }
                default: free(buf); jp_error(p, "escape invalido"); return NULL;
            }
        }
        if (len + 2 > cap) { cap *= 2; buf = realloc(buf, cap); if (!buf) die("Memoria insuficiente."); }
        buf[len++] = (char)c;
    }
    free(buf);
    jp_error(p, "string nao terminada");
    return NULL;
}

static Json *jp_value(JsonParser *p);

static Json *jp_array(JsonParser *p) {
    if (*p->p != '[') return NULL;
    p->p++;
    Json *a = json_new(JSON_ARRAY);
    jp_ws(p);
    if (p->p < p->end && *p->p == ']') { p->p++; return a; }
    while (p->p < p->end) {
        Json *v = jp_value(p);
        if (!v) { json_free(a); return NULL; }
        a->u.array.items = realloc(a->u.array.items, sizeof(Json *) * (a->u.array.len + 1));
        if (!a->u.array.items) die("Memoria insuficiente.");
        a->u.array.items[a->u.array.len++] = v;
        jp_ws(p);
        if (p->p < p->end && *p->p == ',') { p->p++; jp_ws(p); continue; }
        if (p->p < p->end && *p->p == ']') { p->p++; return a; }
        json_free(a); jp_error(p, "',' ou ']' esperado em array"); return NULL;
    }
    json_free(a); jp_error(p, "array nao terminado"); return NULL;
}

static Json *jp_object(JsonParser *p) {
    if (*p->p != '{') return NULL;
    p->p++;
    Json *o = json_new(JSON_OBJECT);
    jp_ws(p);
    if (p->p < p->end && *p->p == '}') { p->p++; return o; }
    while (p->p < p->end) {
        jp_ws(p);
        char *k = jp_string(p);
        if (!k) { json_free(o); return NULL; }
        jp_ws(p);
        if (p->p >= p->end || *p->p != ':') { free(k); json_free(o); jp_error(p, "':' esperado em objeto"); return NULL; }
        p->p++;
        Json *v = jp_value(p);
        if (!v) { free(k); json_free(o); return NULL; }
        o->u.object.pairs = realloc(o->u.object.pairs, sizeof(JsonPair) * (o->u.object.len + 1));
        if (!o->u.object.pairs) die("Memoria insuficiente.");
        o->u.object.pairs[o->u.object.len].key = k;
        o->u.object.pairs[o->u.object.len].value = v;
        o->u.object.len++;
        jp_ws(p);
        if (p->p < p->end && *p->p == ',') { p->p++; jp_ws(p); continue; }
        if (p->p < p->end && *p->p == '}') { p->p++; return o; }
        json_free(o); jp_error(p, "',' ou '}' esperado em objeto"); return NULL;
    }
    json_free(o); jp_error(p, "objeto nao terminado"); return NULL;
}

static Json *jp_number(JsonParser *p) {
    const char *start = p->p;
    if (*p->p == '-') p->p++;
    if (p->p >= p->end) { jp_error(p, "numero invalido"); return NULL; }
    if (*p->p == '0') p->p++;
    else if (isdigit((unsigned char)*p->p)) while (p->p < p->end && isdigit((unsigned char)*p->p)) p->p++;
    else { jp_error(p, "numero invalido"); return NULL; }
    if (p->p < p->end && *p->p == '.') {
        p->p++;
        if (p->p >= p->end || !isdigit((unsigned char)*p->p)) { jp_error(p, "numero invalido"); return NULL; }
        while (p->p < p->end && isdigit((unsigned char)*p->p)) p->p++;
    }
    if (p->p < p->end && (*p->p == 'e' || *p->p == 'E')) {
        p->p++;
        if (p->p < p->end && (*p->p == '+' || *p->p == '-')) p->p++;
        if (p->p >= p->end || !isdigit((unsigned char)*p->p)) { jp_error(p, "numero invalido"); return NULL; }
        while (p->p < p->end && isdigit((unsigned char)*p->p)) p->p++;
    }
    char tmp[128];
    size_t n = (size_t)(p->p - start);
    if (n >= sizeof tmp) { jp_error(p, "numero longo demais"); return NULL; }
    memcpy(tmp, start, n); tmp[n] = 0;
    Json *j = json_new(JSON_NUMBER);
    j->u.n = strtod(tmp, NULL);
    return j;
}

static Json *jp_value(JsonParser *p) {
    jp_ws(p);
    if (p->p >= p->end) { jp_error(p, "valor esperado"); return NULL; }
    char c = *p->p;
    if (c == '{') return jp_object(p);
    if (c == '[') return jp_array(p);
    if (c == '"') { Json *j = json_new(JSON_STRING); j->u.s = jp_string(p); if (!j->u.s) { free(j); return NULL; } return j; }
    if (c == 't') { if (jp_lit(p, "true")) { Json *j = json_new(JSON_BOOL); j->u.b = true; return j; } }
    if (c == 'f') { if (jp_lit(p, "false")) { Json *j = json_new(JSON_BOOL); j->u.b = false; return j; } }
    if (c == 'n') { if (jp_lit(p, "null")) return json_new(JSON_NULL); }
    if (c == '-' || isdigit((unsigned char)c)) return jp_number(p);
    jp_error(p, "valor JSON invalido");
    return NULL;
}

static Json *json_parse_text(const char *text, size_t len, char *erro, size_t errosz) {
    JsonParser p = { .p = text, .end = text + len, .erro = {0} };
    Json *j = jp_value(&p);
    if (!j) {
        snprintf(erro, errosz, "%s", p.erro[0] ? p.erro : "JSON invalido");
        return NULL;
    }
    jp_ws(&p);
    if (p.p != p.end) {
        json_free(j);
        snprintf(erro, errosz, "conteudo extra apos JSON");
        return NULL;
    }
    erro[0] = 0;
    return j;
}

static Json *json_obj_get(Json *o, const char *key) {
    if (!o || o->type != JSON_OBJECT) return NULL;
    for (size_t i = 0; i < o->u.object.len; i++) if (strcmp(o->u.object.pairs[i].key, key) == 0) return o->u.object.pairs[i].value;
    return NULL;
}

static const char *json_str(Json *o, const char *key) {
    Json *v = json_obj_get(o, key);
    return (v && v->type == JSON_STRING) ? v->u.s : NULL;
}

static bool json_bool_default(Json *o, const char *key, bool def) {
    Json *v = json_obj_get(o, key);
    return (v && v->type == JSON_BOOL) ? v->u.b : def;
}

static bool manifesto_biblioteca_permitido(Json *root, const char *tipo) {
    if (strcmp(tipo, "agente") == 0) return json_bool_default(root, "agenteFixo", false);
    if (strcmp(tipo, "conector") == 0) return json_bool_default(root, "conectorFixo", true);
    return false;
}

static bool rel_formato_conector(Json *manifesto, const char *id, char *out, size_t outsz, char *legado, size_t legadosz) {
    const char *decl = json_str(manifesto, "formatoConector");
    if (legado && legadosz) legado[0] = 0;
    if (!decl || !*decl) {
        fmt_checked(out, outsz, "conectores/%s/formatos/formato-%s.json", id, id);
        if (legado && legadosz) fmt_checked(legado, legadosz, "conectores/formatos/formato-%s.json", id);
        return true;
    }
    while (!strncmp(decl, "./", 2)) decl += 2;
    if (!*decl || decl[0] == '/' || strstr(decl, "..") || !ends_with(decl, ".json")) return false;
    if (strchr(decl, '/')) {
        if (strncmp(decl, "conectores/", 11) != 0) return false;
        fmt_checked(out, outsz, "%s", decl);
        return true;
    }
    fmt_checked(out, outsz, "conectores/%s/formatos/%s", id, decl);
    if (legado && legadosz) fmt_checked(legado, legadosz, "conectores/formatos/%s", decl);
    return true;
}

static bool validar_formato_conector(const Config *cfg, Json *manifesto, const char *id, bool biblioteca) {
    char rel[PATH_MAX], legado[PATH_MAX], path[PATH_MAX];
    if (!rel_formato_conector(manifesto, id, rel, sizeof rel, legado, sizeof legado)) return false;
    if (biblioteca) return true; /* pacotes da biblioteca são autocontidos; ao importar, o formato vai declarado no pacote. */
    path_join(path, sizeof path, cfg->base, rel);
    if (is_regular_file(path)) return true;
    if (legado[0]) {
        path_join(path, sizeof path, cfg->base, legado);
        if (is_regular_file(path)) return true;
    }
    return false;
}

static bool json_number_int(Json *o, const char *key, int *out) {
    Json *v = json_obj_get(o, key);
    if (!v || v->type != JSON_NUMBER) return false;
    *out = (int)v->u.n;
    return true;
}

static char *read_file(const char *path, size_t max_bytes, size_t *len_out, char *erro, size_t errosz) {
    struct stat st;
    if (stat(path, &st) != 0) { snprintf(erro, errosz, "stat falhou: %s", strerror(errno)); return NULL; }
    if (!S_ISREG(st.st_mode)) { snprintf(erro, errosz, "nao e arquivo regular"); return NULL; }
    if (st.st_size < 0 || (size_t)st.st_size > max_bytes) { snprintf(erro, errosz, "arquivo excede limite de bytes"); return NULL; }
    FILE *f = fopen(path, "rb");
    if (!f) { snprintf(erro, errosz, "fopen falhou: %s", strerror(errno)); return NULL; }
    char *buf = malloc((size_t)st.st_size + 1);
    if (!buf) die("Memoria insuficiente.");
    size_t n = fread(buf, 1, (size_t)st.st_size, f);
    fclose(f);
    if (n != (size_t)st.st_size) { free(buf); snprintf(erro, errosz, "leitura incompleta"); return NULL; }
    buf[n] = 0;
    *len_out = n;
    erro[0] = 0;
    return buf;
}

static bool valid_id(const char *s) {
    if (!s || !*s) return false;
    size_t n = strlen(s);
    if (n > SIS_MAX_ID) return false;
    if (!isalnum((unsigned char)s[0])) return false;
    for (size_t i = 1; i < n; i++) {
        unsigned char c = (unsigned char)s[i];
        if (!(isalnum(c) || c == '.' || c == '_' || c == '-')) return false;
    }
    return true;
}

static bool valid_entity(const char *s) {
    if (!s || strlen(s) > SIS_MAX_ENTITY) return false;
    const char *id = NULL;
    if (strncmp(s, "agente__", 8) == 0) id = s + 8;
    else if (strncmp(s, "conector__", 10) == 0) id = s + 10;
    else if (strncmp(s, "sistema__", 9) == 0) id = s + 9;
    else return false;
    return valid_id(id);
}

static void entity_parts(const char *entity, char *tipo, size_t tiposz, char *id, size_t idsz) {
    const char *sep = strstr(entity, "__");
    if (!sep) { tipo[0] = id[0] = 0; return; }
    snprintf(tipo, tiposz, "%.*s", (int)(sep - entity), entity);
    snprintf(id, idsz, "%s", sep + 2);
}

static void sanitize_file_id(const char *in, char *out, size_t outsz) {
    size_t j = 0;
    for (size_t i = 0; in && in[i] && j + 1 < outsz; i++) {
        unsigned char c = (unsigned char)in[i];
        out[j++] = (isalnum(c) || c == '.' || c == '_' || c == '-') ? (char)c : '-';
    }
    if (j == 0 && outsz > 1) out[j++] = 'x';
    out[j] = 0;
}

static void redirecionar_saida_ator_resposta(const Config *cfg, const MensagemInfo *mi) {
    if (!cfg || !mi || !mi->mensagem_id[0]) return;
    char dir[PATH_MAX], safe[256], path[PATH_MAX];
    path_join(dir, sizeof dir, cfg->espaco, "tmp/respostas");
    if (!mkdir_p(dir)) return;
    sanitize_file_id(mi->mensagem_id, safe, sizeof safe);
    fmt_checked(path, sizeof path, "%s/%s.out", dir, safe);
    int fd = open(path, O_CREAT | O_WRONLY | O_TRUNC, 0664);
    if (fd < 0) return;
    dup2(fd, STDOUT_FILENO);
    dup2(fd, STDERR_FILENO);
    close(fd);
}

static void mensagem_nome_arquivo(const MensagemInfo *mi, char *out, size_t outsz) {
    char destino[256], mensagem[256];
    sanitize_file_id(mi && mi->destino[0] ? mi->destino : "destino", destino, sizeof destino);
    sanitize_file_id(mi && mi->mensagem_id[0] ? mi->mensagem_id : "mensagem", mensagem, sizeof mensagem);
    fmt_checked(out, outsz, "%s__%s.json", destino, mensagem);
}

static void caixa_dir_destino(const Config *cfg, const char *destino, char *out, size_t outsz) {
    char base[PATH_MAX];
    path_join(base, sizeof base, cfg->espaco, "caixas");
    if (destino && *destino) {
        char safe[256];
        sanitize_file_id(destino, safe, sizeof safe);
        fmt_checked(out, outsz, "%s/%s", base, safe);
    } else {
        fmt_checked(out, outsz, "%s", base);
    }
    mkdir_p(out);
}

static void caixa_path_base_name(const Config *cfg, const char *destino, const char *nome, char *out, size_t outsz) {
    char dir[PATH_MAX];
    caixa_dir_destino(cfg, destino, dir, sizeof dir);
    fmt_checked(out, outsz, "%s/%s", dir, nome ? nome : "mensagem.json");
}

static void caixa_path_mensagem(const Config *cfg, const MensagemInfo *mi, char *out, size_t outsz) {
    char nome[512];
    mensagem_nome_arquivo(mi, nome, sizeof nome);
    caixa_path_base_name(cfg, mi ? mi->destino : "", nome, out, outsz);
}

static uint64_t fnv1a64(const char *s) {
    uint64_t h = 1469598103934665603ULL;
    for (const unsigned char *p = (const unsigned char *)s; p && *p; p++) { h ^= *p; h *= 1099511628211ULL; }
    return h;
}

static bool create_marker_exclusive(const char *path, const char *conteudo) {
    int fd = open(path, O_CREAT | O_EXCL | O_WRONLY, 0664);
    if (fd < 0) return false;
    if (conteudo) write(fd, conteudo, strlen(conteudo));
    close(fd);
    return true;
}

static bool copy_or_rename(const char *src, const char *dst) {
    if (rename(src, dst) == 0) return true;
    if (errno != EXDEV) return false;
    int in = open(src, O_RDONLY);
    if (in < 0) return false;
    int out = open(dst, O_CREAT | O_EXCL | O_WRONLY, 0664);
    if (out < 0) { close(in); return false; }
    char buf[65536];
    ssize_t r;
    bool ok = true;
    while ((r = read(in, buf, sizeof buf)) > 0) {
        char *p = buf;
        ssize_t left = r;
        while (left > 0) {
            ssize_t w = write(out, p, (size_t)left);
            if (w <= 0) { ok = false; break; }
            p += w; left -= w;
        }
        if (!ok) break;
    }
    if (r < 0) ok = false;
    fsync(out);
    close(out); close(in);
    if (ok) unlink(src); else unlink(dst);
    return ok;
}

static void dated_dir(const Config *cfg, const char *classe, char *out, size_t outsz) {
    char base[PATH_MAX], dia[16];
    path_join(base, sizeof base, cfg->espaco, classe);
    today_yyyymmdd(dia, sizeof dia);
    fmt_checked(out, outsz, "%s/%s", base, dia);
    mkdir_p(out);
}

static void estado_marker_path(const Config *cfg, const char *estado, const char *mensagem_id, char *out, size_t outsz) {
    char base[PATH_MAX], estado_dir[PATH_MAX], safe[256];
    path_join(base, sizeof base, cfg->espaco, "estado");
    path_join(estado_dir, sizeof estado_dir, base, estado);
    mkdir_p(estado_dir);
    sanitize_file_id(mensagem_id, safe, sizeof safe);
    fmt_checked(out, outsz, "%s/%s.idx", estado_dir, safe);
}

static bool estado_existe(const Config *cfg, const char *estado, const char *mensagem_id) {
    if (!mensagem_id || !*mensagem_id) return false;
    char p[PATH_MAX];
    estado_marker_path(cfg, estado, mensagem_id, p, sizeof p);
    return is_regular_file(p);
}

static void registrar_estado_terminal(const Config *cfg, const char *classe, const MensagemInfo *mi, const char *arquivo_final) {
    if (!mi || !mi->mensagem_id[0]) return;
    const char *estado = NULL;
    if (strcmp(classe, "concluidas") == 0) estado = "concluidas";
    else if (strcmp(classe, "erro") == 0 || strcmp(classe, "quarentena") == 0) estado = "terminais";
    if (!estado) return;
    char p[PATH_MAX], conteudo[1024];
    estado_marker_path(cfg, estado, mi->mensagem_id, p, sizeof p);
    snprintf(conteudo, sizeof conteudo, "classe=%s\narquivo=%s\nprocessoId=%s\ndestino=%s\n", classe, arquivo_final ? arquivo_final : "", mi->processo_id, mi->destino);
    if (!create_marker_exclusive(p, conteudo)) {
        int fd = open(p, O_WRONLY | O_APPEND);
        if (fd >= 0) { write(fd, conteudo, strlen(conteudo)); close(fd); }
    }
}

static void processo_numero_path(const Config *cfg, const char *processo_id, int numero, char *out, size_t outsz) {
    char base[PATH_MAX], procdir[PATH_MAX], safe[256];
    path_join(base, sizeof base, cfg->espaco, "processos");
    sanitize_file_id(processo_id, safe, sizeof safe);
    path_join(procdir, sizeof procdir, base, safe);
    mkdir_p(procdir);
    fmt_checked(out, outsz, "%s/%08d.idx", procdir, numero);
}

static bool arquivo_contem_linha_id(const char *path, const char *id) {
    FILE *f = fopen(path, "r");
    if (!f) return false;
    char linha[256];
    bool ok = false;
    while (fgets(linha, sizeof linha, f)) {
        if (strncmp(linha, "mensagemId=", 11) == 0) {
            char *v = linha + 11;
            v[strcspn(v, "\r\n")] = 0;
            ok = strcmp(v, id) == 0;
            break;
        }
    }
    fclose(f);
    return ok;
}

static bool registrar_numero_processo(const Config *cfg, const MensagemInfo *mi, char *erro, size_t errosz) {
    if (!mi || mi->numero_mensagem <= 0) return true;
    char p[PATH_MAX], conteudo[512];
    processo_numero_path(cfg, mi->processo_id, mi->numero_mensagem, p, sizeof p);
    snprintf(conteudo, sizeof conteudo, "mensagemId=%s\ndestino=%s\n", mi->mensagem_id, mi->destino);
    if (create_marker_exclusive(p, conteudo)) return true;
    if (arquivo_contem_linha_id(p, mi->mensagem_id)) return true;
    snprintf(erro, errosz, "numeroMensagem duplicado no processo");
    return false;
}

static int bloqueio_sequencial(const Config *cfg, const MensagemInfo *mi, char *motivo, size_t motivosz) {
    if (!mi || !mi->sequencial_no_processo || mi->numero_mensagem <= 1) return 0;
    for (int n = 1; n < mi->numero_mensagem; n++) {
        char p[PATH_MAX];
        processo_numero_path(cfg, mi->processo_id, n, p, sizeof p);
        FILE *f = fopen(p, "r");
        if (!f) {
            snprintf(motivo, motivosz, "aguardando numeroMensagem anterior %d do processo", n);
            return 1;
        }
        char linha[256], depid[SIS_MAX_ID + 1] = {0};
        while (fgets(linha, sizeof linha, f)) {
            if (strncmp(linha, "mensagemId=", 11) == 0) {
                size_t dn = strcspn(linha + 11, "\r\n");
                if (dn >= sizeof(depid)) dn = sizeof(depid) - 1;
                memcpy(depid, linha + 11, dn);
                depid[dn] = 0;
                break;
            }
        }
        fclose(f);
        if (!depid[0] || !estado_existe(cfg, "concluidas", depid)) {
            if (depid[0] && estado_existe(cfg, "terminais", depid)) {
                snprintf(motivo, motivosz, "numeroMensagem anterior %d terminou em erro", n);
                return 2;
            }
            snprintf(motivo, motivosz, "aguardando conclusao do numeroMensagem anterior %d", n);
            return 1;
        }
    }
    return 0;
}

static int bloqueio_dependencias(const Config *cfg, const MensagemInfo *mi, char *motivo, size_t motivosz) {
    if (!mi) return 0;
    for (int i = 0; i < mi->depende_count; i++) {
        const char *dep = mi->depende_de[i];
        if (estado_existe(cfg, "concluidas", dep)) continue;
        if (estado_existe(cfg, "terminais", dep)) {
            snprintf(motivo, motivosz, "dependencia %s terminou em erro/quarentena", dep);
            return 2;
        }
        snprintf(motivo, motivosz, "aguardando dependencia %s", dep);
        return 1;
    }
    return 0;
}

static void remover_auxiliar_mensagem(const Config *cfg, const MensagemInfo *mi, const char *rel_dir, const char *sufixo, const char *evento, const char *detalhe_ok) {
    if (!mi || !mi->mensagem_id[0]) return;
    char dir[PATH_MAX], safe[256], path[PATH_MAX];
    path_join(dir, sizeof dir, cfg->espaco, rel_dir);
    sanitize_file_id(mi->mensagem_id, safe, sizeof safe);
    fmt_checked(path, sizeof path, "%s/%s%s", dir, safe, sufixo ? sufixo : "");
    if (!is_regular_file(path)) return;
    if (unlink(path) == 0) {
        log_event(cfg, evento, path, mi, detalhe_ok ? detalhe_ok : "arquivo auxiliar removido");
    } else {
        char detalhe[256];
        snprintf(detalhe, sizeof detalhe, "falha ao remover arquivo auxiliar: %s", strerror(errno));
        log_event(cfg, "limpeza_auxiliar_falhou", path, mi, detalhe);
    }
}

static void remover_auxiliar_concluida(const Config *cfg, const MensagemInfo *mi, const char *rel_dir, const char *sufixo, const char *evento) {
    remover_auxiliar_mensagem(cfg, mi, rel_dir, sufixo, evento, "mensagem concluida; arquivo auxiliar removido");
}

static void limpar_lixo_mensagem_concluida(const Config *cfg, const MensagemInfo *mi) {
    remover_auxiliar_concluida(cfg, mi, "api-publicadas", ".idx", "api_publicada_limpa");
    remover_auxiliar_concluida(cfg, mi, "tentativas", ".txt", "tentativa_limpa");
    remover_auxiliar_concluida(cfg, mi, "indice", ".idx", "indice_limpo");
}

static bool diretorio_contem_nome_rec(const char *dir, const char *fragmento, int profundidade) {
    DIR *d = opendir(dir);
    if (!d) return false;
    struct dirent *e;
    bool achou = false;
    while ((e = readdir(d))) {
        if (e->d_name[0] == '.') continue;
        if (strstr(e->d_name, fragmento)) { achou = true; break; }
        if (profundidade <= 0) continue;
        char path[PATH_MAX];
        fmt_checked(path, sizeof path, "%s/%s", dir, e->d_name);
        struct stat st;
        if (stat(path, &st) == 0 && S_ISDIR(st.st_mode) && diretorio_contem_nome_rec(path, fragmento, profundidade - 1)) {
            achou = true;
            break;
        }
    }
    closedir(d);
    return achou;
}

static bool mensagem_ativa_existe(const Config *cfg, const char *mensagem_id) {
    if (!mensagem_id || !*mensagem_id) return false;
    char safe[256];
    sanitize_file_id(mensagem_id, safe, sizeof safe);
    const char *dirs[] = {"entrada", "caixas", "processando"};
    for (size_t i = 0; i < sizeof(dirs)/sizeof(dirs[0]); i++) {
        char dir[PATH_MAX];
        path_join(dir, sizeof dir, cfg->espaco, dirs[i]);
        if (diretorio_contem_nome_rec(dir, safe, 1)) return true;
    }
    return false;
}

static void limpar_marcadores_transitorios_orfaos(const Config *cfg, const char *rel_dir, const char *sufixo) {
    char dir[PATH_MAX];
    path_join(dir, sizeof dir, cfg->espaco, rel_dir);
    DIR *d = opendir(dir);
    if (!d) return;
    struct dirent *e;
    int visitados = 0;
    while ((e = readdir(d))) {
        if (visitados++ >= SIS_LIMPEZA_MAX_MARCADORES) break;
        if (e->d_name[0] == '.' || !ends_with(e->d_name, sufixo)) continue;
        char mid[SIS_MAX_ID + 1];
        size_t n = strlen(e->d_name) - strlen(sufixo);
        if (n >= sizeof mid) continue;
        memcpy(mid, e->d_name, n); mid[n] = 0;
        if (estado_existe(cfg, "concluidas", mid) || estado_existe(cfg, "terminais", mid) || !mensagem_ativa_existe(cfg, mid)) {
            char path[PATH_MAX];
            fmt_checked(path, sizeof path, "%s/%s", dir, e->d_name);
            if (unlink(path) == 0) log_event(cfg, "marcador_transitorio_limpo", path, NULL, rel_dir);
        }
    }
    closedir(d);
}

static void limpar_concluidas_materializadas(const Config *cfg) {
    char base[PATH_MAX];
    path_join(base, sizeof base, cfg->espaco, "concluidas");
    DIR *dias = opendir(base);
    if (!dias) return;
    struct dirent *dia;
    while ((dia = readdir(dias))) {
        if (dia->d_name[0] == '.') continue;
        char dir[PATH_MAX];
        fmt_checked(dir, sizeof dir, "%s/%s", base, dia->d_name);
        DIR *d = opendir(dir);
        if (!d) continue;
        struct dirent *e;
        while ((e = readdir(d))) {
            if (e->d_name[0] == '.' || !ends_with(e->d_name, ".json")) continue;
            const char *mid = strstr(e->d_name, "__");
            while (mid) {
                const char *prox = strstr(mid + 2, "__");
                if (!prox) break;
                mid = prox;
            }
            if (!mid) continue;
            mid += 2;
            char id[SIS_MAX_ID + 1];
            size_t n = strlen(mid) - 5;
            if (n >= sizeof id) continue;
            memcpy(id, mid, n); id[n] = 0;
            if (!estado_existe(cfg, "concluidas", id)) continue;
            char path[PATH_MAX];
            fmt_checked(path, sizeof path, "%s/%s", dir, e->d_name);
            if (unlink(path) == 0) log_event(cfg, "mensagem_concluida_limpa", path, NULL, "limpeza de conteúdo concluído legado");
        }
        closedir(d);
    }
    closedir(dias);
}

static void limpar_arquivos_antigos_em_diretorio(const Config *cfg, const char *rel_dir, const char *sufixo, int dias, const char *rotulo) {
    if (dias <= 0) return;
    char dir[PATH_MAX];
    path_join(dir, sizeof dir, cfg->espaco, rel_dir);
    DIR *d = opendir(dir);
    if (!d) return;
    time_t agora = time(NULL);
    time_t corte = agora - (time_t)dias * 86400;
    struct dirent *e;
    while ((e = readdir(d))) {
        if (e->d_name[0] == '.') continue;
        if (sufixo && sufixo[0] && !ends_with(e->d_name, sufixo)) continue;
        char path[PATH_MAX];
        fmt_checked(path, sizeof path, "%s/%s", dir, e->d_name);
        struct stat st;
        if (stat(path, &st) != 0 || !S_ISREG(st.st_mode)) continue;
        if (st.st_mtime >= corte) continue;
        if (unlink(path) == 0) {
            char detalhe[256];
            snprintf(detalhe, sizeof detalhe, "%s removido: %ld dias (corte=%ld)", rotulo, (long)((agora - st.st_mtime) / 86400), (long)dias);
            log_event(cfg, "estado_antigo_limpo", path, NULL, detalhe);
        }
    }
    closedir(d);
}

static void limpar_eventos_antigos(const Config *cfg) {
    limpar_arquivos_antigos_em_diretorio(cfg, "eventos", ".jsonl", cfg->reter_eventos, "evento-log");
}

static void limpar_estado_terminais_antigo(const Config *cfg) {
    limpar_arquivos_antigos_em_diretorio(cfg, "estado/terminais", ".idx", cfg->reter_estado, "estado-terminal");
    limpar_arquivos_antigos_em_diretorio(cfg, "estado/respondida", ".idx", cfg->reter_estado, "estado-respondida");
}

static void limpar_estado_concluidas_antigo(const Config *cfg) {
    if (cfg->reter_estado <= 0) return;
    char dir[PATH_MAX];
    path_join(dir, sizeof dir, cfg->espaco, "estado/concluidas");
    DIR *d = opendir(dir);
    if (!d) return;
    time_t agora = time(NULL);
    time_t corte = agora - (time_t)cfg->reter_estado * 86400;
    struct dirent *e;
    while ((e = readdir(d))) {
        if (e->d_name[0] == '.' || !ends_with(e->d_name, ".idx")) continue;
        char path[PATH_MAX];
        fmt_checked(path, sizeof path, "%s/%s", dir, e->d_name);
        struct stat st;
        if (stat(path, &st) != 0 || !S_ISREG(st.st_mode)) continue;
        if (st.st_mtime >= corte) continue;
        // Extrai mensagemId do nome do arquivo (msg-xxx.idx -> msg-xxx)
        size_t n = strlen(e->d_name) - 4;
        if (n >= SIS_MAX_ID) continue;
        char mid[SIS_MAX_ID + 1];
        memcpy(mid, e->d_name, n); mid[n] = 0;
        // Só remove se não houver mensagens ativas na entrada/caixas/processando
        // que possam depender desta (checagem rápida por nome de arquivo)
        if (mensagem_ativa_existe(cfg, mid)) continue;
        if (unlink(path) == 0) {
            char detalhe[256];
            snprintf(detalhe, sizeof detalhe, "concluida removida: %ld dias", (long)((agora - st.st_mtime) / 86400));
            log_event(cfg, "estado_concluida_limpo", path, NULL, detalhe);
        }
    }
    closedir(d);
}

static void limpar_processos_antigos(const Config *cfg) {
    if (cfg->reter_estado <= 0) return;
    char dir[PATH_MAX];
    path_join(dir, sizeof dir, cfg->espaco, "processos");
    DIR *d = opendir(dir);
    if (!d) return;
    time_t agora = time(NULL);
    time_t corte = agora - (time_t)cfg->reter_estado * 86400;
    struct dirent *e;
    while ((e = readdir(d))) {
        if (e->d_name[0] == '.') continue;
        char sub[PATH_MAX];
        fmt_checked(sub, sizeof sub, "%s/%s", dir, e->d_name);
        struct stat st_sub;
        if (stat(sub, &st_sub) != 0 || !S_ISDIR(st_sub.st_mode)) continue;
        // Lista arquivos .idx dentro do subdiretório de processo
        DIR *sd = opendir(sub);
        if (!sd) continue;
        bool tem_recente = false;
        struct dirent *f;
        while ((f = readdir(sd))) {
            if (f->d_name[0] == '.' || !ends_with(f->d_name, ".idx")) continue;
            char fp[PATH_MAX];
            fmt_checked(fp, sizeof fp, "%s/%s", sub, f->d_name);
            struct stat st_f;
            if (stat(fp, &st_f) == 0 && st_f.st_mtime >= corte) { tem_recente = true; break; }
        }
        closedir(sd);
        if (tem_recente) continue;
        // Remove todos os .idx antigos e o diretório se vazio
        sd = opendir(sub);
        if (!sd) continue;
        while ((f = readdir(sd))) {
            if (f->d_name[0] == '.' || !ends_with(f->d_name, ".idx")) continue;
            char fp[PATH_MAX];
            fmt_checked(fp, sizeof fp, "%s/%s", sub, f->d_name);
            if (unlink(fp) == 0) log_event(cfg, "processo_idx_limpo", fp, NULL, "processo antigo");
        }
        closedir(sd);
        if (rmdir(sub) == 0) log_event(cfg, "processo_dir_limpo", sub, NULL, "diretorio de processo antigo removido");
    }
    closedir(d);
}

static void limpar_vinculos_antigos(const Config *cfg) {
    limpar_arquivos_antigos_em_diretorio(cfg, "vinculos", ".idx", cfg->reter_estado, "vinculo");
    limpar_arquivos_antigos_em_diretorio(cfg, "vinculos", ".rev", cfg->reter_estado, "vinculo-rev");
}

static void limpar_idempotencia_antiga(const Config *cfg) {
    limpar_arquivos_antigos_em_diretorio(cfg, "idempotencia", ".idx", cfg->reter_idempotencia, "idempotencia");
}

static void limpar_residuos_espaco(const Config *cfg) {
    limpar_marcadores_transitorios_orfaos(cfg, "api-publicadas", ".idx");
    limpar_marcadores_transitorios_orfaos(cfg, "indice", ".idx");
    limpar_marcadores_transitorios_orfaos(cfg, "tentativas", ".txt");
    limpar_concluidas_materializadas(cfg);
    limpar_eventos_antigos(cfg);
    limpar_estado_concluidas_antigo(cfg);
    limpar_estado_terminais_antigo(cfg);
    limpar_processos_antigos(cfg);
    limpar_vinculos_antigos(cfg);
    limpar_idempotencia_antiga(cfg);
}

static bool move_to_class(const Config *cfg, const char *src, const char *classe, const char *prefix, MensagemInfo *mi, const char *motivo) {
    if (strcmp(classe, "concluidas") == 0) {
        registrar_estado_terminal(cfg, classe, mi, src);
        log_event(cfg, classe, src, mi, motivo);
        limpar_lixo_mensagem_concluida(cfg, mi);
        if (unlink(src) != 0 && errno != ENOENT) {
            char detalhe[256];
            snprintf(detalhe, sizeof detalhe, "falha ao remover conteudo concluido: %s", strerror(errno));
            log_event(cfg, "limpeza_concluida_falhou", src, mi, detalhe);
            return false;
        }
        log_event(cfg, "mensagem_concluida_limpa", src, mi, "conteudo concluido removido; estado/concluidas preservado");
        return true;
    }

    char dir[PATH_MAX], nome[512], dst[PATH_MAX];
    dated_dir(cfg, classe, dir, sizeof dir);
    if (mi && mi->destino[0] && mi->mensagem_id[0]) {
        mensagem_nome_arquivo(mi, nome, sizeof nome);
    } else {
        char safe[256];
        sanitize_file_id(base_name(src), safe, sizeof safe);
        fmt_checked(nome, sizeof nome, "%s.json", safe);
    }
    fmt_checked(dst, sizeof dst, "%s/%s%s", dir, prefix ? prefix : "", nome);
    if (is_regular_file(dst)) {
        char extra[64], sem_ext[PATH_MAX];
        snprintf(extra, sizeof extra, "-%ld-%d", (long)time(NULL), getpid());
        snprintf(sem_ext, sizeof sem_ext, "%.*s", (int)(strlen(dst) > 5 ? strlen(dst) - 5 : strlen(dst)), dst);
        fmt_checked(dst, sizeof dst, "%s%s.json", sem_ext, extra);
    }
    bool ok = copy_or_rename(src, dst);
    if (ok) {
        registrar_estado_terminal(cfg, classe, mi, dst);
        log_event(cfg, classe, dst, mi, motivo);
        if (strcmp(classe, "concluidas") == 0) {
            limpar_lixo_mensagem_concluida(cfg, mi);
            if (unlink(dst) == 0) log_event(cfg, "mensagem_concluida_limpa", dst, mi, "conteudo concluido removido; estado/concluidas preservado");
        }
    }
    return ok;
}

static void ensure_layout(const Config *cfg) {
    const char *dirs[] = {
        "tmp", "entrada", "caixas", "processando", "concluidas", "erro",
        "quarentena", "duplicadas", "locks", "eventos", "indice", "idempotencia", "api-publicadas", "tentativas", "estado", "estado/concluidas", "estado/terminais", "estado/respondida", "processos", "vinculos"
    };
    mkdir_p(cfg->espaco);
    for (size_t i = 0; i < sizeof(dirs)/sizeof(dirs[0]); i++) {
        char p[PATH_MAX]; path_join(p, sizeof p, cfg->espaco, dirs[i]); mkdir_p(p);
    }
}

static bool validar_tipo_msg(const char *s) {
    return s && (!strcmp(s, "solicitacao") || !strcmp(s, "resposta") || !strcmp(s, "evento") || !strcmp(s, "erro") || !strcmp(s, "comando") || !strcmp(s, "consulta"));
}

static bool validar_prioridade(const char *s) {
    return !s || !*s || !strcmp(s, "baixa") || !strcmp(s, "normal") || !strcmp(s, "alta") || !strcmp(s, "critica");
}

static MensagemInfo extrair_validar_mensagem(Json *root) {
    MensagemInfo mi;
    memset(&mi, 0, sizeof mi);
    if (!root || root->type != JSON_OBJECT) { snprintf(mi.erro, sizeof mi.erro, "raiz deve ser objeto JSON"); return mi; }
    Json *sis = json_obj_get(root, "_sistema");
    Json *proto = json_obj_get(root, "_protocolo");
    Json *payload = json_obj_get(root, "payload");
    if (!sis || sis->type != JSON_OBJECT) { snprintf(mi.erro, sizeof mi.erro, "_sistema ausente ou invalido"); return mi; }
    if (!proto || proto->type != JSON_OBJECT) { snprintf(mi.erro, sizeof mi.erro, "_protocolo ausente ou invalido"); return mi; }
    if (!payload || payload->type != JSON_OBJECT) { snprintf(mi.erro, sizeof mi.erro, "payload ausente ou nao objeto"); return mi; }
    const char *nome = json_str(proto, "nome");
    int versao = 0;
    if (!nome || strcmp(nome, SIS_PROTO_NOME) != 0) { snprintf(mi.erro, sizeof mi.erro, "_protocolo.nome invalido"); return mi; }
    if (!json_number_int(proto, "versao", &versao) || versao != SIS_PROTO_VERSAO) { snprintf(mi.erro, sizeof mi.erro, "_protocolo.versao invalida"); return mi; }
    const char *mid = json_str(proto, "mensagemId");
    const char *pid = json_str(proto, "processoId");
    const char *orig = json_str(proto, "origem");
    const char *dest = json_str(proto, "destino");
    const char *tipo = json_str(proto, "tipo");
    const char *prio = json_str(proto, "prioridade");
    const char *criado = json_str(proto, "criadoEm");
    if (!criado) criado = json_str(sis, "criadoEm");
    if (!valid_id(mid)) { snprintf(mi.erro, sizeof mi.erro, "mensagemId invalido"); return mi; }
    if (!valid_id(pid)) { snprintf(mi.erro, sizeof mi.erro, "processoId invalido"); return mi; }
    if (!valid_entity(orig)) { snprintf(mi.erro, sizeof mi.erro, "origem invalida"); return mi; }
    if (!valid_entity(dest)) { snprintf(mi.erro, sizeof mi.erro, "destino invalido"); return mi; }
    if (!validar_tipo_msg(tipo)) { snprintf(mi.erro, sizeof mi.erro, "tipo de mensagem invalido"); return mi; }
    if (!validar_prioridade(prio)) { snprintf(mi.erro, sizeof mi.erro, "prioridade invalida"); return mi; }
    snprintf(mi.mensagem_id, sizeof mi.mensagem_id, "%s", mid);
    snprintf(mi.processo_id, sizeof mi.processo_id, "%s", pid);
    snprintf(mi.origem, sizeof mi.origem, "%s", orig);
    snprintf(mi.destino, sizeof mi.destino, "%s", dest);
    snprintf(mi.tipo, sizeof mi.tipo, "%s", tipo);
    snprintf(mi.prioridade, sizeof mi.prioridade, "%s", prio ? prio : "normal");
    mi.prioridade_peso = prioridade_peso(mi.prioridade);
    mi.criado_ts = parse_iso_utc(criado);
    if (mi.criado_ts == 0) mi.criado_ts = time(NULL);
    int numero = 0;
    if (json_number_int(proto, "numeroMensagem", &numero) && numero > 0) mi.numero_mensagem = numero;
    mi.sequencial_no_processo = json_bool_default(proto, "sequencialNoProcesso", false);
    mi.nao_processar_antes_ts = parse_iso_utc(json_str(proto, "naoProcessarAntesDe"));
    mi.expira_ts = parse_iso_utc(json_str(proto, "expiraEm"));
    Json *ra = json_obj_get(proto, "respostaA");
    if (ra && ra->type == JSON_STRING) {
        if (!valid_id(ra->u.s)) { snprintf(mi.erro, sizeof mi.erro, "respostaA invalido"); return mi; }
        snprintf(mi.resposta_a, sizeof mi.resposta_a, "%s", ra->u.s);
    } else if (ra && ra->type != JSON_NULL) { snprintf(mi.erro, sizeof mi.erro, "respostaA deve ser string ou null"); return mi; }
    Json *deps = json_obj_get(proto, "dependeDe");
    if (deps) {
        if (deps->type != JSON_ARRAY) { snprintf(mi.erro, sizeof mi.erro, "dependeDe deve ser array"); return mi; }
        if (deps->u.array.len > 64) { snprintf(mi.erro, sizeof mi.erro, "dependeDe excede 64 itens"); return mi; }
        for (size_t i = 0; i < deps->u.array.len; i++) {
            Json *dep = deps->u.array.items[i];
            if (!dep || dep->type != JSON_STRING || !valid_id(dep->u.s)) { snprintf(mi.erro, sizeof mi.erro, "dependeDe contem id invalido"); return mi; }
            snprintf(mi.depende_de[mi.depende_count++], sizeof mi.depende_de[0], "%s", dep->u.s);
        }
    }
    Json *idem = json_obj_get(proto, "idempotencia");
    if (idem && idem->type == JSON_OBJECT) {
        const char *ch = json_str(idem, "chave");
        if (!ch || strlen(ch) < 8 || strlen(ch) >= sizeof(mi.idempotencia)) { snprintf(mi.erro, sizeof mi.erro, "idempotencia.chave ausente, curta ou longa demais"); return mi; }
        snprintf(mi.idempotencia, sizeof mi.idempotencia, "%s", ch);
        const char *escopo_idem = json_str(idem, "escopo");
        if (escopo_idem && *escopo_idem) {
            if (!valid_id(escopo_idem)) { snprintf(mi.erro, sizeof mi.erro, "idempotencia.escopo invalido"); return mi; }
            snprintf(mi.idempotencia_escopo, sizeof mi.idempotencia_escopo, "%s", escopo_idem);
        }
    }
    mi.valido = true;
    return mi;
}

static Ator carregar_ator(const Config *cfg, const char *entidade) {
    Ator a;
    memset(&a, 0, sizeof a);
    snprintf(a.entidade, sizeof a.entidade, "%s", entidade ? entidade : "");
    if (!valid_entity(entidade)) return a;
    entity_parts(entidade, a.tipo, sizeof a.tipo, a.id, sizeof a.id);
    if (strcmp(a.tipo, "sistema") == 0) {
        a.existe = true;
        a.ativo = true;
        return a;
    }
    char dir[PATH_MAX], nome[256];
    bool fallback_biblioteca = false;
    snprintf(nome, sizeof nome, "%s.json", a.id);
    if (strcmp(a.tipo, "agente") == 0) path_join(dir, sizeof dir, cfg->base, "agentes");
    else if (strcmp(a.tipo, "conector") == 0) path_join(dir, sizeof dir, cfg->base, "conectores");
    else return a;
    if (strcmp(a.tipo, "conector") == 0) {
        char sub[PATH_MAX];
        snprintf(sub, sizeof sub, "%s/%s.json", a.id, a.id);
        path_join(a.arquivo, sizeof a.arquivo, dir, sub);
    } else {
        path_join(a.arquivo, sizeof a.arquivo, dir, nome);
    }
    if (!is_regular_file(a.arquivo)) {
        if (strcmp(a.tipo, "conector") == 0) {
            path_join(a.arquivo, sizeof a.arquivo, dir, nome); /* compatibilidade com layout legado */
        }
        if (!is_regular_file(a.arquivo)) {
            char raiz_sisc[PATH_MAX], subdir[64];
            if (raiz_sisc_de_base(cfg->base, raiz_sisc, sizeof raiz_sisc) != 0) return a;
            snprintf(subdir, sizeof subdir, "%s", strcmp(a.tipo, "agente") == 0 ? "core/biblioteca-agentes-fixos" : "core/biblioteca-conectores");
            path_join(dir, sizeof dir, raiz_sisc, subdir);
            path_join(a.arquivo, sizeof a.arquivo, dir, nome);
            if (!is_regular_file(a.arquivo)) return a;
            fallback_biblioteca = true;
        }
    }
    a.existe = true;
    a.ativo = true;
    char erro[256]; size_t len = 0;
    char *txt = read_file(a.arquivo, cfg->max_bytes, &len, erro, sizeof erro);
    if (!txt) return a;
    Json *root = json_parse_text(txt, len, erro, sizeof erro);
    free(txt);
    if (!root) return a;
    if (fallback_biblioteca && !manifesto_biblioteca_permitido(root, a.tipo)) {
        json_free(root);
        memset(&a, 0, sizeof a);
        snprintf(a.entidade, sizeof a.entidade, "%s", entidade ? entidade : "");
        return a;
    }
    if (strcmp(a.tipo, "conector") == 0 && !validar_formato_conector(cfg, root, a.id, fallback_biblioteca)) {
        json_free(root);
        memset(&a, 0, sizeof a);
        snprintf(a.entidade, sizeof a.entidade, "%s", entidade ? entidade : "");
        return a;
    }
    a.ativo = json_bool_default(root, "ativo", true);
    const char *cmd = json_str(root, "comando");
    if (!cmd) cmd = json_str(root, "executavel");
    const char *cmd_lote = json_str(root, "comandoLote");
    if (!cmd_lote) cmd_lote = json_str(root, "executavelLote");
    if (!cmd_lote) cmd_lote = json_str(root, "handlerLerMensagens");
    int lote_max = 0;
    json_number_int(root, "loteMaximo", &lote_max);
    Json *controlador = json_obj_get(root, "controlador");
    if (controlador && controlador->type == JSON_OBJECT) {
        const char *c0 = json_str(controlador, "executavel");
        if (!c0) c0 = json_str(controlador, "comando");
        if (c0) cmd = c0;
        const char *cl0 = json_str(controlador, "executavelLote");
        if (!cl0) cl0 = json_str(controlador, "comandoLote");
        if (!cl0) cl0 = json_str(controlador, "handlerLerMensagens");
        if (cl0) cmd_lote = cl0;
        int lm0 = 0;
        if (json_number_int(controlador, "loteMaximo", &lm0)) lote_max = lm0;
        if (!json_bool_default(controlador, "ativo", true)) a.ativo = false;
    }
    Json *exec = json_obj_get(root, "execucao");
    if (exec && exec->type == JSON_OBJECT) {
        const char *c2 = json_str(exec, "comando");
        if (!c2) c2 = json_str(exec, "executavel");
        if (c2) cmd = c2;
        const char *cl2 = json_str(exec, "executavelLote");
        if (!cl2) cl2 = json_str(exec, "comandoLote");
        if (!cl2) cl2 = json_str(exec, "handlerLerMensagens");
        if (cl2) cmd_lote = cl2;
        int lm2 = 0;
        if (json_number_int(exec, "loteMaximo", &lm2)) lote_max = lm2;
        if (!json_bool_default(exec, "ativo", true)) a.ativo = false;
    }
    if (cmd_lote && strlen(cmd_lote) > 0 && strlen(cmd_lote) < sizeof(a.comando_lote)) {
        a.tem_comando_lote = true;
        snprintf(a.comando_lote, sizeof a.comando_lote, "%s", cmd_lote);
        if (lote_max > 0) a.lote_max = lote_max;
    }
    if (cmd && strlen(cmd) > 0 && strlen(cmd) < sizeof(a.comando)) {
        a.tem_comando = true;
        snprintf(a.comando, sizeof a.comando, "%s", cmd);
    }
    json_free(root);
    return a;
}

static Ator carregar_ator_cacheado(const Config *cfg, const char *entidade) {
    if (!entidade || !*entidade) return carregar_ator(cfg, entidade);
    uint64_t h = fnv1a64(entidade);
    AtorCacheEntry *e = &g_ator_cache[h % SIS_ATOR_CACHE_MAX];
    time_t agora = time(NULL);
    if (e->usado && strcmp(e->entidade, entidade) == 0) {
        if (agora == e->checado) return e->ator;
        time_t mt = e->ator.arquivo[0] ? file_mtime(e->ator.arquivo) : 0;
        e->checado = agora;
        if (mt == e->mtime) return e->ator;
    }
    Ator a = carregar_ator(cfg, entidade);
    memset(e, 0, sizeof *e);
    e->usado = true;
    snprintf(e->entidade, sizeof e->entidade, "%s", entidade);
    e->ator = a;
    e->mtime = a.arquivo[0] ? file_mtime(a.arquivo) : 0;
    e->checado = agora;
    return a;
}

static bool mensagem_publicada_por_api(const Config *cfg, Json *root, const MensagemInfo *mi, char *motivo, size_t motivosz) {
    Json *sis = json_obj_get(root, "_sistema");
    Json *payload = json_obj_get(root, "payload");
    Json *api = payload && payload->type == JSON_OBJECT ? json_obj_get(payload, "_api") : NULL;
    const char *transporte = sis && sis->type == JSON_OBJECT ? json_str(sis, "transporte") : NULL;
    const char *regra = api && api->type == JSON_OBJECT ? json_str(api, "regraComunicacao") : NULL;
    const char *formato = api && api->type == JSON_OBJECT ? json_str(api, "formatoEscuta") : NULL;
    if (!transporte || strcmp(transporte, "pp --api") != 0) {
        snprintf(motivo, motivosz, "mensagem recusada: publicacao direta no espaco proibida; use pp --api");
        return false;
    }
    if (!regra || strcmp(regra, "somente-espaco-sisc-sem-acesso-direto") != 0) {
        snprintf(motivo, motivosz, "mensagem recusada: payload._api.regraComunicacao ausente/invalida; use pp --api");
        return false;
    }
    if (!formato || strcmp(formato, "formatos-escuta/protocolo-mensagem-siscore.v1.json") != 0) {
        snprintf(motivo, motivosz, "mensagem recusada: formatoEscuta canonico ausente/invalido; use pp --api");
        return false;
    }
    if (!mi || !mi->mensagem_id[0]) {
        snprintf(motivo, motivosz, "mensagem recusada: mensagemId ausente para validacao de publicacao via API");
        return false;
    }
    char dir[PATH_MAX], safe[256], marker[PATH_MAX];
    path_join(dir, sizeof dir, cfg->espaco, "api-publicadas");
    sanitize_file_id(mi->mensagem_id, safe, sizeof safe);
    fmt_checked(marker, sizeof marker, "%s/%s.idx", dir, safe);
    if (!is_regular_file(marker)) {
        snprintf(motivo, motivosz, "mensagem recusada: sem registro em espaco/api-publicadas; somente pp --api pode publicar");
        return false;
    }
    if (!cfg->validar_marcador_api) return true;

    char erro[256]; size_t len = 0;
    char *m = read_file(marker, cfg->max_bytes, &len, erro, sizeof erro);
    if (!m) {
        snprintf(motivo, motivosz, "mensagem recusada: marcador api-publicadas ilegivel");
        return false;
    }
    char linha_mid[256], linha_dest[256];
    snprintf(linha_mid, sizeof linha_mid, "mensagemId=%s", mi->mensagem_id);
    snprintf(linha_dest, sizeof linha_dest, "destino=%s", mi->destino);
    bool ok = strstr(m, "publicadoPor=pp --api") && strstr(m, "regraComunicacao=somente-espaco-sisc-sem-acesso-direto") && strstr(m, linha_mid) && strstr(m, linha_dest);
    free(m);
    if (!ok) {
        snprintf(motivo, motivosz, "mensagem recusada: marcador api-publicadas nao confere com mensagem; use pp --api");
        return false;
    }
    return true;
}

static int comparar_paths_lista(const void *pa, const void *pb) {
    const char *a = *(char * const *)pa;
    const char *b = *(char * const *)pb;
    return strcmp(a, b);
}

static bool lista_arquivos_cheia(const ListaArquivos *l, int max) {
    return max > 0 && l->len >= max;
}

static bool lista_arquivos_add(ListaArquivos *l, const char *path, int max) {
    if (lista_arquivos_cheia(l, max)) return false;
    if (l->len >= l->cap) {
        int novo = l->cap > 0 ? l->cap * 2 : 1024;
        if (max > 0 && novo > max) novo = max;
        char **nv = realloc(l->items, sizeof(char *) * (size_t)novo);
        if (!nv) die("Memoria insuficiente.");
        l->items = nv;
        l->cap = novo;
    }
    l->items[l->len++] = xstrdup(path);
    return true;
}

static void listar_json_dir(const char *dir, ListaArquivos *out, int max, int profundidade) {
    if (lista_arquivos_cheia(out, max)) return;
    DIR *d = opendir(dir);
    if (!d) return;
    struct dirent *e;
    while ((e = readdir(d))) {
        if (e->d_name[0] == '.') continue;
        char path[PATH_MAX];
        fmt_checked(path, sizeof path, "%s/%s", dir, e->d_name);
        struct stat st;
        if (stat(path, &st) != 0) continue;
        if (S_ISREG(st.st_mode)) {
            if (!ends_with(e->d_name, ".json")) continue;
            if (strstr(e->d_name, ".tmp")) continue;
            if (!lista_arquivos_add(out, path, max)) break;
        } else if (profundidade > 0 && S_ISDIR(st.st_mode)) {
            listar_json_dir(path, out, max, profundidade - 1);
            if (lista_arquivos_cheia(out, max)) break;
        }
    }
    closedir(d);
}

static ListaArquivos listar_json(const char *dir, int max) {
    ListaArquivos out;
    memset(&out, 0, sizeof out);
    listar_json_dir(dir, &out, max, 1);
    if (out.len > 1) qsort(out.items, (size_t)out.len, sizeof(char *), comparar_paths_lista);
    return out;
}

static void lista_arquivos_free(ListaArquivos *l) {
    if (!l) return;
    for (int i = 0; i < l->len; i++) free(l->items[i]);
    free(l->items);
    memset(l, 0, sizeof *l);
}

static bool processar_entrada(const Config *cfg, const char *arquivo) {
    MensagemInfo mi;
    memset(&mi, 0, sizeof mi);
    char claim[PATH_MAX];
    fmt_checked(claim, sizeof claim, "%s/processando/roteando-%ld-%d-%s", cfg->espaco, (long)time(NULL), getpid(), base_name(arquivo));
    if (rename(arquivo, claim) != 0) return false;

    char erro[512]; size_t len = 0;
    char *txt = read_file(claim, cfg->max_bytes, &len, erro, sizeof erro);
    if (!txt) {
        snprintf(mi.erro, sizeof mi.erro, "%s", erro);
        move_to_class(cfg, claim, "quarentena", "read-", &mi, mi.erro);
        return true;
    }
    Json *root = json_parse_text(txt, len, erro, sizeof erro);
    free(txt);
    if (!root) {
        snprintf(mi.erro, sizeof mi.erro, "%s", erro);
        move_to_class(cfg, claim, "quarentena", "json-", &mi, mi.erro);
        return true;
    }
    mi = extrair_validar_mensagem(root);
    if (!mi.valido) {
        json_free(root);
        move_to_class(cfg, claim, "quarentena", "protocolo-", &mi, mi.erro);
        return true;
    }
    char motivo_api[512];
    if (!mensagem_publicada_por_api(cfg, root, &mi, motivo_api, sizeof motivo_api)) {
        json_free(root);
        move_to_class(cfg, claim, "quarentena", "api-", &mi, motivo_api);
        return true;
    }
    json_free(root);

    Ator origem = carregar_ator_cacheado(cfg, mi.origem);
    Ator destino = carregar_ator_cacheado(cfg, mi.destino);
    if (!origem.existe || !origem.ativo) {
        move_to_class(cfg, claim, "quarentena", "origem-", &mi, "origem inexistente ou inativa");
        return true;
    }
    if (!destino.existe || !destino.ativo) {
        move_to_class(cfg, claim, "quarentena", "destino-", &mi, "destino inexistente ou inativo");
        return true;
    }

    char idxdir[PATH_MAX], idx[PATH_MAX], idxcontent[512];
    path_join(idxdir, sizeof idxdir, cfg->espaco, "indice");
    fmt_checked(idx, sizeof idx, "%s/%s.idx", idxdir, mi.mensagem_id);
    snprintf(idxcontent, sizeof idxcontent, "destino=%s\nprocessoId=%s\nmensagemId=%s\n", mi.destino, mi.processo_id, mi.mensagem_id);
    if (estado_existe(cfg, "concluidas", mi.mensagem_id) || estado_existe(cfg, "terminais", mi.mensagem_id)) {
        move_to_class(cfg, claim, "duplicadas", "msgid-", &mi, "mensagemId ja finalizado anteriormente");
        return true;
    }
    if (!create_marker_exclusive(idx, idxcontent)) {
        move_to_class(cfg, claim, "duplicadas", "msgid-", &mi, "mensagemId ja registrado");
        return true;
    }

    if (mi.idempotencia[0]) {
        char idemdir[PATH_MAX], idem[PATH_MAX], comb[1200];
        path_join(idemdir, sizeof idemdir, cfg->espaco, "idempotencia");
        const char *escopo_idem = mi.idempotencia_escopo[0] ? mi.idempotencia_escopo : mi.processo_id;
        snprintf(comb, sizeof comb, "%s|%s|%s|%s", mi.origem, mi.destino, escopo_idem, mi.idempotencia);
        fmt_checked(idem, sizeof idem, "%s/%016llx.idx", idemdir, (unsigned long long)fnv1a64(comb));
        if (!create_marker_exclusive(idem, idxcontent) && !arquivo_contem_linha_id(idem, mi.mensagem_id)) {
            move_to_class(cfg, claim, "duplicadas", "idem-", &mi, "idempotencyKey ja registrada neste escopo");
            return true;
        }
    }

    char erro_ordem[256];
    if (!registrar_numero_processo(cfg, &mi, erro_ordem, sizeof erro_ordem)) {
        move_to_class(cfg, claim, "quarentena", "ordem-", &mi, erro_ordem);
        return true;
    }

    if (mi.resposta_a[0]) {
        char pai_idx[PATH_MAX];
        fmt_checked(pai_idx, sizeof pai_idx, "%s/%s.idx", idxdir, mi.resposta_a);
        if (!is_regular_file(pai_idx) && !estado_existe(cfg, "concluidas", mi.resposta_a)) {
            move_to_class(cfg, claim, "quarentena", "respostaA-", &mi, "respostaA referencia mensagem inexistente ou nao concluida");
            return true;
        }
        char vdir[PATH_MAX], vinculo[PATH_MAX];
        path_join(vdir, sizeof vdir, cfg->espaco, "vinculos");
        mkdir_p(vdir);
        fmt_checked(vinculo, sizeof vinculo, "%s/%s__%s.idx", vdir, mi.resposta_a, mi.mensagem_id);
        if (!create_marker_exclusive(vinculo, "")) {
            log_event(cfg, "vinculo_ja_registrado", claim, &mi, "relacao pai-filho ja existe");
        }
        char vinculo_rev[PATH_MAX];
        fmt_checked(vinculo_rev, sizeof vinculo_rev, "%s/%s__%s.rev", vdir, mi.mensagem_id, mi.resposta_a);
        create_marker_exclusive(vinculo_rev, "");
        char resp_path[PATH_MAX];
        estado_marker_path(cfg, "respondida", mi.resposta_a, resp_path, sizeof resp_path);
        if (!create_marker_exclusive(resp_path, idxcontent)) {
            log_event(cfg, "mensagem_ja_respondida", claim, &mi, "original ja marcada como respondida");
        }
        log_event(cfg, "resposta_vinculada", claim, &mi, mi.resposta_a);
    }

    char destpath[PATH_MAX];
    caixa_path_mensagem(cfg, &mi, destpath, sizeof destpath);
    if (is_regular_file(destpath)) {
        move_to_class(cfg, claim, "duplicadas", "caixa-", &mi, "arquivo ja existe na caixa do destino");
        return true;
    }
    if (!copy_or_rename(claim, destpath)) {
        move_to_class(cfg, claim, "erro", "entrega-", &mi, "falha ao entregar na caixa do destino");
        return true;
    }
    log_event(cfg, "entregue", destpath, &mi, "mensagem validada e entregue na caixa do destino");
    remover_auxiliar_mensagem(cfg, &mi, "api-publicadas", ".idx", "api_publicada_validada_limpa", "mensagem validada e entregue; marcador de publicacao removido");
    return true;
}

static int rotear_entradas(const Config *cfg) {
    char entrada[PATH_MAX]; path_join(entrada, sizeof entrada, cfg->espaco, "entrada");
    ListaArquivos files = listar_json(entrada, cfg->lote);
    int proc = 0;
    for (int i = 0; i < files.len && !g_parar; i++) if (processar_entrada(cfg, files.items[i])) proc++;
    lista_arquivos_free(&files);
    return proc;
}

static int tentativa_get(const Config *cfg, const char *mid) {
    char dir[PATH_MAX], p[PATH_MAX]; path_join(dir, sizeof dir, cfg->espaco, "tentativas"); fmt_checked(p, sizeof p, "%s/%s.txt", dir, mid);
    FILE *f = fopen(p, "r"); if (!f) return 0;
    int n = 0; fscanf(f, "%d", &n); fclose(f); return n;
}

static int tentativa_inc(const Config *cfg, const char *mid) {
    char dir[PATH_MAX], p[PATH_MAX]; path_join(dir, sizeof dir, cfg->espaco, "tentativas"); mkdir_p(dir); fmt_checked(p, sizeof p, "%s/%s.txt", dir, mid);
    int n = tentativa_get(cfg, mid) + 1;
    FILE *f = fopen(p, "w"); if (f) { fprintf(f, "%d\n", n); fclose(f); }
    return n;
}

static bool extrair_info_arquivo(const Config *cfg, const char *arquivo, MensagemInfo *mi) {
    memset(mi, 0, sizeof *mi);
    char erro[256]; size_t len = 0; char *txt = read_file(arquivo, cfg->max_bytes, &len, erro, sizeof erro);
    if (!txt) { snprintf(mi->erro, sizeof mi->erro, "%s", erro); return false; }
    Json *root = json_parse_text(txt, len, erro, sizeof erro); free(txt);
    if (!root) { snprintf(mi->erro, sizeof mi->erro, "%s", erro); return false; }
    *mi = extrair_validar_mensagem(root); json_free(root);
    return mi->valido;
}

static void shell_quote(const char *in, char *out, size_t outsz) {
    size_t j = 0;
    if (j + 1 >= outsz) die("quote excede limite interno.");
    out[j++] = '\'';
    for (size_t i = 0; in && in[i]; i++) {
        if (in[i] == '\'') {
            const char *q = "'\\''";
            size_t qn = strlen(q);
            if (j + qn >= outsz) die("quote excede limite interno.");
            memcpy(out + j, q, qn);
            j += qn;
        } else {
            if (j + 1 >= outsz) die("quote excede limite interno.");
            out[j++] = in[i];
        }
    }
    if (j + 1 >= outsz) die("quote excede limite interno.");
    out[j++] = '\'';
    out[j] = 0;
}

static void montar_comando_acionamento(const Ator *ator, const char *arquivo, char *out, size_t outsz) {
    char arquivo_q[PATH_MAX * 2];
    shell_quote(arquivo, arquivo_q, sizeof arquivo_q);
    if (strcmp(ator->tipo, "conector") == 0) {
        fmt_checked(out, outsz, "%s ler-mensagem %s", ator->comando, arquivo_q);
    } else {
        fmt_checked(out, outsz, "%s --ler-mensagem %s", ator->comando, arquivo_q);
    }
}

static bool comando_exec_simples(const char *cmd) {
    if (!cmd || !*cmd) return false;
    for (const unsigned char *p = (const unsigned char *)cmd; *p; p++) {
        if (!(isalnum(*p) || *p == '/' || *p == '.' || *p == '_' || *p == '-' || *p == '+')) return false;
    }
    return true;
}

static void exec_acionamento_ator(const Ator *ator, const char *arquivo, const char *comando_shell) {
    if (ator && comando_exec_simples(ator->comando)) {
        if (strcmp(ator->tipo, "conector") == 0) {
            execlp(ator->comando, ator->comando, "ler-mensagem", arquivo, (char *)NULL);
        } else {
            execlp(ator->comando, ator->comando, "--ler-mensagem", arquivo, (char *)NULL);
        }
    }
    execl("/bin/sh", "sh", "-c", comando_shell, (char *)NULL);
}

static void mensagem_info_minima(MensagemInfo *mi, const char *mensagem_id, const char *processo_id, const char *origem, const char *destino) {
    memset(mi, 0, sizeof *mi);
    mi->valido = true;
    snprintf(mi->mensagem_id, sizeof mi->mensagem_id, "%s", mensagem_id ? mensagem_id : "");
    snprintf(mi->processo_id, sizeof mi->processo_id, "%s", processo_id ? processo_id : "");
    snprintf(mi->origem, sizeof mi->origem, "%s", origem ? origem : "");
    snprintf(mi->destino, sizeof mi->destino, "%s", destino ? destino : "");
    snprintf(mi->prioridade, sizeof mi->prioridade, "normal");
    mi->prioridade_peso = prioridade_peso(mi->prioridade);
    mi->criado_ts = time(NULL);
}

static void mensagem_info_de_job(const Job *job, MensagemInfo *mi) {
    mensagem_info_minima(mi, job ? job->mensagem_id : "", job ? job->processo_id : "", job ? job->origem : "", job ? job->destino : "");
}

static void mensagem_info_de_item(const ItemAgendavel *item, MensagemInfo *mi) {
    mensagem_info_minima(mi, item ? item->mensagem_id : "", item ? item->processo_id : "", item ? item->origem : "", item ? item->destino : "");
}

static void mensagem_info_de_jobmsg(const JobMensagem *jm, MensagemInfo *mi) {
    mensagem_info_minima(mi, jm ? jm->mensagem_id : "", jm ? jm->processo_id : "", jm ? jm->origem : "", jm ? jm->destino : "");
}

static void preencher_jobmsg_de_item(const ItemAgendavel *item, const char *arquivo_claim, JobMensagem *jm) {
    memset(jm, 0, sizeof *jm);
    snprintf(jm->arquivo, sizeof jm->arquivo, "%s", arquivo_claim ? arquivo_claim : "");
    snprintf(jm->destino, sizeof jm->destino, "%s", item ? item->destino : "");
    snprintf(jm->origem, sizeof jm->origem, "%s", item ? item->origem : "");
    snprintf(jm->processo_id, sizeof jm->processo_id, "%s", item ? item->processo_id : "");
    snprintf(jm->mensagem_id, sizeof jm->mensagem_id, "%s", item ? item->mensagem_id : "");
}

static void job_free(Job *job) {
    if (!job) return;
    if (job->lista_lote[0]) unlink(job->lista_lote);
    free(job->msgs);
    memset(job, 0, sizeof *job);
}

static void job_remover(Job *jobs, int *njobs, int idx) {
    if (!jobs || !njobs || idx < 0 || idx >= *njobs) return;
    int last = *njobs - 1;
    job_free(&jobs[idx]);
    if (idx != last) {
        jobs[idx] = jobs[last];
        memset(&jobs[last], 0, sizeof jobs[last]);
    }
    (*njobs)--;
}

static bool spawn_job(const Config *cfg, const Ator *ator, const char *arquivo, const ItemAgendavel *item, Job *job) {
    MensagemInfo mi;
    if (item) {
        mensagem_info_de_item(item, &mi);
    } else if (!extrair_info_arquivo(cfg, arquivo, &mi)) {
        move_to_class(cfg, arquivo, "quarentena", "processar-", &mi, mi.erro);
        return false;
    }

    char comando_final[4096];
    montar_comando_acionamento(ator, arquivo, comando_final, sizeof comando_final);

    pid_t pid = fork();
    if (pid < 0) return false;
    if (pid == 0) {
        setpgid(0, 0);
        if (chdir(cfg->base) != 0) _exit(126);
        redirecionar_saida_ator_resposta(cfg, &mi);
        exec_acionamento_ator(ator, arquivo, comando_final);
        _exit(127);
    }
    setpgid(pid, pid);
    memset(job, 0, sizeof *job);
    job->pid = pid;
    job->iniciado = time(NULL);
    snprintf(job->destino, sizeof job->destino, "%s", ator->entidade);
    snprintf(job->origem, sizeof job->origem, "%s", mi.origem);
    snprintf(job->processo_id, sizeof job->processo_id, "%s", mi.processo_id);
    snprintf(job->arquivo, sizeof job->arquivo, "%s", arquivo);
    snprintf(job->mensagem_id, sizeof job->mensagem_id, "%s", mi.mensagem_id);
    snprintf(job->comando, sizeof job->comando, "%s", comando_final);
    log_event(cfg, "ator_iniciado", arquivo, &mi, comando_final);
    return true;
}

static void montar_comando_lote(const Ator *ator, const char *lista, char *out, size_t outsz) {
    char lista_q[PATH_MAX * 2];
    shell_quote(lista, lista_q, sizeof lista_q);
    if (strcmp(ator->tipo, "conector") == 0) {
        fmt_checked(out, outsz, "%s ler-mensagens-arquivo %s", ator->comando_lote, lista_q);
    } else {
        fmt_checked(out, outsz, "%s --ler-mensagens-arquivo %s", ator->comando_lote, lista_q);
    }
}

static void exec_acionamento_lote_ator(const Ator *ator, const char *lista, const char *comando_shell) {
    if (ator && comando_exec_simples(ator->comando_lote)) {
        if (strcmp(ator->tipo, "conector") == 0) {
            execlp(ator->comando_lote, ator->comando_lote, "ler-mensagens-arquivo", lista, (char *)NULL);
        } else {
            execlp(ator->comando_lote, ator->comando_lote, "--ler-mensagens-arquivo", lista, (char *)NULL);
        }
    }
    execl("/bin/sh", "sh", "-c", comando_shell, (char *)NULL);
}

static bool escrever_lista_lote(const Config *cfg, const JobMensagem *msgs, int nmsgs, char *out, size_t outsz) {
    static unsigned seq = 0;
    char tmpdir[PATH_MAX];
    path_join(tmpdir, sizeof tmpdir, cfg->espaco, "tmp");
    mkdir_p(tmpdir);
    fmt_checked(out, outsz, "%s/lote-%ld-%d-%u.lst", tmpdir, (long)time(NULL), getpid(), ++seq);
    FILE *f = fopen(out, "wx");
    if (!f) return false;
    for (int i = 0; i < nmsgs; i++) fprintf(f, "%s\n", msgs[i].arquivo);
    fflush(f);
    fsync(fileno(f));
    if (fclose(f) != 0) { unlink(out); out[0] = 0; return false; }
    return true;
}

static bool spawn_job_lote(const Config *cfg, const Ator *ator, const JobMensagem *msgs, int nmsgs, Job *job) {
    if (!ator || !ator->tem_comando_lote || !msgs || nmsgs <= 0) return false;
    char lista[PATH_MAX];
    if (!escrever_lista_lote(cfg, msgs, nmsgs, lista, sizeof lista)) return false;

    char comando_final[4096];
    montar_comando_lote(ator, lista, comando_final, sizeof comando_final);

    pid_t pid = fork();
    if (pid < 0) { unlink(lista); return false; }
    if (pid == 0) {
        setpgid(0, 0);
        if (chdir(cfg->base) != 0) _exit(126);
        exec_acionamento_lote_ator(ator, lista, comando_final);
        _exit(127);
    }
    setpgid(pid, pid);
    memset(job, 0, sizeof *job);
    job->pid = pid;
    job->iniciado = time(NULL);
    snprintf(job->destino, sizeof job->destino, "%s", ator->entidade);
    snprintf(job->origem, sizeof job->origem, "%s", msgs[0].origem);
    snprintf(job->processo_id, sizeof job->processo_id, "%s", msgs[0].processo_id);
    snprintf(job->arquivo, sizeof job->arquivo, "%s", msgs[0].arquivo);
    snprintf(job->mensagem_id, sizeof job->mensagem_id, "%s", msgs[0].mensagem_id);
    snprintf(job->comando, sizeof job->comando, "%s", comando_final);
    snprintf(job->lista_lote, sizeof job->lista_lote, "%s", lista);
    job->msgs = xcalloc((size_t)nmsgs, sizeof(JobMensagem));
    memcpy(job->msgs, msgs, sizeof(JobMensagem) * (size_t)nmsgs);
    job->nmsgs = nmsgs;
    MensagemInfo mi;
    mensagem_info_de_jobmsg(&msgs[0], &mi);
    char detalhe[64];
    snprintf(detalhe, sizeof detalhe, "mensagens=%d", nmsgs);
    log_event(cfg, "ator_lote_iniciado", lista, &mi, detalhe);
    return true;
}

static void sinalizar_job(pid_t pid, int sig) {
    if (pid <= 0) return;
    if (kill(-pid, sig) != 0) kill(pid, sig);
}

static void reentregar_job(const Config *cfg, const Job *job, const char *evento, const char *detalhe) {
    MensagemInfo mi;
    mensagem_info_de_job(job, &mi);
    char dst[PATH_MAX];
    caixa_path_base_name(cfg, job ? job->destino : "", base_name(job ? job->arquivo : "mensagem.json"), dst, sizeof dst);
    if (copy_or_rename(job->arquivo, dst)) log_event(cfg, evento, dst, &mi, detalhe);
    else log_event(cfg, "reentrega_falhou", job->arquivo, &mi, detalhe);
}

static void reentregar_jobmsg(const Config *cfg, const JobMensagem *jm, const char *evento, const char *detalhe) {
    MensagemInfo mi;
    mensagem_info_de_jobmsg(jm, &mi);
    char dst[PATH_MAX];
    caixa_path_base_name(cfg, jm ? jm->destino : "", base_name(jm ? jm->arquivo : "mensagem.json"), dst, sizeof dst);
    if (copy_or_rename(jm->arquivo, dst)) log_event(cfg, evento, dst, &mi, detalhe);
    else log_event(cfg, "reentrega_falhou", jm ? jm->arquivo : "", &mi, detalhe);
}

static void finalizar_job_resultado(const Config *cfg, Job *job, bool ok, const char *detalhe_base) {
    if (!job) return;
    if (job->nmsgs > 0 && job->msgs) {
        for (int i = 0; i < job->nmsgs; i++) {
            MensagemInfo mi;
            mensagem_info_de_jobmsg(&job->msgs[i], &mi);
            if (ok) {
                move_to_class(cfg, job->msgs[i].arquivo, "concluidas", "", &mi, detalhe_base);
            } else {
                int tent = tentativa_inc(cfg, job->msgs[i].mensagem_id);
                char detalhe[256];
                snprintf(detalhe, sizeof detalhe, "%s; tentativa %d", detalhe_base, tent);
                if (tent >= cfg->max_tentativas) move_to_class(cfg, job->msgs[i].arquivo, "erro", "", &mi, detalhe);
                else reentregar_jobmsg(cfg, &job->msgs[i], "reentregue", detalhe);
            }
        }
        return;
    }

    MensagemInfo mi;
    mensagem_info_de_job(job, &mi);
    if (ok) {
        move_to_class(cfg, job->arquivo, "concluidas", "", &mi, detalhe_base);
    } else {
        int tent = tentativa_inc(cfg, job->mensagem_id);
        char detalhe[256];
        snprintf(detalhe, sizeof detalhe, "%s; tentativa %d", detalhe_base, tent);
        if (tent >= cfg->max_tentativas) move_to_class(cfg, job->arquivo, "erro", "", &mi, detalhe);
        else reentregar_job(cfg, job, "reentregue", detalhe);
    }
}

static int coletar_jobs(Job *jobs, int *njobs, const Config *cfg, bool bloquear) {
    int finalizados = 0;
    while (*njobs > 0) {
        bool mudou = false;
        for (int i = 0; i < *njobs; i++) {
            int st = 0;
            pid_t r = waitpid(jobs[i].pid, &st, WNOHANG);
            time_t agora = time(NULL);
            if (r < 0 && errno == ECHILD) {
                finalizar_job_resultado(cfg, &jobs[i], false, "processo filho nao encontrado");
                job_remover(jobs, njobs, i);
                finalizados++;
                mudou = true;
                break;
            }
            if (r == 0 && cfg->timeout_ator > 0 && agora - jobs[i].iniciado >= cfg->timeout_ator) {
                sinalizar_job(jobs[i].pid, SIGTERM);
                usleep(200000);
                if (waitpid(jobs[i].pid, &st, WNOHANG) == 0) sinalizar_job(jobs[i].pid, SIGKILL);
                jobs[i].timeout_enviado = true;
                r = waitpid(jobs[i].pid, &st, 0);
            }
            if (r == jobs[i].pid) {
                bool ok = !jobs[i].timeout_enviado && WIFEXITED(st) && WEXITSTATUS(st) == 0;
                finalizar_job_resultado(cfg, &jobs[i], ok, ok ? "ator concluiu com codigo 0" : (jobs[i].timeout_enviado ? "timeout do ator" : "ator falhou"));
                job_remover(jobs, njobs, i);
                finalizados++;
                mudou = true;
                break;
            }
        }
        if (!bloquear || !mudou) break;
    }
    return finalizados;
}

static void cancelar_jobs_ativos(const Config *cfg, Job *jobs, int *njobs) {
    if (!jobs || !njobs || *njobs <= 0) return;
    coletar_jobs(jobs, njobs, cfg, true);
    if (*njobs <= 0) return;
    for (int i = 0; i < *njobs; i++) sinalizar_job(jobs[i].pid, SIGTERM);
    usleep(200000);
    for (int i = 0; i < *njobs; i++) {
        int st = 0;
        pid_t r = waitpid(jobs[i].pid, &st, WNOHANG);
        if (r == 0) sinalizar_job(jobs[i].pid, SIGKILL);
    }
    for (int i = 0; i < *njobs; i++) {
        int st = 0;
        waitpid(jobs[i].pid, &st, 0);
        if (jobs[i].nmsgs > 0 && jobs[i].msgs) {
            for (int j = 0; j < jobs[i].nmsgs; j++) {
                reentregar_jobmsg(cfg, &jobs[i].msgs[j], "reentregue_shutdown", "escuta encerrada; mensagem devolvida para a caixa");
            }
        } else {
            reentregar_job(cfg, &jobs[i], "reentregue_shutdown", "escuta encerrada; mensagem devolvida para a caixa");
        }
        job_free(&jobs[i]);
    }
    *njobs = 0;
}

static void recuperar_processando_antigo(const Config *cfg) {
    char procdir[PATH_MAX]; path_join(procdir, sizeof procdir, cfg->espaco, "processando");
    ListaArquivos files = listar_json(procdir, cfg->lote);
    time_t agora = time(NULL);
    for (int i = 0; i < files.len && !g_parar; i++) {
        const char *p = files.items[i];
        time_t mt = file_mtime(p);
        if (mt <= 0 || agora - mt < cfg->timeout_processamento) continue;

        MensagemInfo mi; extrair_info_arquivo(cfg, p, &mi);
        if (strncmp(base_name(p), "roteando-", 9) == 0) {
            char entrada[PATH_MAX], dst[PATH_MAX];
            path_join(entrada, sizeof entrada, cfg->espaco, "entrada");
            fmt_checked(dst, sizeof dst, "%s/recuperado-%ld-%s", entrada, (long)agora, base_name(p));
            copy_or_rename(p, dst);
            log_event(cfg, "recuperado_para_entrada", dst, &mi, "roteamento abandonado devolvido para validacao completa");
            continue;
        }

        int tent = tentativa_inc(cfg, mi.mensagem_id[0] ? mi.mensagem_id : base_name(p));
        if (tent >= cfg->max_tentativas) {
            move_to_class(cfg, p, "erro", "stale-", &mi, "processamento abandonado excedeu tentativas");
        } else {
            char dst[PATH_MAX];
            caixa_path_base_name(cfg, mi.destino, base_name(p), dst, sizeof dst);
            copy_or_rename(p, dst);
            log_event(cfg, "recuperado", dst, &mi, "processamento antigo devolvido a caixa");
        }
    }
    lista_arquivos_free(&files);
}

static int comparar_itens_agendaveis(const void *pa, const void *pb) {
    const ItemAgendavel *a = (const ItemAgendavel *)pa;
    const ItemAgendavel *b = (const ItemAgendavel *)pb;
    if (a->prioridade_peso != b->prioridade_peso) return a->prioridade_peso - b->prioridade_peso;
    int cdest = strcmp(a->destino, b->destino);
    if (cdest != 0) return cdest;
    int anum = a->numero_mensagem > 0 ? a->numero_mensagem : INT_MAX;
    int bnum = b->numero_mensagem > 0 ? b->numero_mensagem : INT_MAX;
    if (anum != bnum) return anum < bnum ? -1 : 1;
    if (a->criado_ts != b->criado_ts) return a->criado_ts < b->criado_ts ? -1 : 1;
    return strcmp(a->arquivo, b->arquivo);
}

static int acionar_atores(const Config *cfg, Job *jobs, int *njobs, bool drenar_lote) {
    if (!jobs || !njobs) return 0;
    if (*njobs >= cfg->paralelo && !drenar_lote) return 0;

    ItemAgendavel *itens = xcalloc((size_t)cfg->lote, sizeof(ItemAgendavel));
    int nitens = 0;
    int iniciados = 0;
    char caixas[PATH_MAX]; path_join(caixas, sizeof caixas, cfg->espaco, "caixas");
    ListaArquivos files = listar_json(caixas, cfg->lote);
    time_t agora = time(NULL);

    for (int i = 0; i < files.len && nitens < cfg->lote; i++) {
        const char *arquivo = files.items[i];
        MensagemInfo mi;
        if (!extrair_info_arquivo(cfg, arquivo, &mi)) {
            move_to_class(cfg, arquivo, "quarentena", "caixa-", &mi, mi.erro);
            continue;
        }
        if (mi.expira_ts > 0 && mi.expira_ts <= agora) {
            move_to_class(cfg, arquivo, "erro", "expirada-", &mi, "mensagem expirada antes do processamento");
            continue;
        }
        if (mi.nao_processar_antes_ts > 0 && mi.nao_processar_antes_ts > agora) {
            continue;
        }
        char motivo[256];
        int dep = bloqueio_dependencias(cfg, &mi, motivo, sizeof motivo);
        if (dep == 2) {
            move_to_class(cfg, arquivo, "erro", "dep-", &mi, motivo);
            continue;
        }
        if (dep == 1) {
            log_event(cfg, "bloqueada_dependencia", arquivo, &mi, motivo);
            continue;
        }
        int seq = bloqueio_sequencial(cfg, &mi, motivo, sizeof motivo);
        if (seq == 2) {
            move_to_class(cfg, arquivo, "erro", "seq-", &mi, motivo);
            continue;
        }
        if (seq == 1) {
            log_event(cfg, "bloqueada_sequencia", arquivo, &mi, motivo);
            continue;
        }
        Ator ator = carregar_ator_cacheado(cfg, mi.destino);
        if (!ator.existe || !ator.ativo) {
            move_to_class(cfg, arquivo, "quarentena", "ator-", &mi, "destinatario deixou de existir ou esta inativo");
            continue;
        }
        if (!ator.tem_comando && !ator.tem_comando_lote) {
            move_to_class(cfg, arquivo, "quarentena", "controlador-", &mi, "destinatario sem controlador.executavel/comando/comandoLote declarado");
            continue;
        }
        itens[nitens].arquivo = arquivo;
        snprintf(itens[nitens].destino, sizeof itens[nitens].destino, "%s", mi.destino);
        snprintf(itens[nitens].origem, sizeof itens[nitens].origem, "%s", mi.origem);
        snprintf(itens[nitens].processo_id, sizeof itens[nitens].processo_id, "%s", mi.processo_id);
        snprintf(itens[nitens].mensagem_id, sizeof itens[nitens].mensagem_id, "%s", mi.mensagem_id);
        itens[nitens].prioridade_peso = mi.prioridade_peso;
        itens[nitens].numero_mensagem = mi.numero_mensagem;
        itens[nitens].criado_ts = mi.criado_ts;
        nitens++;
    }

    qsort(itens, (size_t)nitens, sizeof(ItemAgendavel), comparar_itens_agendaveis);

    for (int i = 0; i < nitens && !g_parar; i++) {
        while (*njobs >= cfg->paralelo) {
            if (g_parar || !drenar_lote) goto fim;
            coletar_jobs(jobs, njobs, cfg, true);
            if (*njobs >= cfg->paralelo) usleep(100000);
        }

        Ator ator = carregar_ator_cacheado(cfg, itens[i].destino);
        if (!ator.existe || !ator.ativo || (!ator.tem_comando && !ator.tem_comando_lote)) {
            MensagemInfo mi_spawn; mensagem_info_de_item(&itens[i], &mi_spawn);
            move_to_class(cfg, itens[i].arquivo, "quarentena", "ator-", &mi_spawn, "destinatario indisponivel no momento do acionamento");
            continue;
        }

        if (ator.tem_comando_lote) {
            int limite = ator.lote_max > 0 ? ator.lote_max : cfg->lote_ator;
            if (limite > cfg->lote_ator) limite = cfg->lote_ator;
            if (limite < 1) limite = 1;
            JobMensagem *msgs = xcalloc((size_t)limite, sizeof(JobMensagem));
            int nmsgs = 0;
            int j = i;
            for (; j < nitens && nmsgs < limite; j++) {
                if (strcmp(itens[j].destino, itens[i].destino) != 0) break;
                char procdir[PATH_MAX], claim[PATH_MAX];
                path_join(procdir, sizeof procdir, cfg->espaco, "processando");
                fmt_checked(claim, sizeof claim, "%s/%s", procdir, base_name(itens[j].arquivo));
                if (rename(itens[j].arquivo, claim) != 0) continue;
                preencher_jobmsg_de_item(&itens[j], claim, &msgs[nmsgs++]);
            }
            if (nmsgs > 0) {
                if (spawn_job_lote(cfg, &ator, msgs, nmsgs, &jobs[*njobs])) { (*njobs)++; iniciados++; }
                else {
                    for (int k = 0; k < nmsgs; k++) {
                        MensagemInfo mi_spawn; mensagem_info_de_jobmsg(&msgs[k], &mi_spawn);
                        move_to_class(cfg, msgs[k].arquivo, "erro", "spawn-lote-", &mi_spawn, "falha ao iniciar lote do ator");
                    }
                }
            }
            free(msgs);
            i = j - 1;
            continue;
        }

        char procdir[PATH_MAX], claim[PATH_MAX];
        path_join(procdir, sizeof procdir, cfg->espaco, "processando");
        fmt_checked(claim, sizeof claim, "%s/%s", procdir, base_name(itens[i].arquivo));
        if (rename(itens[i].arquivo, claim) != 0) continue;
        if (spawn_job(cfg, &ator, claim, &itens[i], &jobs[*njobs])) { (*njobs)++; iniciados++; }
        else {
            MensagemInfo mi_spawn; mensagem_info_de_item(&itens[i], &mi_spawn);
            move_to_class(cfg, claim, "erro", "spawn-", &mi_spawn, "falha ao iniciar ator");
        }
    }

fim:
    free(itens);
    lista_arquivos_free(&files);
    return iniciados;
}

static int count_json_recursive_one_level(const char *dir) {
    int total = 0;
    DIR *d = opendir(dir); if (!d) return 0;
    struct dirent *e;
    while ((e = readdir(d))) {
        if (e->d_name[0] == '.') continue;
        char p[PATH_MAX]; path_join(p, sizeof p, dir, e->d_name);
        struct stat st; if (stat(p, &st) != 0) continue;
        if (S_ISREG(st.st_mode) && ends_with(e->d_name, ".json")) total++;
        else if (S_ISDIR(st.st_mode)) {
            DIR *sd = opendir(p); if (!sd) continue;
            struct dirent *x;
            while ((x = readdir(sd))) if (x->d_name[0] != '.' && ends_with(x->d_name, ".json")) total++;
            closedir(sd);
        }
    }
    closedir(d);
    return total;
}

static void print_status(const Config *cfg) {
    const char *names[] = {"entrada", "caixas", "processando", "concluidas", "erro", "quarentena", "duplicadas"};
    printf("{\n  \"espaco\": "); json_escape(stdout, cfg->espaco); printf(",\n");
    printf("  \"contagens\": {\n");
    for (size_t i = 0; i < sizeof(names)/sizeof(names[0]); i++) {
        char p[PATH_MAX]; path_join(p, sizeof p, cfg->espaco, names[i]);
        printf("    \"%s\": %d%s\n", names[i], count_json_recursive_one_level(p), i + 1 == sizeof(names)/sizeof(names[0]) ? "" : ",");
    }
    printf("  },\n");
    printf("  \"capacidade\": {\n");
    printf("    \"loteAtual\": %d,\n", cfg->lote);
    printf("    \"lotePadrao\": %d,\n", SIS_DEFAULT_LOTE);
    printf("    \"paraleloAtual\": %d,\n", cfg->paralelo);
    printf("    \"paraleloPadrao\": %d,\n", SIS_DEFAULT_PARALELO);
    printf("    \"paraleloMaximoConfiguravel\": %d,\n", SIS_MAX_PARALELO);
    printf("    \"loteAtorAtual\": %d,\n", cfg->lote_ator);
    printf("    \"loteAtorPadrao\": %d,\n", SIS_DEFAULT_LOTE_ATOR);
    printf("    \"intervaloLimpezaAtual\": %d,\n", cfg->intervalo_limpeza);
    printf("    \"logDetalhado\": %s,\n", cfg->log_detalhado ? "true" : "false");
    printf("    \"validarMarcadorApi\": %s\n", cfg->validar_marcador_api ? "true" : "false");
    printf("  }\n}\n");
}

static int limite_paralelo_sistema(void) {
    long cap = SIS_MAX_PARALELO;
#ifdef RLIMIT_NPROC
    struct rlimit rl;
    if (getrlimit(RLIMIT_NPROC, &rl) == 0 && rl.rlim_cur != RLIM_INFINITY && rl.rlim_cur > 0) {
        long nproc = (long)rl.rlim_cur;
        if (nproc > 64 && nproc - 32 < cap) cap = nproc - 32;
    }
#endif
    if (cap < 1) cap = 1;
    return (int)cap;
}

static void usage(const char *argv0) {
    printf("Uso: %s [--base=DIR] [--once] [--init] [--status] [--intervalo=N] [--lote=N] [--paralelo=N] [--lote-ator=N] [--timeout-ator=N] [--timeout-processamento=N] [--max-tentativas=N] [--max-bytes=N] [--reter-eventos=N] [--reter-estado=N] [--reter-idempotencia=N] [--intervalo-limpeza=N] [--log-detalhado|--log-compacto] [--validar-marcador-api]\n", argv0);
    printf("\n");
    printf("Capacidade de alta carga:\n");
    printf("  --lote=N                 Mensagens avaliadas por ciclo (padrao: %d; maximo: %d)\n", SIS_DEFAULT_LOTE, SIS_MAX_LOTE);
    printf("  --paralelo=N             Atores simultaneos no pool local (padrao: %d; maximo configuravel: %d)\n", SIS_DEFAULT_PARALELO, SIS_MAX_PARALELO);
    printf("  --lote-ator=N            Arquivos por acionamento em lote quando o ator declara comandoLote (padrao: %d; maximo: %d)\n", SIS_DEFAULT_LOTE_ATOR, SIS_MAX_LOTE_ATOR);
    printf("  --intervalo=N            Intervalo do daemon entre ciclos (padrao: %d segundo)\n", SIS_DEFAULT_INTERVALO);
    printf("  --intervalo-limpeza=N    Segundos entre limpezas pesadas de marcadores antigos (padrao: %d; 0 desabilita)\n", SIS_DEFAULT_INTERVALO_LIMPEZA);
    printf("  --log-compacto           Registra sucessos por estado/ciclo e suprime eventos ruidosos (padrao)\n");
    printf("  --log-detalhado          Registra cada entrega/inicio/conclusao para auditoria fina\n");
    printf("  --validar-marcador-api   Le e confere cada marcador api-publicadas (mais seguro, mais I/O)\n");
    printf("\n");
    printf("Retencao (dias, 0=desabilita limpeza por idade):\n");
    printf("  --reter-eventos=N        Dias para manter logs de eventos .jsonl (padrao: %d)\n", SIS_DEFAULT_RETER_EVENTOS);
    printf("  --reter-estado=N         Dias para manter estado/concluidas, terminais, respondida, processos e vinculos (padrao: %d)\n", SIS_DEFAULT_RETER_ESTADO);
    printf("  --reter-idempotencia=N   Dias para manter marcadores de idempotencia (padrao: %d)\n", SIS_DEFAULT_RETER_IDEMPOTENCIA);
    printf("\n");
    printf("Exemplo com todas as flags juntas:\n");
    printf("  %s --base=/caminho --once --init --status --intervalo=1 --lote=20000 --paralelo=512 --lote-ator=100 --timeout-ator=300 --timeout-processamento=900 --max-tentativas=3 --max-bytes=5242880 --intervalo-limpeza=300\n", argv0);
}

static Config parse_args(int argc, char **argv) {
    Config c;
    memset(&c, 0, sizeof c);
    if (!getcwd(c.base, sizeof c.base)) die("Nao foi possivel obter diretorio atual.");
    c.once = 0; c.init = 0; c.status = 0;
    c.intervalo = SIS_DEFAULT_INTERVALO;
    c.lote = SIS_DEFAULT_LOTE;
    c.paralelo = SIS_DEFAULT_PARALELO;
    c.timeout_ator = SIS_DEFAULT_TIMEOUT_ATOR;
    c.timeout_processamento = SIS_DEFAULT_TIMEOUT_PROCESSAMENTO;
    c.max_tentativas = SIS_DEFAULT_MAX_TENTATIVAS;
    c.max_bytes = SIS_DEFAULT_MAX_BYTES;
    c.reter_eventos = SIS_DEFAULT_RETER_EVENTOS;
    c.reter_estado = SIS_DEFAULT_RETER_ESTADO;
    c.reter_idempotencia = SIS_DEFAULT_RETER_IDEMPOTENCIA;
    c.intervalo_limpeza = SIS_DEFAULT_INTERVALO_LIMPEZA;
    c.lote_ator = SIS_DEFAULT_LOTE_ATOR;
    c.log_detalhado = 0;
    c.validar_marcador_api = 0;
    for (int i = 1; i < argc; i++) {
        const char *a = argv[i];
        if (strcmp(a, "--once") == 0) c.once = 1;
        else if (strcmp(a, "--init") == 0) c.init = 1;
        else if (strcmp(a, "--status") == 0) c.status = 1;
        else if (strcmp(a, "--log-detalhado") == 0) c.log_detalhado = 1;
        else if (strcmp(a, "--log-compacto") == 0) c.log_detalhado = 0;
        else if (strcmp(a, "--validar-marcador-api") == 0) c.validar_marcador_api = 1;
        else if (strcmp(a, "--marcador-api-rapido") == 0) c.validar_marcador_api = 0;
        else if (strcmp(a, "--help") == 0 || strcmp(a, "-h") == 0) { usage(argv[0]); exit(0); }
        else if (strncmp(a, "--base=", 7) == 0) snprintf(c.base, sizeof c.base, "%s", a + 7);
        else if (strncmp(a, "--intervalo=", 12) == 0) c.intervalo = atoi(a + 12);
        else if (strncmp(a, "--lote=", 7) == 0) c.lote = atoi(a + 7);
        else if (strncmp(a, "--paralelo=", 11) == 0) c.paralelo = atoi(a + 11);
        else if (strncmp(a, "--lote-ator=", 12) == 0) c.lote_ator = atoi(a + 12);
        else if (strncmp(a, "--timeout-ator=", 15) == 0) c.timeout_ator = atoi(a + 15);
        else if (strncmp(a, "--timeout-processamento=", 24) == 0) c.timeout_processamento = atoi(a + 24);
        else if (strncmp(a, "--max-tentativas=", 17) == 0) c.max_tentativas = atoi(a + 17);
        else if (strncmp(a, "--max-bytes=", 12) == 0) c.max_bytes = (size_t)atoll(a + 12);
        else if (strncmp(a, "--reter-eventos=", 16) == 0) c.reter_eventos = atoi(a + 16);
        else if (strncmp(a, "--reter-estado=", 15) == 0) c.reter_estado = atoi(a + 15);
        else if (strncmp(a, "--reter-idempotencia=", 21) == 0) c.reter_idempotencia = atoi(a + 21);
        else if (strncmp(a, "--intervalo-limpeza=", 20) == 0) c.intervalo_limpeza = atoi(a + 20);
        else die("Argumento invalido: %s", a);
    }
    if (c.intervalo < 1) c.intervalo = 1;
    if (c.lote < 1) c.lote = 1;
    if (c.lote > SIS_MAX_LOTE) c.lote = SIS_MAX_LOTE;
    if (c.paralelo < 1) c.paralelo = 1;
    int paralelo_cap = limite_paralelo_sistema();
    if (c.paralelo > paralelo_cap) c.paralelo = paralelo_cap;
    if (c.lote_ator < 1) c.lote_ator = 1;
    if (c.lote_ator > SIS_MAX_LOTE_ATOR) c.lote_ator = SIS_MAX_LOTE_ATOR;
    if (c.intervalo_limpeza < 0) c.intervalo_limpeza = 0;
    if (c.timeout_ator < 1) c.timeout_ator = 1;
    if (c.timeout_processamento < 10) c.timeout_processamento = 10;
    if (c.max_tentativas < 1) c.max_tentativas = 1;
    path_join(c.espaco, sizeof c.espaco, c.base, "espaco");
    return c;
}

static int acquire_lock(const Config *cfg) {
    char lockdir[PATH_MAX], lockfile[PATH_MAX];
    path_join(lockdir, sizeof lockdir, cfg->espaco, "locks"); mkdir_p(lockdir);
    path_join(lockfile, sizeof lockfile, lockdir, "escuta.lock");
    int fd = open(lockfile, O_CREAT | O_RDWR, 0664);
    if (fd < 0) die("Nao foi possivel abrir lock: %s", strerror(errno));
    fcntl(fd, F_SETFD, FD_CLOEXEC);
    if (flock(fd, LOCK_EX | LOCK_NB) != 0) die("escuta.c ja esta em execucao para este espaco.");
    char buf[64]; snprintf(buf, sizeof buf, "%ld\n", (long)getpid());
    ftruncate(fd, 0); write(fd, buf, strlen(buf)); fsync(fd);
    return fd;
}

static void release_lock(const Config *cfg, int fd) {
    char lockdir[PATH_MAX], lockfile[PATH_MAX];
    path_join(lockdir, sizeof lockdir, cfg->espaco, "locks");
    path_join(lockfile, sizeof lockfile, lockdir, "escuta.lock");
    unlink(lockfile);
    close(fd);
}

int main(int argc, char **argv) {
    Config cfg = parse_args(argc, argv);
    ensure_layout(&cfg);
    if (cfg.init) {
        printf("Estrutura do espaco inicializada em %s\n", cfg.espaco);
        if (!cfg.status && !cfg.once) return 0;
    }
    if (cfg.status) { print_status(&cfg); return 0; }
    signal(SIGTERM, sinalizar_parada);
    signal(SIGINT, sinalizar_parada);
    signal(SIGHUP, sinalizar_parada);
    int lockfd = acquire_lock(&cfg);
    Job *jobs = xcalloc((size_t)cfg.paralelo, sizeof(Job));
    int njobs = 0;
    time_t ultima_limpeza = time(NULL);
    char inicio[192];
    snprintf(inicio, sizeof inicio, "%s; lote=%d paralelo=%d lote_ator=%d intervalo=%d limpeza=%d log=%s marcador_api=%s", cfg.once ? "modo once" : "modo continuo", cfg.lote, cfg.paralelo, cfg.lote_ator, cfg.intervalo, cfg.intervalo_limpeza, cfg.log_detalhado ? "detalhado" : "compacto", cfg.validar_marcador_api ? "estrito" : "rapido");
    log_event(&cfg, "escuta_iniciada", NULL, NULL, inicio);
    do {
        int finalizadas = coletar_jobs(jobs, &njobs, &cfg, false);
        if (g_parar) break;
        time_t agora = time(NULL);
        if (cfg.intervalo_limpeza > 0 && agora - ultima_limpeza >= cfg.intervalo_limpeza) {
            limpar_residuos_espaco(&cfg);
            ultima_limpeza = agora;
        }
        recuperar_processando_antigo(&cfg);
        int roteadas = rotear_entradas(&cfg);
        finalizadas += coletar_jobs(jobs, &njobs, &cfg, false);
        int acionadas = acionar_atores(&cfg, jobs, &njobs, true);
        char detalhe[192]; snprintf(detalhe, sizeof detalhe, "roteadas=%d acionadas=%d finalizadas=%d jobs_ativos=%d", roteadas, acionadas, finalizadas, njobs);
        log_event(&cfg, "ciclo_concluido", NULL, NULL, detalhe);
        if (cfg.once) {
            if (roteadas == 0 && acionadas == 0 && finalizadas == 0 && njobs == 0) break;
            usleep((roteadas || acionadas || finalizadas) ? 10000 : 100000);
            continue;
        }
        if (njobs >= cfg.paralelo) usleep(100000);
        else sleep((unsigned int)cfg.intervalo);
    } while (1);
    if (g_parar && njobs > 0) {
        log_event(&cfg, "escuta_parando", NULL, NULL, "sinal recebido; interrompendo atores ativos e reentregando mensagens");
        cancelar_jobs_ativos(&cfg, jobs, &njobs);
    }
    free(jobs);
    log_event(&cfg, "escuta_finalizada", NULL, NULL, g_parar ? "encerramento por sinal" : "encerramento normal");
    log_event_close();
    release_lock(&cfg, lockfd);
    return 0;
}