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

1476 lines
48 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_matches_now(const char* expr, time_t now_ts) {
if (!expr || expr[0] == '\0') {
return 0;
}
char buf[TRIGGER_FILTER_JSON_MAX];
snprintf(buf, sizeof(buf), "%s", expr);
char* fields[5] = {0};
int nf = 0;
char* saveptr = NULL;
char* tok = strtok_r(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 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 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;
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--;
while (*spec && isspace((unsigned char)*spec)) spec++;
const char* slash = memchr(spec, '/', len);
if (slash) {
const char* model = slash + 1;
size_t model_len = len - (size_t)(model - 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 = time(NULL);
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;
}
char* adoption_json = nostr_handler_get_self_events_by_kind_json(10123);
if (!adoption_json) {
return 0;
}
cJSON* adoption_events = cJSON_Parse(adoption_json);
free(adoption_json);
if (!adoption_events || !cJSON_IsArray(adoption_events) || cJSON_GetArraySize(adoption_events) <= 0) {
cJSON_Delete(adoption_events);
return 0;
}
cJSON* list_event = cJSON_GetArrayItem(adoption_events, 0);
cJSON* list_tags = list_event ? cJSON_GetObjectItemCaseSensitive(list_event, "tags") : NULL;
if (!list_tags || !cJSON_IsArray(list_tags)) {
cJSON_Delete(adoption_events);
return 0;
}
int loaded = 0;
int tn = cJSON_GetArraySize(list_tags);
for (int i = 0; i < tn; i++) {
cJSON* tag = cJSON_GetArrayItem(list_tags, i);
if (!tag || !cJSON_IsArray(tag) || cJSON_GetArraySize(tag) < 2) continue;
cJSON* key = cJSON_GetArrayItem(tag, 0);
cJSON* val = cJSON_GetArrayItem(tag, 1);
if (!key || !val || !cJSON_IsString(key) || !cJSON_IsString(val) ||
!key->valuestring || !val->valuestring || strcmp(key->valuestring, "a") != 0) {
continue;
}
int kind = 0;
char pubkey[65] = {0};
char d_tag[65] = {0};
if (parse_address_tag(val->valuestring, &kind, pubkey, d_tag) != 0) {
continue;
}
cJSON* skill_events = NULL;
if (strcmp(pubkey, mgr->cfg->keys.public_key_hex) == 0) {
char* skill_json = nostr_handler_get_self_events_by_kind_json(kind);
if (skill_json) {
cJSON* all_events = cJSON_Parse(skill_json);
free(skill_json);
if (all_events && cJSON_IsArray(all_events)) {
skill_events = cJSON_CreateArray();
if (skill_events) {
int all_n = cJSON_GetArraySize(all_events);
for (int ai = 0; ai < all_n; ai++) {
cJSON* ev = cJSON_GetArrayItem(all_events, ai);
cJSON* ev_pubkey = ev ? cJSON_GetObjectItemCaseSensitive(ev, "pubkey") : NULL;
cJSON* ev_tags = ev ? cJSON_GetObjectItemCaseSensitive(ev, "tags") : NULL;
cJSON* ev_d = find_tag_value_string(ev_tags, "d");
if (!ev_pubkey || !cJSON_IsString(ev_pubkey) || !ev_pubkey->valuestring ||
strcmp(ev_pubkey->valuestring, pubkey) != 0 ||
!ev_d || !cJSON_IsString(ev_d) || !ev_d->valuestring ||
strcmp(ev_d->valuestring, d_tag) != 0) {
continue;
}
cJSON* dup = cJSON_Duplicate(ev, 1);
if (dup) {
cJSON_AddItemToArray(skill_events, dup);
}
}
}
}
cJSON_Delete(all_events);
}
} else {
cJSON* skill_filter = cJSON_CreateObject();
cJSON* sk_kinds = cJSON_CreateArray();
cJSON* sk_authors = cJSON_CreateArray();
cJSON* d_values = cJSON_CreateArray();
if (!skill_filter || !sk_kinds || !sk_authors || !d_values) {
cJSON_Delete(skill_filter);
cJSON_Delete(sk_kinds);
cJSON_Delete(sk_authors);
cJSON_Delete(d_values);
continue;
}
cJSON_AddItemToArray(sk_kinds, cJSON_CreateNumber(kind));
cJSON_AddItemToObject(skill_filter, "kinds", sk_kinds);
cJSON_AddItemToArray(sk_authors, cJSON_CreateString(pubkey));
cJSON_AddItemToObject(skill_filter, "authors", sk_authors);
cJSON_AddItemToArray(d_values, cJSON_CreateString(d_tag));
cJSON_AddItemToObject(skill_filter, "#d", d_values);
cJSON_AddNumberToObject(skill_filter, "limit", 1);
char* skill_json = nostr_handler_query_json(skill_filter, 2000);
cJSON_Delete(skill_filter);
if (skill_json) {
skill_events = cJSON_Parse(skill_json);
free(skill_json);
}
}
if (!skill_events || !cJSON_IsArray(skill_events) || cJSON_GetArraySize(skill_events) <= 0) {
cJSON_Delete(skill_events);
continue;
}
cJSON* skill_event = cJSON_GetArrayItem(skill_events, 0);
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)) {
cJSON_Delete(skill_events);
continue;
}
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* 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";
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') {
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);
int is_enabled = (strcmp(enabled_s, "false") == 0 || strcmp(enabled_s, "0") == 0) ? 0 : 1;
if (trigger_manager_add(mgr,
d_tag,
content->valuestring,
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++;
}
}
cJSON_Delete(skill_events);
}
cJSON_Delete(adoption_events);
DEBUG_INFO("[didactyl] trigger manager loaded %d trigger(s) from skills", loaded);
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) {
snprintf(t->cron_expr, sizeof(t->cron_expr), "%s", filter_json);
}
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) {
snprintf(t->cron_expr, sizeof(t->cron_expr), "%s", filter_json);
} 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;
if (!cron_matches_now(expr, now)) {
continue;
}
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", 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);
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_fire_dm(trigger_manager_t* mgr,
const char* sender_pubkey_hex,
const char* message,
didactyl_sender_tier_t tier,
const char* source_label) {
if (!mgr || !sender_pubkey_hex || !message) {
return -1;
}
cJSON* event = cJSON_CreateObject();
if (!event) {
return -1;
}
cJSON_AddStringToObject(event, "type", "dm");
cJSON_AddStringToObject(event, "sender_pubkey", sender_pubkey_hex);
cJSON_AddStringToObject(event, "message", message);
cJSON_AddStringToObject(event, "tier",
tier == DIDACTYL_SENDER_ADMIN ? "admin" :
(tier == DIDACTYL_SENDER_WOT ? "wot" : "stranger"));
cJSON_AddNumberToObject(event, "created_at", (double)time(NULL));
int fired_total = 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_DM) {
continue;
}
if (!dm_filter_matches_tier(t->filter_json, tier)) {
continue;
}
int fired = maybe_fire_trigger_locked(mgr, i, event, source_label ? source_label : "dm");
if (fired > 0) {
fired_total += fired;
}
}
pthread_mutex_unlock(&mgr->mutex);
cJSON_Delete(event);
return fired_total;
}
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);
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