From 52cc975a7dc48e23f76f262cf8ee1772fcbda89b Mon Sep 17 00:00:00 2001 From: Xiangao Lu Date: Tue, 14 Apr 2026 10:27:27 +0800 Subject: [PATCH] enhance: route agent responses as events --- .../fatfs_image/automation/automations.json | 21 ++- .../claw_modules/claw_core/CMakeLists.txt | 2 + .../claw_core/include/claw_core.h | 4 + .../claw_modules/claw_core/src/claw_core.c | 143 +++++++++++++++++- .../claw_event_router/src/claw_event_router.c | 44 +----- 5 files changed, 174 insertions(+), 40 deletions(-) diff --git a/application/basic_demo/fatfs_image/automation/automations.json b/application/basic_demo/fatfs_image/automation/automations.json index 7412f22..089ab44 100644 --- a/application/basic_demo/fatfs_image/automation/automations.json +++ b/application/basic_demo/fatfs_image/automation/automations.json @@ -131,17 +131,32 @@ "type": "run_agent", "input": { "target_channel": "{{event.source_channel}}", + "target_chat_id": "{{event.chat_id}}", "session_policy": "chat" } - }, + } + ] + }, + { + "id": "agent_response_send_message", + "description": "Deliver final agent responses to the original IM chat.", + "enabled": true, + "consume_on_match": true, + "ack": "{{event.source_channel}} agent response sent", + "match": { + "source_cap": "claw_core", + "event_type": "agent_response", + "content_type": "text" + }, + "actions": [ { "type": "send_message", "input": { "channel": "{{event.source_channel}}", "chat_id": "{{event.chat_id}}", - "message": "{{last.output}}" + "message": "{{event.text}}" } } ] } -] \ No newline at end of file +] diff --git a/components/claw_modules/claw_core/CMakeLists.txt b/components/claw_modules/claw_core/CMakeLists.txt index e8549c8..2a5d037 100644 --- a/components/claw_modules/claw_core/CMakeLists.txt +++ b/components/claw_modules/claw_core/CMakeLists.txt @@ -16,4 +16,6 @@ idf_component_register( esp-tls json mbedtls + PRIV_REQUIRES + claw_event_router ) diff --git a/components/claw_modules/claw_core/include/claw_core.h b/components/claw_modules/claw_core/include/claw_core.h index 830faf7..ba748dc 100644 --- a/components/claw_modules/claw_core/include/claw_core.h +++ b/components/claw_modules/claw_core/include/claw_core.h @@ -25,8 +25,12 @@ typedef enum { CLAW_CORE_COMPLETION_DONE = 0, } claw_core_completion_type_t; +#define CLAW_CORE_REQUEST_FLAG_PUBLISH_RESPONSE_EVENT (1U << 0) +#define CLAW_CORE_REQUEST_FLAG_SKIP_RESPONSE_QUEUE (1U << 1) + typedef struct { uint32_t request_id; + uint32_t flags; const char *session_id; const char *user_text; const char *source_channel; diff --git a/components/claw_modules/claw_core/src/claw_core.c b/components/claw_modules/claw_core/src/claw_core.c index c6340eb..bccc142 100644 --- a/components/claw_modules/claw_core/src/claw_core.c +++ b/components/claw_modules/claw_core/src/claw_core.c @@ -10,6 +10,7 @@ #include #include #include +#include #include "cJSON.h" #include "esp_log.h" @@ -17,6 +18,7 @@ #include "freertos/semphr.h" #include "freertos/task.h" +#include "claw_event_router.h" #include "claw_core_llm.h" static const char *TAG = "claw_core"; @@ -306,6 +308,141 @@ static void move_response_item(claw_core_response_t *dst, claw_core_response_ite memset(src, 0, sizeof(*src)); } +static int64_t claw_core_now_ms(void) +{ + struct timeval tv = {0}; + + gettimeofday(&tv, NULL); + return ((int64_t)tv.tv_sec * 1000LL) + (tv.tv_usec / 1000LL); +} + +static esp_err_t build_response_payload_json(const claw_core_request_t *request, + const claw_core_response_t *response, + char **out_payload_json) +{ + cJSON *root = NULL; + char *payload_json = NULL; + + if (!request || !response || !out_payload_json) { + return ESP_ERR_INVALID_ARG; + } + *out_payload_json = NULL; + + root = cJSON_CreateObject(); + if (!root) { + return ESP_ERR_NO_MEM; + } + + if (!cJSON_AddNumberToObject(root, "request_id", (double)request->request_id) || + !cJSON_AddStringToObject(root, + "status", + response->status == CLAW_CORE_RESPONSE_STATUS_OK ? "ok" : "error")) { + cJSON_Delete(root); + return ESP_ERR_NO_MEM; + } + + payload_json = cJSON_PrintUnformatted(root); + cJSON_Delete(root); + if (!payload_json) { + return ESP_ERR_NO_MEM; + } + + *out_payload_json = payload_json; + return ESP_OK; +} + +static esp_err_t build_agent_response_event(const claw_core_request_t *request, + const claw_core_response_t *response, + claw_event_t *out_event, + char **out_payload_json) +{ + const char *channel = NULL; + const char *chat_id = NULL; + int64_t now_ms; + esp_err_t err; + + if (!request || !response || !out_event || !out_payload_json || + response->status != CLAW_CORE_RESPONSE_STATUS_OK || + !response->text || !response->text[0]) { + return ESP_ERR_INVALID_ARG; + } + + memset(out_event, 0, sizeof(*out_event)); + *out_payload_json = NULL; + + err = build_response_payload_json(request, response, out_payload_json); + if (err != ESP_OK) { + return err; + } + + now_ms = claw_core_now_ms(); + channel = (response->target_channel && response->target_channel[0]) ? + response->target_channel : request->source_channel; + chat_id = (response->target_chat_id && response->target_chat_id[0]) ? + response->target_chat_id : request->source_chat_id; + + snprintf(out_event->event_id, sizeof(out_event->event_id), + "agent-%" PRIu32 "-%" PRId64, + request->request_id, + now_ms); + strlcpy(out_event->source_cap, "claw_core", sizeof(out_event->source_cap)); + strlcpy(out_event->event_type, "agent_response", sizeof(out_event->event_type)); + strlcpy(out_event->source_channel, channel ? channel : "", sizeof(out_event->source_channel)); + strlcpy(out_event->chat_id, chat_id ? chat_id : "", sizeof(out_event->chat_id)); + strlcpy(out_event->content_type, "text", sizeof(out_event->content_type)); + snprintf(out_event->message_id, sizeof(out_event->message_id), + "agent-%" PRIu32, + request->request_id); + if (request->source_message_id && request->source_message_id[0]) { + strlcpy(out_event->correlation_id, + request->source_message_id, + sizeof(out_event->correlation_id)); + } else { + snprintf(out_event->correlation_id, sizeof(out_event->correlation_id), + "%" PRIu32, + request->request_id); + } + out_event->timestamp_ms = now_ms; + out_event->session_policy = CLAW_EVENT_SESSION_POLICY_CHAT; + out_event->text = response->text; + out_event->payload_json = *out_payload_json; + + return ESP_OK; +} + +static void publish_response_event_if_requested(const claw_core_request_item_t *request, + const claw_core_response_item_t *response) +{ + claw_event_t event = {0}; + char *payload_json = NULL; + esp_err_t err; + + if (!request || !response || + !(request->view.flags & CLAW_CORE_REQUEST_FLAG_PUBLISH_RESPONSE_EVENT)) { + return; + } + if (response->view.status != CLAW_CORE_RESPONSE_STATUS_OK || + !response->view.text || !response->view.text[0]) { + return; + } + + err = build_agent_response_event(&request->view, + &response->view, + &event, + &payload_json); + if (err == ESP_OK) { + err = claw_event_router_publish(&event); + } + if (err != ESP_OK) { + ESP_LOGE(TAG, + "Failed to publish response event for request_id=%" PRIu32 ": %s", + request->view.request_id, + esp_err_to_name(err)); + } + + free(payload_json); +} + static esp_err_t append_user_message(cJSON *messages, const char *text) { cJSON *user_msg = NULL; @@ -859,7 +996,10 @@ static void claw_core_task(void *arg) } finish_request: - if (push_response(&response) != ESP_OK) { + publish_response_event_if_requested(&request, &response); + if (request.view.flags & CLAW_CORE_REQUEST_FLAG_SKIP_RESPONSE_QUEUE) { + free_response_item(&response); + } else if (push_response(&response) != ESP_OK) { ESP_LOGE(TAG, "Failed to enqueue response for request_id=%" PRIu32, request.view.request_id); free_response_item(&response); } @@ -1053,6 +1193,7 @@ esp_err_t claw_core_submit(const claw_core_request_t *request, uint32_t timeout_ } item.view.request_id = request->request_id; + item.view.flags = request->flags; item.owned_session_id = dup_string(request->session_id); item.owned_user_text = dup_string(request->user_text); item.owned_source_channel = dup_string(request->source_channel); diff --git a/components/claw_modules/claw_event_router/src/claw_event_router.c b/components/claw_modules/claw_event_router/src/claw_event_router.c index 9b4ae99..cb11923 100644 --- a/components/claw_modules/claw_event_router/src/claw_event_router.c +++ b/components/claw_modules/claw_event_router/src/claw_event_router.c @@ -1485,8 +1485,8 @@ static esp_err_t claw_event_router_execute_agent_action( const char *session_policy = NULL; claw_event_t agent_event = {0}; claw_core_request_t request = {0}; - claw_core_response_t response = {0}; char session_id[128] = {0}; + char submit_output[32] = {0}; esp_err_t err; input_root = cJSON_Parse(action->input_json); @@ -1514,6 +1514,8 @@ static esp_err_t claw_event_router_execute_agent_action( if (claw_event_router_build_session_id_with_config(&agent_event, session_id, sizeof(session_id)) > 0) { request.session_id = session_id; } + request.flags = CLAW_CORE_REQUEST_FLAG_PUBLISH_RESPONSE_EVENT | + CLAW_CORE_REQUEST_FLAG_SKIP_RESPONSE_QUEUE; request.user_text = (text && text[0]) ? text : (event->text ? event->text : ""); request.source_channel = event->source_channel; request.source_chat_id = event->chat_id; @@ -1524,20 +1526,14 @@ static esp_err_t claw_event_router_execute_agent_action( request.target_chat_id = (target_chat_id && target_chat_id[0]) ? target_chat_id : event->chat_id; err = claw_core_submit(&request, s_runtime.config.core_submit_timeout_ms); - if (err == ESP_OK) { - err = claw_core_receive_for(request.request_id, - &response, - s_runtime.config.core_receive_timeout_ms); - } if (err == ESP_OK) { + snprintf(submit_output, sizeof(submit_output), "request_id=%" PRIu32, request.request_id); claw_event_router_update_last_output(ctx, "agent", request.target_channel, - response.status == CLAW_CORE_RESPONSE_STATUS_OK ? "ok" : "error", - response.status == CLAW_CORE_RESPONSE_STATUS_OK ? - (response.text ? response.text : "") : - (response.error_message ? response.error_message : "")); + "queued", + submit_output); } else { claw_event_router_update_last_output(ctx, "agent", request.target_channel, "error", esp_err_to_name(err)); @@ -1545,9 +1541,9 @@ static esp_err_t claw_event_router_execute_agent_action( if (result) { result->action_count++; - if (err != ESP_OK || response.status != CLAW_CORE_RESPONSE_STATUS_OK) { + if (err != ESP_OK) { result->failed_actions++; - result->last_error = err != ESP_OK ? err : ESP_FAIL; + result->last_error = err; } } @@ -1556,9 +1552,6 @@ static esp_err_t claw_event_router_execute_agent_action( } cJSON_Delete(rendered_input); - if (err == ESP_OK) { - claw_core_response_free(&response); - } return err; } @@ -1826,27 +1819,6 @@ static esp_err_t claw_event_router_run_default_agent(const claw_event_t *event, event, ctx, result); - if (err == ESP_OK) { - claw_event_router_action_t send_action = { - .kind = CLAW_EVENT_ROUTER_ACTION_SEND_MESSAGE, - .input_json = strdup("{\"channel\":\"{{event.channel}}\",\"chat_id\":\"{{event.chat_id}}\",\"message\":\"{{last.output}}\"}"), - .caller = CLAW_CAP_CALLER_SYSTEM, - .capture_output = false, - }; - if (!send_action.input_json) { - err = ESP_ERR_NO_MEM; - } else { - err = claw_event_router_execute_send_message_action(&(claw_event_router_rule_t) { - .id = "__default_agent__", - }, - &send_action, - event, - ctx, - result); - } - free(send_action.input_json); - } - cJSON_Delete(ctx); free(action.input_json); return err;