laantungir/didactyl
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity
Files
59bd2603e2fcb5c69e6c154e3419ec6d014e1ea0
didactyl/src/trigger_manager.c
T

1730 lines
56 KiB
C
Raw Blame History

#define _POSIX_C_SOURCE 200809L
#include "trigger_manager.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <ctype.h>
#include "cjson/cJSON.h"
#include "debug.h"
#include "nostr_handler.h"
#include "llm.h"
void agent_on_trigger(const char* skill_d_tag,
const char* skill_content,
cJSON* triggering_event,
const char* relay_url);
typedef struct {
trigger_manager_t* mgr;
char skill_d_tag[TRIGGER_SKILL_D_TAG_MAX];
} trigger_subscription_ctx_t;
static int clamp_enabled(int enabled) {
return enabled ? 1 : 0;
}
trigger_type_t trigger_type_from_string(const char* s) {
if (!s) {
return TRIGGER_TYPE_NOSTR_SUBSCRIPTION;
}
if (strcmp(s, "webhook") == 0) {
return TRIGGER_TYPE_WEBHOOK;
}
if (strcmp(s, "cron") == 0) {
return TRIGGER_TYPE_CRON;
}
if (strcmp(s, "chain") == 0) {
return TRIGGER_TYPE_CHAIN;
}
if (strcmp(s, "dm") == 0) {
return TRIGGER_TYPE_DM;
}
return TRIGGER_TYPE_NOSTR_SUBSCRIPTION;
}
const char* trigger_type_to_string(trigger_type_t t) {
switch (t) {
case TRIGGER_TYPE_WEBHOOK:
return "webhook";
case TRIGGER_TYPE_CRON:
return "cron";
case TRIGGER_TYPE_CHAIN:
return "chain";
case TRIGGER_TYPE_DM:
return "dm";
case TRIGGER_TYPE_NOSTR_SUBSCRIPTION:
default:
return "nostr-subscription";
}
}
static int is_int_in_csv_list(const char* text, int value) {
if (!text || text[0] == '\0') {
return 0;
}
const char* p = text;
while (*p) {
while (*p == ' ') p++;
char* endptr = NULL;
long n = strtol(p, &endptr, 10);
if (endptr == p) {
return 0;
}
if ((int)n == value) {
return 1;
}
p = endptr;
while (*p == ' ') p++;
if (*p == ',') {
p++;
continue;
}
if (*p == '\0') {
break;
}
return 0;
}
return 0;
}
static int cron_field_token_matches(const char* token, int value, int min_v, int max_v) {
if (!token || token[0] == '\0') {
return 0;
}
if (strcmp(token, "*") == 0) {
return 1;
}
const char* slash = strchr(token, '/');
int step = 0;
char base[64];
if (slash) {
size_t base_len = (size_t)(slash - token);
if (base_len == 0 || base_len >= sizeof(base)) {
return 0;
}
memcpy(base, token, base_len);
base[base_len] = '\0';
char* endptr = NULL;
long step_l = strtol(slash + 1, &endptr, 10);
if (endptr == slash + 1 || *endptr != '\0' || step_l <= 0 || step_l > 1024) {
return 0;
}
step = (int)step_l;
} else {
snprintf(base, sizeof(base), "%s", token);
}
int range_start = min_v;
int range_end = max_v;
if (strcmp(base, "*") == 0) {
range_start = min_v;
range_end = max_v;
} else {
const char* dash = strchr(base, '-');
if (dash) {
char left[32];
char right[32];
size_t left_len = (size_t)(dash - base);
size_t right_len = strlen(dash + 1);
if (left_len == 0 || right_len == 0 || left_len >= sizeof(left) || right_len >= sizeof(right)) {
return 0;
}
memcpy(left, base, left_len);
left[left_len] = '\0';
memcpy(right, dash + 1, right_len + 1U);
char* e1 = NULL;
char* e2 = NULL;
long a = strtol(left, &e1, 10);
long b = strtol(right, &e2, 10);
if (!e1 || *e1 != '\0' || !e2 || *e2 != '\0') {
return 0;
}
range_start = (int)a;
range_end = (int)b;
} else if (strchr(base, ',')) {
if (!is_int_in_csv_list(base, value)) {
return 0;
}
if (step <= 1) {
return 1;
}
return ((value - min_v) % step) == 0;
} else {
char* e = NULL;
long n = strtol(base, &e, 10);
if (!e || *e != '\0') {
return 0;
}
range_start = (int)n;
range_end = (int)n;
}
}
if (range_start < min_v || range_end > max_v || range_start > range_end) {
return 0;
}
if (value < range_start || value > range_end) {
return 0;
}
if (step > 1 && ((value - range_start) % step) != 0) {
return 0;
}
return 1;
}
static int cron_field_matches(const char* field, int value, int min_v, int max_v) {
if (!field || field[0] == '\0') {
return 0;
}
char tmp[128];
snprintf(tmp, sizeof(tmp), "%s", field);
char* saveptr = NULL;
char* token = strtok_r(tmp, ",", &saveptr);
while (token) {
while (*token == ' ') token++;
if (cron_field_token_matches(token, value, min_v, max_v)) {
return 1;
}
token = strtok_r(NULL, ",", &saveptr);
}
return 0;
}
static int cron_normalize_expression(const char* expr, char* out, size_t out_sz, int* out_expanded) {
if (out_expanded) {
*out_expanded = 0;
}
if (!expr || !out || out_sz == 0) {
return -1;
}
while (*expr && isspace((unsigned char)*expr)) {
expr++;
}
if (*expr == '\0') {
return -1;
}
const char* mapped = NULL;
if (strcmp(expr, "*") == 0) {
mapped = "* * * * *";
} else if (strcmp(expr, "@yearly") == 0 || strcmp(expr, "@annually") == 0) {
mapped = "0 0 1 1 *";
} else if (strcmp(expr, "@monthly") == 0) {
mapped = "0 0 1 * *";
} else if (strcmp(expr, "@weekly") == 0) {
mapped = "0 0 * * 0";
} else if (strcmp(expr, "@daily") == 0 || strcmp(expr, "@midnight") == 0) {
mapped = "0 0 * * *";
} else if (strcmp(expr, "@hourly") == 0) {
mapped = "0 * * * *";
}
const char* src = mapped ? mapped : expr;
int n = snprintf(out, out_sz, "%s", src);
if (n < 0 || (size_t)n >= out_sz) {
return -1;
}
if (mapped && out_expanded) {
*out_expanded = 1;
}
return 0;
}
static int cron_field_token_valid(const char* token, int min_v, int max_v) {
if (!token || token[0] == '\0') {
return 0;
}
if (strcmp(token, "*") == 0) {
return 1;
}
const char* slash = strchr(token, '/');
int step = 0;
char base[64];
if (slash) {
size_t base_len = (size_t)(slash - token);
if (base_len == 0 || base_len >= sizeof(base)) {
return 0;
}
memcpy(base, token, base_len);
base[base_len] = '\0';
char* endptr = NULL;
long step_l = strtol(slash + 1, &endptr, 10);
if (endptr == slash + 1 || *endptr != '\0' || step_l <= 0 || step_l > 1024) {
return 0;
}
step = (int)step_l;
} else {
int n = snprintf(base, sizeof(base), "%s", token);
if (n < 0 || (size_t)n >= sizeof(base)) {
return 0;
}
}
if (strcmp(base, "*") == 0) {
return step == 0 || step <= (max_v - min_v + 1);
}
const char* dash = strchr(base, '-');
if (dash) {
char left[32];
char right[32];
size_t left_len = (size_t)(dash - base);
size_t right_len = strlen(dash + 1);
if (left_len == 0 || right_len == 0 || left_len >= sizeof(left) || right_len >= sizeof(right)) {
return 0;
}
memcpy(left, base, left_len);
left[left_len] = '\0';
memcpy(right, dash + 1, right_len + 1U);
char* e1 = NULL;
char* e2 = NULL;
long a = strtol(left, &e1, 10);
long b = strtol(right, &e2, 10);
if (!e1 || *e1 != '\0' || !e2 || *e2 != '\0') {
return 0;
}
if ((int)a < min_v || (int)b > max_v || (int)a > (int)b) {
return 0;
}
return 1;
}
char* e = NULL;
long n = strtol(base, &e, 10);
if (!e || *e != '\0') {
return 0;
}
if ((int)n < min_v || (int)n > max_v) {
return 0;
}
return 1;
}
static int cron_field_valid(const char* field, int min_v, int max_v) {
if (!field || field[0] == '\0') {
return 0;
}
char tmp[128];
int n = snprintf(tmp, sizeof(tmp), "%s", field);
if (n < 0 || (size_t)n >= sizeof(tmp)) {
return 0;
}
int tokens = 0;
char* saveptr = NULL;
char* token = strtok_r(tmp, ",", &saveptr);
while (token) {
while (*token == ' ') token++;
if (!cron_field_token_valid(token, min_v, max_v)) {
return 0;
}
tokens++;
token = strtok_r(NULL, ",", &saveptr);
}
return tokens > 0;
}
static int cron_expr_is_valid(const char* expr, char* normalized_out, size_t normalized_out_size) {
if (!normalized_out || normalized_out_size == 0) {
return 0;
}
if (cron_normalize_expression(expr, normalized_out, normalized_out_size, NULL) != 0) {
return 0;
}
char parse_buf[TRIGGER_FILTER_JSON_MAX];
int n = snprintf(parse_buf, sizeof(parse_buf), "%s", normalized_out);
if (n < 0 || (size_t)n >= sizeof(parse_buf)) {
return 0;
}
char* fields[5] = {0};
int nf = 0;
char* saveptr = NULL;
char* tok = strtok_r(parse_buf, " \t", &saveptr);
while (tok && nf < 5) {
fields[nf++] = tok;
tok = strtok_r(NULL, " \t", &saveptr);
}
if (nf != 5 || tok != NULL) {
return 0;
}
if (!cron_field_valid(fields[0], 0, 59)) return 0;
if (!cron_field_valid(fields[1], 0, 23)) return 0;
if (!cron_field_valid(fields[2], 1, 31)) return 0;
if (!cron_field_valid(fields[3], 1, 12)) return 0;
if (!cron_field_valid(fields[4], 0, 6)) return 0;
return 1;
}
static int cron_matches_now(const char* expr, time_t now_ts) {
if (!expr || expr[0] == '\0') {
return 0;
}
char normalized[TRIGGER_FILTER_JSON_MAX];
if (!cron_expr_is_valid(expr, normalized, sizeof(normalized))) {
return 0;
}
char parse_buf[TRIGGER_FILTER_JSON_MAX];
snprintf(parse_buf, sizeof(parse_buf), "%s", normalized);
char* fields[5] = {0};
int nf = 0;
char* saveptr = NULL;
char* tok = strtok_r(parse_buf, " \t", &saveptr);
while (tok && nf < 5) {
fields[nf++] = tok;
tok = strtok_r(NULL, " \t", &saveptr);
}
if (nf != 5 || tok != NULL) {
return 0;
}
struct tm tm_now;
localtime_r(&now_ts, &tm_now);
int minute = tm_now.tm_min;
int hour = tm_now.tm_hour;
int mday = tm_now.tm_mday;
int month = tm_now.tm_mon + 1;
int wday = tm_now.tm_wday;
if (!cron_field_matches(fields[0], minute, 0, 59)) return 0;
if (!cron_field_matches(fields[1], hour, 0, 23)) return 0;
if (!cron_field_matches(fields[2], mday, 1, 31)) return 0;
if (!cron_field_matches(fields[3], month, 1, 12)) return 0;
if (!cron_field_matches(fields[4], wday, 0, 6)) return 0;
return 1;
}
static int ensure_capacity(trigger_manager_t* mgr, int needed) {
if (!mgr || needed <= 0) {
return -1;
}
if (mgr->capacity >= needed) {
return 0;
}
int new_cap = mgr->capacity > 0 ? mgr->capacity : TRIGGER_DEFAULT_MAX_ACTIVE;
while (new_cap < needed) {
if (new_cap > 1024) {
return -1;
}
new_cap *= 2;
}
active_trigger_t* grown = (active_trigger_t*)realloc(mgr->triggers, (size_t)new_cap * sizeof(active_trigger_t));
if (!grown) {
return -1;
}
if (new_cap > mgr->capacity) {
memset(&grown[mgr->capacity], 0, (size_t)(new_cap - mgr->capacity) * sizeof(active_trigger_t));
}
mgr->triggers = grown;
mgr->capacity = new_cap;
return 0;
}
static int find_trigger_index_locked(trigger_manager_t* mgr, const char* skill_d_tag) {
if (!mgr || !skill_d_tag || skill_d_tag[0] == '\0') {
return -1;
}
for (int i = 0; i < mgr->count; i++) {
if (strncmp(mgr->triggers[i].skill_d_tag, skill_d_tag, TRIGGER_SKILL_D_TAG_MAX) == 0) {
return i;
}
}
return -1;
}
static cJSON* find_tag_value_string(cJSON* tags, const char* key) {
if (!tags || !key || !cJSON_IsArray(tags)) return NULL;
int n = cJSON_GetArraySize(tags);
for (int i = 0; i < n; i++) {
cJSON* tag = cJSON_GetArrayItem(tags, i);
if (!tag || !cJSON_IsArray(tag) || cJSON_GetArraySize(tag) < 2) continue;
cJSON* k = cJSON_GetArrayItem(tag, 0);
cJSON* v = cJSON_GetArrayItem(tag, 1);
if (k && v && cJSON_IsString(k) && cJSON_IsString(v) &&
k->valuestring && v->valuestring && strcmp(k->valuestring, key) == 0) {
return v;
}
}
return NULL;
}
static __attribute__((unused)) int parse_address_tag(const char* addr, int* out_kind, char out_pubkey[65], char out_d_tag[65]) {
if (!addr || !out_kind || !out_pubkey || !out_d_tag) {
return -1;
}
const char* p1 = strchr(addr, ':');
if (!p1) return -1;
const char* p2 = strchr(p1 + 1, ':');
if (!p2) return -1;
char kind_buf[16] = {0};
size_t kind_len = (size_t)(p1 - addr);
if (kind_len == 0 || kind_len >= sizeof(kind_buf)) return -1;
memcpy(kind_buf, addr, kind_len);
int kind = atoi(kind_buf);
if (kind != 31123 && kind != 31124) return -1;
size_t pub_len = (size_t)(p2 - (p1 + 1));
if (pub_len != 64U) return -1;
memcpy(out_pubkey, p1 + 1, 64U);
out_pubkey[64] = '\0';
size_t d_tag_len = strlen(p2 + 1);
if (d_tag_len == 0 || d_tag_len >= 65U) return -1;
memcpy(out_d_tag, p2 + 1, d_tag_len + 1U);
*out_kind = kind;
return 0;
}
static const char* trim_ws_local(const char* s) {
while (s && *s && isspace((unsigned char)*s)) s++;
return s ? s : "";
}
static void parse_trigger_runtime_tags(cJSON* tags,
const char** out_llm,
const char** out_tools,
int* out_has_max_tokens,
int* out_max_tokens,
int* out_has_temperature,
double* out_temperature,
int* out_has_seed,
int* out_seed) {
if (out_llm) *out_llm = NULL;
if (out_tools) *out_tools = NULL;
if (out_has_max_tokens) *out_has_max_tokens = 0;
if (out_max_tokens) *out_max_tokens = 0;
if (out_has_temperature) *out_has_temperature = 0;
if (out_temperature) *out_temperature = 0.0;
if (out_has_seed) *out_has_seed = 0;
if (out_seed) *out_seed = 0;
if (!tags || !cJSON_IsArray(tags)) {
return;
}
cJSON* llm = find_tag_value_string(tags, "llm");
cJSON* tools = find_tag_value_string(tags, "tools");
cJSON* max_tokens = find_tag_value_string(tags, "max_tokens");
cJSON* temperature = find_tag_value_string(tags, "temperature");
cJSON* seed = find_tag_value_string(tags, "seed");
if (out_llm && llm && cJSON_IsString(llm) && llm->valuestring && llm->valuestring[0] != '\0') {
*out_llm = llm->valuestring;
}
if (out_tools && tools && cJSON_IsString(tools) && tools->valuestring && tools->valuestring[0] != '\0') {
*out_tools = tools->valuestring;
}
if (out_has_max_tokens && out_max_tokens && max_tokens && cJSON_IsString(max_tokens) && max_tokens->valuestring) {
char* endptr = NULL;
long n = strtol(trim_ws_local(max_tokens->valuestring), &endptr, 10);
if (endptr && *trim_ws_local(endptr) == '\0' && n > 0 && n <= 1000000) {
*out_has_max_tokens = 1;
*out_max_tokens = (int)n;
}
}
if (out_has_temperature && out_temperature && temperature && cJSON_IsString(temperature) && temperature->valuestring) {
char* endptr = NULL;
double n = strtod(trim_ws_local(temperature->valuestring), &endptr);
if (endptr && *trim_ws_local(endptr) == '\0' && n >= 0.0 && n <= 2.0) {
*out_has_temperature = 1;
*out_temperature = n;
}
}
if (out_has_seed && out_seed && seed && cJSON_IsString(seed) && seed->valuestring) {
char* endptr = NULL;
long n = strtol(trim_ws_local(seed->valuestring), &endptr, 10);
if (endptr && *trim_ws_local(endptr) == '\0') {
*out_has_seed = 1;
*out_seed = (int)n;
}
}
}
static int trigger_manager_reconcile_from_relays(trigger_manager_t* mgr) {
if (!mgr || !mgr->cfg) {
return -1;
}
cJSON* filter = cJSON_CreateObject();
cJSON* kinds = cJSON_CreateArray();
cJSON* authors = cJSON_CreateArray();
if (!filter || !kinds || !authors) {
cJSON_Delete(filter);
cJSON_Delete(kinds);
cJSON_Delete(authors);
return -1;
}
cJSON_AddItemToArray(kinds, cJSON_CreateNumber(31123));
cJSON_AddItemToArray(kinds, cJSON_CreateNumber(31124));
cJSON_AddItemToObject(filter, "kinds", kinds);
cJSON_AddItemToArray(authors, cJSON_CreateString(mgr->cfg->keys.public_key_hex));
cJSON_AddItemToObject(filter, "authors", authors);
cJSON_AddNumberToObject(filter, "limit", 300);
char* events_json = nostr_handler_query_json(filter, 8000);
cJSON_Delete(filter);
if (!events_json) {
DEBUG_WARN("[didactyl] trigger reconcile skipped: relay query returned no data");
return -1;
}
cJSON* events = cJSON_Parse(events_json);
free(events_json);
if (!events || !cJSON_IsArray(events)) {
cJSON_Delete(events);
DEBUG_WARN("[didactyl] trigger reconcile skipped: invalid JSON from relay query");
return -1;
}
int considered = 0;
int loaded = 0;
int event_count = cJSON_GetArraySize(events);
for (int i = 0; i < event_count; i++) {
cJSON* skill_event = cJSON_GetArrayItem(events, i);
cJSON* content = skill_event ? cJSON_GetObjectItemCaseSensitive(skill_event, "content") : NULL;
cJSON* tags = skill_event ? cJSON_GetObjectItemCaseSensitive(skill_event, "tags") : NULL;
if (!content || !cJSON_IsString(content) || !content->valuestring || !tags || !cJSON_IsArray(tags)) {
continue;
}
cJSON* d = find_tag_value_string(tags, "d");
cJSON* trigger = find_tag_value_string(tags, "trigger");
cJSON* filter_j = find_tag_value_string(tags, "filter");
cJSON* action = find_tag_value_string(tags, "action");
const char* d_tag = (d && cJSON_IsString(d) && d->valuestring) ? d->valuestring : NULL;
const char* trigger_s = (trigger && cJSON_IsString(trigger) && trigger->valuestring) ? trigger->valuestring : NULL;
const char* filter_s = (filter_j && cJSON_IsString(filter_j) && filter_j->valuestring) ? filter_j->valuestring : NULL;
const char* action_s = (action && cJSON_IsString(action) && action->valuestring) ? action->valuestring : "llm";
considered++;
if (!d_tag || d_tag[0] == '\0') {
continue;
}
int trigger_supported = trigger_s &&
(strcmp(trigger_s, "nostr-subscription") == 0 ||
strcmp(trigger_s, "webhook") == 0 ||
strcmp(trigger_s, "cron") == 0 ||
strcmp(trigger_s, "chain") == 0 ||
strcmp(trigger_s, "dm") == 0);
if (!trigger_supported || !filter_s || filter_s[0] == '\0') {
continue;
}
if (strcmp(action_s, "template") == 0) {
DEBUG_WARN("[didactyl] trigger action template is deprecated; forcing llm for d_tag=%s", d_tag);
}
const char* llm_s = NULL;
const char* tools_s = NULL;
int has_max_tokens = 0;
int max_tokens = 0;
int has_temperature = 0;
double temperature = 0.0;
int has_seed = 0;
int seed = 0;
parse_trigger_runtime_tags(tags,
&llm_s,
&tools_s,
&has_max_tokens,
&max_tokens,
&has_temperature,
&temperature,
&has_seed,
&seed);
if (trigger_manager_add(mgr,
d_tag,
content->valuestring,
filter_s,
TRIGGER_ACTION_LLM,
trigger_s,
1,
llm_s,
tools_s,
has_max_tokens,
max_tokens,
has_temperature,
temperature,
has_seed,
seed) == 0) {
loaded++;
}
}
cJSON_Delete(events);
DEBUG_INFO("[didactyl] trigger reconcile complete: loaded=%d considered=%d", loaded, considered);
return 0;
}
static void apply_trigger_runtime_to_llm_config(const active_trigger_t* t, llm_config_t* cfg) {
if (!t || !cfg) return;
if (t->llm_spec[0] != '\0') {
const char* spec = t->llm_spec;
while (*spec && isspace((unsigned char)*spec)) spec++;
const char* comma = strchr(spec, ',');
size_t len = comma ? (size_t)(comma - spec) : strlen(spec);
while (len > 0 && isspace((unsigned char)spec[len - 1])) len--;
const char* slash = memchr(spec, '/', len);
if (slash) {
size_t provider_len = (size_t)(slash - spec);
const char* model = slash + 1;
size_t model_len = len - (size_t)(model - spec);
while (provider_len > 0 && isspace((unsigned char)spec[provider_len - 1])) provider_len--;
while (model_len > 0 && isspace((unsigned char)*model)) {
model++;
model_len--;
}
while (model_len > 0 && isspace((unsigned char)model[model_len - 1])) model_len--;
if (provider_len > 0 && provider_len < sizeof(cfg->provider)) {
snprintf(cfg->provider, sizeof(cfg->provider), "%.*s", (int)provider_len, spec);
}
if (model_len > 0 && model_len < sizeof(cfg->model)) {
snprintf(cfg->model, sizeof(cfg->model), "%.*s", (int)model_len, model);
}
} else if (len > 0 && len < sizeof(cfg->model)) {
snprintf(cfg->model, sizeof(cfg->model), "%.*s", (int)len, spec);
}
}
if (t->has_max_tokens && t->max_tokens > 0) {
cfg->max_tokens = t->max_tokens;
}
if (t->has_temperature) {
cfg->temperature = t->temperature;
}
}
static void execute_llm_action(const active_trigger_t* t, cJSON* event, const char* relay_url) {
if (!t || !event) {
return;
}
llm_config_t old_cfg;
int had_old_cfg = (llm_get_config(&old_cfg) == 0);
int overridden = 0;
if (had_old_cfg &&
(t->llm_spec[0] != '\0' || t->has_max_tokens || t->has_temperature)) {
llm_config_t next = old_cfg;
apply_trigger_runtime_to_llm_config(t, &next);
if (llm_set_config(&next) == 0) {
overridden = 1;
}
}
agent_on_trigger(t->skill_d_tag, t->skill_content, event, relay_url);
if (overridden) {
(void)llm_set_config(&old_cfg);
}
}
static int maybe_fire_trigger_locked(trigger_manager_t* mgr, int index, cJSON* event, const char* relay_url) {
active_trigger_t* t = &mgr->triggers[index];
if (!t->enabled) {
return 0;
}
cJSON* created_at = cJSON_GetObjectItemCaseSensitive(event, "created_at");
time_t created_ts = (created_at && cJSON_IsNumber(created_at)) ? (time_t)created_at->valuedouble : 0;
if (created_ts > 0 && created_ts <= t->last_seen_created_at) {
return 0;
}
time_t now = time(NULL);
int cooldown = mgr->cfg->triggers.cooldown_seconds;
if (cooldown < 0) cooldown = 0;
if (cooldown > 0 && t->last_fired > 0 && (now - t->last_fired) < cooldown) {
return 0;
}
t->last_fired = now;
if (created_ts > t->last_seen_created_at) {
t->last_seen_created_at = created_ts;
}
active_trigger_t trigger_copy = *t;
pthread_mutex_unlock(&mgr->mutex);
(void)mgr;
execute_llm_action(&trigger_copy, event, relay_url);
pthread_mutex_lock(&mgr->mutex);
return 1;
}
static void on_trigger_subscription_eose(cJSON** events, int event_count, void* user_data) {
(void)events;
(void)event_count;
(void)user_data;
}
static void close_trigger_subscription_locked(active_trigger_t* t) {
if (!t) {
return;
}
if (t->subscription) {
(void)nostr_handler_close_subscription(t->subscription);
t->subscription = NULL;
}
if (t->subscription_ctx) {
free(t->subscription_ctx);
t->subscription_ctx = NULL;
}
}
static void on_trigger_subscription_event(cJSON* event, const char* relay_url, void* user_data) {
trigger_subscription_ctx_t* ctx = (trigger_subscription_ctx_t*)user_data;
if (!ctx || !ctx->mgr || !event) {
return;
}
(void)trigger_manager_fire(ctx->mgr, ctx->skill_d_tag, event, relay_url);
}
static int register_trigger_subscription_locked(trigger_manager_t* mgr, active_trigger_t* t) {
if (!mgr || !t) {
return -1;
}
close_trigger_subscription_locked(t);
if (t->trigger_type != TRIGGER_TYPE_NOSTR_SUBSCRIPTION) {
return 0;
}
if (!t->enabled || t->filter_json[0] == '\0') {
return 0;
}
cJSON* filter = cJSON_Parse(t->filter_json);
if (!filter || !cJSON_IsObject(filter)) {
cJSON_Delete(filter);
return -1;
}
cJSON* since = cJSON_GetObjectItemCaseSensitive(filter, "since");
if (!since || !cJSON_IsNumber(since)) {
time_t now = time(NULL);
time_t default_since = now > 30 ? now - 30 : 0;
cJSON_AddNumberToObject(filter, "since", (double)default_since);
}
cJSON* limit = cJSON_GetObjectItemCaseSensitive(filter, "limit");
if (!limit || !cJSON_IsNumber(limit)) {
cJSON_AddNumberToObject(filter, "limit", 200);
}
trigger_subscription_ctx_t* ctx = (trigger_subscription_ctx_t*)calloc(1, sizeof(*ctx));
if (!ctx) {
cJSON_Delete(filter);
return -1;
}
ctx->mgr = mgr;
snprintf(ctx->skill_d_tag, sizeof(ctx->skill_d_tag), "%s", t->skill_d_tag);
nostr_pool_subscription_t* sub = nostr_handler_subscribe_with_filter(
filter,
on_trigger_subscription_event,
on_trigger_subscription_eose,
ctx,
0,
0,
NOSTR_POOL_EOSE_FULL_SET,
30,
120);
cJSON_Delete(filter);
if (!sub) {
free(ctx);
return -1;
}
t->subscription = sub;
t->subscription_ctx = ctx;
return 0;
}
int trigger_manager_init(trigger_manager_t* mgr, didactyl_config_t* cfg) {
if (!mgr || !cfg) {
return -1;
}
memset(mgr, 0, sizeof(*mgr));
mgr->cfg = cfg;
if (pthread_mutex_init(&mgr->mutex, NULL) != 0) {
return -1;
}
int initial_cap = cfg->triggers.max_active > 0 ? cfg->triggers.max_active : TRIGGER_DEFAULT_MAX_ACTIVE;
mgr->capacity = initial_cap;
mgr->triggers = (active_trigger_t*)calloc((size_t)mgr->capacity, sizeof(active_trigger_t));
if (!mgr->triggers) {
pthread_mutex_destroy(&mgr->mutex);
memset(mgr, 0, sizeof(*mgr));
return -1;
}
mgr->last_poll_at = 0;
mgr->last_reconcile_at = 0;
DEBUG_INFO("[didactyl] trigger manager initialized (capacity=%d)", mgr->capacity);
return 0;
}
int trigger_manager_load_from_skills(trigger_manager_t* mgr) {
if (!mgr || !mgr->cfg) {
return -1;
}
int loaded = 0;
int considered = 0;
const int kinds[2] = {31123, 31124};
for (int k = 0; k < 2; k++) {
char* skill_json = nostr_handler_get_self_events_by_kind_json(kinds[k]);
if (!skill_json) {
continue;
}
cJSON* skill_events = cJSON_Parse(skill_json);
free(skill_json);
if (!skill_events || !cJSON_IsArray(skill_events)) {
cJSON_Delete(skill_events);
continue;
}
int sn = cJSON_GetArraySize(skill_events);
for (int i = 0; i < sn; i++) {
cJSON* skill_event = cJSON_GetArrayItem(skill_events, i);
cJSON* content = skill_event ? cJSON_GetObjectItemCaseSensitive(skill_event, "content") : NULL;
cJSON* tags = skill_event ? cJSON_GetObjectItemCaseSensitive(skill_event, "tags") : NULL;
if (!content || !cJSON_IsString(content) || !content->valuestring || !tags || !cJSON_IsArray(tags)) {
continue;
}
cJSON* d = find_tag_value_string(tags, "d");
cJSON* trigger = find_tag_value_string(tags, "trigger");
cJSON* filter = find_tag_value_string(tags, "filter");
cJSON* action = find_tag_value_string(tags, "action");
const char* d_tag = (d && cJSON_IsString(d) && d->valuestring) ? d->valuestring : NULL;
const char* trigger_s = (trigger && cJSON_IsString(trigger) && trigger->valuestring) ? trigger->valuestring : NULL;
const char* filter_s = (filter && cJSON_IsString(filter) && filter->valuestring) ? filter->valuestring : NULL;
const char* action_s = (action && cJSON_IsString(action) && action->valuestring) ? action->valuestring : "llm";
considered++;
if (!d_tag || d_tag[0] == '\0') {
continue;
}
int trigger_supported = trigger_s &&
(strcmp(trigger_s, "nostr-subscription") == 0 ||
strcmp(trigger_s, "webhook") == 0 ||
strcmp(trigger_s, "cron") == 0 ||
strcmp(trigger_s, "chain") == 0 ||
strcmp(trigger_s, "dm") == 0);
if (!trigger_supported || !filter_s || filter_s[0] == '\0') {
continue;
}
if (strcmp(action_s, "template") == 0) {
DEBUG_WARN("[didactyl] trigger action template is deprecated; forcing llm for d_tag=%s", d_tag);
}
const char* llm_s = NULL;
const char* tools_s = NULL;
int has_max_tokens = 0;
int max_tokens = 0;
int has_temperature = 0;
double temperature = 0.0;
int has_seed = 0;
int seed = 0;
parse_trigger_runtime_tags(tags,
&llm_s,
&tools_s,
&has_max_tokens,
&max_tokens,
&has_temperature,
&temperature,
&has_seed,
&seed);
if (trigger_manager_add(mgr,
d_tag,
content->valuestring,
filter_s,
TRIGGER_ACTION_LLM,
trigger_s,
1,
llm_s,
tools_s,
has_max_tokens,
max_tokens,
has_temperature,
temperature,
has_seed,
seed) == 0) {
loaded++;
}
}
cJSON_Delete(skill_events);
}
DEBUG_INFO("[didactyl] trigger manager loaded %d trigger(s) from self skills (considered=%d)", loaded, considered);
return 0;
}
int trigger_manager_load_from_startup_events(trigger_manager_t* mgr) {
if (!mgr || !mgr->cfg) {
return -1;
}
int loaded = 0;
int considered = 0;
for (int i = 0; i < mgr->cfg->startup_event_count; i++) {
startup_event_t* ev = &mgr->cfg->startup_events[i];
if (!ev) {
continue;
}
if (ev->kind != 31123 && ev->kind != 31124) {
continue;
}
considered++;
if (!ev->content || ev->content[0] == '\0' || !ev->tags_json || ev->tags_json[0] == '\0') {
continue;
}
cJSON* tags = cJSON_Parse(ev->tags_json);
if (!tags || !cJSON_IsArray(tags)) {
cJSON_Delete(tags);
DEBUG_WARN("[didactyl] startup trigger skip: invalid tags_json for startup event index=%d", i);
continue;
}
cJSON* d = find_tag_value_string(tags, "d");
cJSON* trigger = find_tag_value_string(tags, "trigger");
cJSON* filter = find_tag_value_string(tags, "filter");
cJSON* action = find_tag_value_string(tags, "action");
cJSON* enabled = find_tag_value_string(tags, "enabled");
const char* d_tag = (d && cJSON_IsString(d) && d->valuestring) ? d->valuestring : NULL;
const char* trigger_s = (trigger && cJSON_IsString(trigger) && trigger->valuestring) ? trigger->valuestring : NULL;
const char* filter_s = (filter && cJSON_IsString(filter) && filter->valuestring) ? filter->valuestring : NULL;
const char* action_s = (action && cJSON_IsString(action) && action->valuestring) ? action->valuestring : "llm";
const char* enabled_s = (enabled && cJSON_IsString(enabled) && enabled->valuestring) ? enabled->valuestring : "true";
if (!d_tag || d_tag[0] == '\0') {
cJSON_Delete(tags);
DEBUG_WARN("[didactyl] startup trigger skip: missing d tag for startup event index=%d", i);
continue;
}
int trigger_supported = trigger_s &&
(strcmp(trigger_s, "nostr-subscription") == 0 ||
strcmp(trigger_s, "webhook") == 0 ||
strcmp(trigger_s, "cron") == 0 ||
strcmp(trigger_s, "chain") == 0 ||
strcmp(trigger_s, "dm") == 0);
if (!trigger_supported || !filter_s || filter_s[0] == '\0') {
cJSON_Delete(tags);
continue;
}
if (strcmp(action_s, "template") == 0) {
DEBUG_WARN("[didactyl] startup trigger action template is deprecated; forcing llm for d_tag=%s", d_tag);
}
int is_enabled = (strcmp(enabled_s, "false") == 0 || strcmp(enabled_s, "0") == 0) ? 0 : 1;
const char* llm_s = NULL;
const char* tools_s = NULL;
int has_max_tokens = 0;
int max_tokens = 0;
int has_temperature = 0;
double temperature = 0.0;
int has_seed = 0;
int seed = 0;
parse_trigger_runtime_tags(tags,
&llm_s,
&tools_s,
&has_max_tokens,
&max_tokens,
&has_temperature,
&temperature,
&has_seed,
&seed);
if (trigger_manager_add(mgr,
d_tag,
ev->content,
filter_s,
TRIGGER_ACTION_LLM,
trigger_s,
is_enabled,
llm_s,
tools_s,
has_max_tokens,
max_tokens,
has_temperature,
temperature,
has_seed,
seed) == 0) {
loaded++;
DEBUG_INFO("[didactyl] startup trigger registered d_tag=%s action=%s enabled=%d", d_tag, "llm", is_enabled);
} else {
DEBUG_WARN("[didactyl] startup trigger failed d_tag=%s", d_tag);
}
cJSON_Delete(tags);
}
DEBUG_INFO("[didactyl] trigger manager loaded %d trigger(s) from startup config (considered=%d)", loaded, considered);
return 0;
}
int trigger_manager_add(trigger_manager_t* mgr,
const char* skill_d_tag,
const char* content,
const char* filter_json,
trigger_action_type_t action_type,
const char* trigger_type_str,
int enabled,
const char* llm_spec,
const char* tools_policy,
int has_max_tokens,
int max_tokens,
int has_temperature,
double temperature,
int has_seed,
int seed) {
if (!mgr || !skill_d_tag || skill_d_tag[0] == '\0' || !content || !filter_json) {
return -1;
}
if (action_type == TRIGGER_ACTION_TEMPLATE) {
DEBUG_WARN("[didactyl] trigger action template is deprecated; forcing llm for d_tag=%s", skill_d_tag);
action_type = TRIGGER_ACTION_LLM;
}
if (action_type != TRIGGER_ACTION_LLM) {
return -1;
}
pthread_mutex_lock(&mgr->mutex);
int existing = find_trigger_index_locked(mgr, skill_d_tag);
if (existing >= 0) {
pthread_mutex_unlock(&mgr->mutex);
return trigger_manager_update(mgr,
skill_d_tag,
content,
filter_json,
action_type,
trigger_type_str,
enabled,
llm_spec,
tools_policy,
has_max_tokens,
max_tokens,
has_temperature,
temperature,
has_seed,
seed);
}
int max_active = mgr->cfg ? mgr->cfg->triggers.max_active : TRIGGER_DEFAULT_MAX_ACTIVE;
if (max_active < 1) max_active = TRIGGER_DEFAULT_MAX_ACTIVE;
if (mgr->count >= max_active) {
pthread_mutex_unlock(&mgr->mutex);
DEBUG_WARN("[didactyl] trigger add rejected: max_active reached (%d)", max_active);
return -1;
}
if (ensure_capacity(mgr, mgr->count + 1) != 0) {
pthread_mutex_unlock(&mgr->mutex);
return -1;
}
active_trigger_t* t = &mgr->triggers[mgr->count];
memset(t, 0, sizeof(*t));
snprintf(t->skill_d_tag, sizeof(t->skill_d_tag), "%s", skill_d_tag);
snprintf(t->skill_content, sizeof(t->skill_content), "%s", content);
snprintf(t->filter_json, sizeof(t->filter_json), "%s", filter_json);
t->action_type = action_type;
t->enabled = clamp_enabled(enabled);
t->trigger_type = trigger_type_from_string(trigger_type_str);
t->last_fired = 0;
t->last_seen_created_at = 0;
t->last_cron_fire = 0;
if (t->trigger_type == TRIGGER_TYPE_CRON) {
char normalized_cron[sizeof(t->cron_expr)] = {0};
if (!cron_expr_is_valid(filter_json, normalized_cron, sizeof(normalized_cron))) {
pthread_mutex_unlock(&mgr->mutex);
DEBUG_WARN("[didactyl] trigger add rejected: invalid cron expression d_tag=%s expr=%s", skill_d_tag, filter_json);
memset(t, 0, sizeof(*t));
return -1;
}
snprintf(t->cron_expr, sizeof(t->cron_expr), "%s", normalized_cron);
}
snprintf(t->llm_spec, sizeof(t->llm_spec), "%s", llm_spec ? llm_spec : "");
snprintf(t->tools_policy, sizeof(t->tools_policy), "%s", tools_policy ? tools_policy : "");
t->has_max_tokens = has_max_tokens ? 1 : 0;
t->max_tokens = max_tokens;
t->has_temperature = has_temperature ? 1 : 0;
t->temperature = temperature;
t->has_seed = has_seed ? 1 : 0;
t->seed = seed;
t->subscription = NULL;
t->subscription_ctx = NULL;
if (register_trigger_subscription_locked(mgr, t) != 0) {
pthread_mutex_unlock(&mgr->mutex);
DEBUG_WARN("[didactyl] trigger add rejected: failed to create subscription d_tag=%s", skill_d_tag);
memset(t, 0, sizeof(*t));
return -1;
}
mgr->count++;
pthread_mutex_unlock(&mgr->mutex);
DEBUG_INFO("[didactyl] trigger added d_tag=%s action=%d enabled=%d", skill_d_tag, (int)action_type, clamp_enabled(enabled));
return 0;
}
int trigger_manager_remove(trigger_manager_t* mgr, const char* skill_d_tag) {
if (!mgr || !skill_d_tag || skill_d_tag[0] == '\0') {
return -1;
}
pthread_mutex_lock(&mgr->mutex);
int idx = find_trigger_index_locked(mgr, skill_d_tag);
if (idx < 0) {
pthread_mutex_unlock(&mgr->mutex);
return 0;
}
close_trigger_subscription_locked(&mgr->triggers[idx]);
if (idx < mgr->count - 1) {
memmove(&mgr->triggers[idx],
&mgr->triggers[idx + 1],
(size_t)(mgr->count - idx - 1) * sizeof(active_trigger_t));
}
mgr->count--;
memset(&mgr->triggers[mgr->count], 0, sizeof(active_trigger_t));
pthread_mutex_unlock(&mgr->mutex);
DEBUG_INFO("[didactyl] trigger removed d_tag=%s", skill_d_tag);
return 0;
}
int trigger_manager_find(trigger_manager_t* mgr, const char* skill_d_tag, active_trigger_t* out) {
if (!mgr || !skill_d_tag || skill_d_tag[0] == '\0' || !out) {
return -1;
}
pthread_mutex_lock(&mgr->mutex);
int idx = find_trigger_index_locked(mgr, skill_d_tag);
if (idx < 0) {
pthread_mutex_unlock(&mgr->mutex);
return -1;
}
*out = mgr->triggers[idx];
out->subscription = NULL;
out->subscription_ctx = NULL;
pthread_mutex_unlock(&mgr->mutex);
return 0;
}
int trigger_manager_fire(trigger_manager_t* mgr,
const char* skill_d_tag,
cJSON* event,
const char* source_label) {
if (!mgr || !skill_d_tag || skill_d_tag[0] == '\0' || !event) {
return -1;
}
pthread_mutex_lock(&mgr->mutex);
int idx = find_trigger_index_locked(mgr, skill_d_tag);
if (idx < 0) {
pthread_mutex_unlock(&mgr->mutex);
return -1;
}
int fired = maybe_fire_trigger_locked(mgr, idx, event, source_label);
pthread_mutex_unlock(&mgr->mutex);
return fired;
}
int trigger_manager_fire_chains(trigger_manager_t* mgr,
const char* source_skill_d_tag,
cJSON* source_event,
const char* source_label) {
if (!mgr || !source_skill_d_tag || source_skill_d_tag[0] == '\0') {
return -1;
}
static __thread int s_chain_depth = 0;
if (s_chain_depth >= 5) {
DEBUG_WARN("[didactyl] chain trigger depth limit reached for source=%s", source_skill_d_tag);
return 0;
}
int fired_count = 0;
s_chain_depth++;
pthread_mutex_lock(&mgr->mutex);
int count_snapshot = mgr->count;
for (int i = 0; i < count_snapshot; i++) {
active_trigger_t* t = &mgr->triggers[i];
if (!t->enabled || t->trigger_type != TRIGGER_TYPE_CHAIN) {
continue;
}
if (t->filter_json[0] == '\0' || strcmp(t->filter_json, source_skill_d_tag) != 0) {
continue;
}
time_t now = time(NULL);
int cooldown = mgr->cfg ? mgr->cfg->triggers.cooldown_seconds : 0;
if (cooldown < 0) cooldown = 0;
if (cooldown > 0 && t->last_fired > 0 && (now - t->last_fired) < cooldown) {
continue;
}
t->last_fired = now;
active_trigger_t trigger_copy = *t;
trigger_copy.subscription = NULL;
trigger_copy.subscription_ctx = NULL;
pthread_mutex_unlock(&mgr->mutex);
cJSON* event = cJSON_CreateObject();
if (event) {
cJSON_AddStringToObject(event, "type", "chain");
cJSON_AddStringToObject(event, "source_d_tag", source_skill_d_tag);
cJSON_AddStringToObject(event, "source_label", source_label ? source_label : "chain");
cJSON_AddNumberToObject(event, "created_at", (double)now);
if (source_event) {
cJSON* dup = cJSON_Duplicate(source_event, 1);
if (dup) {
cJSON_AddItemToObject(event, "source_event", dup);
}
}
(void)mgr;
execute_llm_action(&trigger_copy, event, "chain");
cJSON_Delete(event);
fired_count++;
}
pthread_mutex_lock(&mgr->mutex);
}
pthread_mutex_unlock(&mgr->mutex);
s_chain_depth--;
return fired_count;
}
int trigger_manager_update(trigger_manager_t* mgr,
const char* skill_d_tag,
const char* content,
const char* filter_json,
trigger_action_type_t action_type,
const char* trigger_type_str,
int enabled,
const char* llm_spec,
const char* tools_policy,
int has_max_tokens,
int max_tokens,
int has_temperature,
double temperature,
int has_seed,
int seed) {
if (!mgr || !skill_d_tag || skill_d_tag[0] == '\0' || !content || !filter_json) {
return -1;
}
if (action_type == TRIGGER_ACTION_TEMPLATE) {
DEBUG_WARN("[didactyl] trigger action template is deprecated; forcing llm for d_tag=%s", skill_d_tag);
action_type = TRIGGER_ACTION_LLM;
}
if (action_type != TRIGGER_ACTION_LLM) {
return -1;
}
pthread_mutex_lock(&mgr->mutex);
int idx = find_trigger_index_locked(mgr, skill_d_tag);
if (idx < 0) {
pthread_mutex_unlock(&mgr->mutex);
return trigger_manager_add(mgr,
skill_d_tag,
content,
filter_json,
action_type,
trigger_type_str,
enabled,
llm_spec,
tools_policy,
has_max_tokens,
max_tokens,
has_temperature,
temperature,
has_seed,
seed);
}
active_trigger_t* t = &mgr->triggers[idx];
snprintf(t->skill_content, sizeof(t->skill_content), "%s", content);
snprintf(t->filter_json, sizeof(t->filter_json), "%s", filter_json);
t->action_type = action_type;
t->enabled = clamp_enabled(enabled);
t->trigger_type = trigger_type_from_string(trigger_type_str);
if (t->trigger_type == TRIGGER_TYPE_CRON) {
char normalized_cron[sizeof(t->cron_expr)] = {0};
if (!cron_expr_is_valid(filter_json, normalized_cron, sizeof(normalized_cron))) {
pthread_mutex_unlock(&mgr->mutex);
DEBUG_WARN("[didactyl] trigger update rejected: invalid cron expression d_tag=%s expr=%s", skill_d_tag, filter_json);
return -1;
}
snprintf(t->cron_expr, sizeof(t->cron_expr), "%s", normalized_cron);
} else {
t->cron_expr[0] = '\0';
}
snprintf(t->llm_spec, sizeof(t->llm_spec), "%s", llm_spec ? llm_spec : "");
snprintf(t->tools_policy, sizeof(t->tools_policy), "%s", tools_policy ? tools_policy : "");
t->has_max_tokens = has_max_tokens ? 1 : 0;
t->max_tokens = max_tokens;
t->has_temperature = has_temperature ? 1 : 0;
t->temperature = temperature;
t->has_seed = has_seed ? 1 : 0;
t->seed = seed;
if (register_trigger_subscription_locked(mgr, t) != 0) {
pthread_mutex_unlock(&mgr->mutex);
DEBUG_WARN("[didactyl] trigger update failed to (re)subscribe d_tag=%s", skill_d_tag);
return -1;
}
pthread_mutex_unlock(&mgr->mutex);
DEBUG_INFO("[didactyl] trigger updated d_tag=%s action=%d enabled=%d", skill_d_tag, (int)action_type, clamp_enabled(enabled));
return 0;
}
int trigger_manager_active_count(trigger_manager_t* mgr) {
if (!mgr) {
return 0;
}
int active = 0;
pthread_mutex_lock(&mgr->mutex);
for (int i = 0; i < mgr->count; i++) {
if (mgr->triggers[i].enabled) {
active++;
}
}
pthread_mutex_unlock(&mgr->mutex);
return active;
}
int trigger_manager_poll(trigger_manager_t* mgr) {
if (!mgr) {
return -1;
}
time_t now = time(NULL);
if (now <= 0) {
return 0;
}
if (mgr->last_poll_at > 0 && (now - mgr->last_poll_at) < 30) {
return 0;
}
mgr->last_poll_at = now;
int fired = 0;
pthread_mutex_lock(&mgr->mutex);
int count_snapshot = mgr->count;
for (int i = 0; i < count_snapshot; i++) {
active_trigger_t* t = &mgr->triggers[i];
if (!t->enabled || t->trigger_type != TRIGGER_TYPE_CRON) {
continue;
}
const char* expr = (t->cron_expr[0] != '\0') ? t->cron_expr : t->filter_json;
char normalized_expr[64] = {0};
if (!cron_expr_is_valid(expr, normalized_expr, sizeof(normalized_expr))) {
DEBUG_WARN("[didactyl] cron trigger skipped: invalid expression d_tag=%s expr=%s", t->skill_d_tag, expr ? expr : "");
continue;
}
if (!cron_matches_now(normalized_expr, now)) {
continue;
}
DEBUG_INFO("[didactyl] cron trigger matched d_tag=%s expr=%s", t->skill_d_tag, normalized_expr);
if (t->last_cron_fire > 0 && (now - t->last_cron_fire) < 50) {
continue;
}
t->last_cron_fire = now;
active_trigger_t trigger_copy = *t;
trigger_copy.subscription = NULL;
trigger_copy.subscription_ctx = NULL;
pthread_mutex_unlock(&mgr->mutex);
cJSON* event = cJSON_CreateObject();
if (event) {
cJSON_AddStringToObject(event, "type", "cron");
cJSON_AddStringToObject(event, "d_tag", trigger_copy.skill_d_tag);
cJSON_AddStringToObject(event, "cron_expr", normalized_expr);
cJSON_AddNumberToObject(event, "created_at", (double)now);
(void)mgr;
execute_llm_action(&trigger_copy, event, "cron");
cJSON_Delete(event);
fired++;
}
pthread_mutex_lock(&mgr->mutex);
}
pthread_mutex_unlock(&mgr->mutex);
if (mgr->last_reconcile_at <= 0 || (now - mgr->last_reconcile_at) >= 900) {
mgr->last_reconcile_at = now;
(void)trigger_manager_reconcile_from_relays(mgr);
}
return fired;
}
static int dm_filter_matches_tier(const char* filter_json, didactyl_sender_tier_t tier) {
if (!filter_json || filter_json[0] == '\0') {
return 0;
}
cJSON* root = cJSON_Parse(filter_json);
if (!root || !cJSON_IsObject(root)) {
cJSON_Delete(root);
return 0;
}
cJSON* from = cJSON_GetObjectItemCaseSensitive(root, "from");
const char* from_s = (from && cJSON_IsString(from) && from->valuestring) ? from->valuestring : "admin";
int match = 0;
if (strcmp(from_s, "any") == 0) {
match = 1;
} else if (strcmp(from_s, "admin") == 0) {
match = (tier == DIDACTYL_SENDER_ADMIN);
} else if (strcmp(from_s, "wot") == 0) {
match = (tier == DIDACTYL_SENDER_WOT);
}
cJSON_Delete(root);
return match;
}
int trigger_manager_get_dm_skills(trigger_manager_t* mgr,
const char* sender_pubkey_hex,
didactyl_sender_tier_t tier,
char** out_d_tags,
int max_d_tags) {
(void)sender_pubkey_hex;
if (!mgr || !out_d_tags || max_d_tags <= 0) {
return -1;
}
int count = 0;
pthread_mutex_lock(&mgr->mutex);
int count_snapshot = mgr->count;
for (int i = 0; i < count_snapshot && count < max_d_tags; i++) {
active_trigger_t* t = &mgr->triggers[i];
if (!t->enabled || t->trigger_type != TRIGGER_TYPE_DM) {
continue;
}
if (!dm_filter_matches_tier(t->filter_json, tier)) {
continue;
}
out_d_tags[count++] = t->skill_d_tag;
}
pthread_mutex_unlock(&mgr->mutex);
return count;
}
char* trigger_manager_status_json(trigger_manager_t* mgr) {
if (!mgr) {
cJSON* err = cJSON_CreateObject();
if (!err) {
return NULL;
}
cJSON_AddBoolToObject(err, "success", 0);
cJSON_AddStringToObject(err, "error", "trigger manager unavailable");
char* out = cJSON_PrintUnformatted(err);
cJSON_Delete(err);
return out;
}
cJSON* root = cJSON_CreateObject();
cJSON* arr = cJSON_CreateArray();
if (!root || !arr) {
cJSON_Delete(root);
cJSON_Delete(arr);
return NULL;
}
pthread_mutex_lock(&mgr->mutex);
cJSON_AddBoolToObject(root, "success", 1);
cJSON_AddNumberToObject(root, "count", mgr->count);
int active = 0;
for (int i = 0; i < mgr->count; i++) {
active_trigger_t* t = &mgr->triggers[i];
if (t->enabled) {
active++;
}
cJSON* item = cJSON_CreateObject();
if (!item) {
continue;
}
cJSON_AddStringToObject(item, "skill_d_tag", t->skill_d_tag);
cJSON_AddStringToObject(item, "filter_json", t->filter_json);
cJSON_AddStringToObject(item, "type", trigger_type_to_string(t->trigger_type));
cJSON_AddStringToObject(item, "action", "llm");
cJSON_AddBoolToObject(item, "enabled", t->enabled ? 1 : 0);
cJSON_AddBoolToObject(item, "subscribed", t->subscription ? 1 : 0);
cJSON_AddStringToObject(item, "llm", t->llm_spec);
cJSON_AddStringToObject(item, "tools", t->tools_policy);
cJSON_AddBoolToObject(item, "has_max_tokens", t->has_max_tokens ? 1 : 0);
if (t->has_max_tokens) cJSON_AddNumberToObject(item, "max_tokens", t->max_tokens);
cJSON_AddBoolToObject(item, "has_temperature", t->has_temperature ? 1 : 0);
if (t->has_temperature) cJSON_AddNumberToObject(item, "temperature", t->temperature);
cJSON_AddBoolToObject(item, "has_seed", t->has_seed ? 1 : 0);
if (t->has_seed) cJSON_AddNumberToObject(item, "seed", t->seed);
cJSON_AddNumberToObject(item, "last_fired", (double)t->last_fired);
cJSON_AddNumberToObject(item, "last_seen_created_at", (double)t->last_seen_created_at);
if (t->trigger_type == TRIGGER_TYPE_CRON) {
cJSON_AddStringToObject(item, "cron_expr", t->cron_expr);
cJSON_AddNumberToObject(item, "last_cron_fire", (double)t->last_cron_fire);
}
cJSON_AddItemToArray(arr, item);
}
cJSON_AddNumberToObject(root, "active", active);
cJSON_AddItemToObject(root, "triggers", arr);
pthread_mutex_unlock(&mgr->mutex);
char* out = cJSON_PrintUnformatted(root);
cJSON_Delete(root);
return out;
}
void trigger_manager_cleanup(trigger_manager_t* mgr) {
if (!mgr) {
return;
}
pthread_mutex_lock(&mgr->mutex);
for (int i = 0; i < mgr->count; i++) {
close_trigger_subscription_locked(&mgr->triggers[i]);
}
free(mgr->triggers);
mgr->triggers = NULL;
mgr->count = 0;
mgr->capacity = 0;
mgr->cfg = NULL;
mgr->last_poll_at = 0;
pthread_mutex_unlock(&mgr->mutex);
pthread_mutex_destroy(&mgr->mutex);
memset(mgr, 0, sizeof(*mgr));
DEBUG_INFO("[didactyl] trigger manager cleaned up");
}
Reference in New Issue View Git Blame Copy Permalink