diff options
Diffstat (limited to 'src/mcp/tools/read_messages.c')
| -rw-r--r-- | src/mcp/tools/read_messages.c | 298 |
1 files changed, 273 insertions, 25 deletions
diff --git a/src/mcp/tools/read_messages.c b/src/mcp/tools/read_messages.c index 5a9e49f..1581534 100644 --- a/src/mcp/tools/read_messages.c +++ b/src/mcp/tools/read_messages.c @@ -3,25 +3,186 @@ #include "mcp/tools/common/common.h" #include "mcp/tools/common/context.h" +#include <stdint.h> #include <stdlib.h> #include <string.h> #include "base/allocator.h" +#include "storage/codec.h" + +#define READ_DEFAULT_LIMIT 100 +#define READ_MAX_LIMIT 1000 +#define READ_RESPONSE_BUDGET (8U * 1024U * 1024U - 64U * 1024U) +#define CURSOR_BYTES 47U +#define CURSOR_TEXT_MAX 64U + +typedef struct history_cursor +{ + int64_t timestamp; + char message_id[27]; +} history_cursor_t; + +static uint64_t +session_hash(const char *value) +{ + uint64_t hash = UINT64_C(1469598103934665603); + while (*value) + { + hash ^= (unsigned char)*value++; + hash *= UINT64_C(1099511628211); + } + return hash; +} + +static size_t +base64url_encode(const unsigned char *input, size_t length, char *output, size_t capacity) +{ + static const char alphabet[] = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_"; + size_t source = 0, target = 0; + while (source < length) + { + uint32_t value = (uint32_t)input[source++] << 16; + unsigned available = 1; + if (source < length) + { + value |= (uint32_t)input[source++] << 8; + available++; + } + if (source < length) + { + value |= input[source++]; + available++; + } + if (target + available + 1 >= capacity) + { + return 0; + } + output[target++] = alphabet[(value >> 18) & 63]; + output[target++] = alphabet[(value >> 12) & 63]; + if (available > 1) + { + output[target++] = alphabet[(value >> 6) & 63]; + } + if (available > 2) + { + output[target++] = alphabet[value & 63]; + } + } + output[target] = '\0'; + return target; +} static int -compare_history(const void *left, const void *right) +base64_value(unsigned char value) +{ + if (value >= 'A' && value <= 'Z') + { + return value - 'A'; + } + if (value >= 'a' && value <= 'z') + { + return value - 'a' + 26; + } + if (value >= '0' && value <= '9') + { + return value - '0' + 52; + } + if (value == '-') + { + return 62; + } + if (value == '_') + { + return 63; + } + return -1; +} + +static bool +base64url_decode(const char *input, unsigned char output[CURSOR_BYTES]) { - const tiny_store_entry_t *a = *(const tiny_store_entry_t *const *)left; - const tiny_store_entry_t *b = *(const tiny_store_entry_t *const *)right; - if (a->message.timestamp_ms < b->message.timestamp_ms) + size_t length = strlen(input), source = 0, target = 0; + uint32_t accumulator = 0; + unsigned bits = 0; + if (length != 63) + { + return false; + } + while (source < length) + { + int value = base64_value((unsigned char)input[source++]); + if (value < 0) + { + return false; + } + accumulator = (accumulator << 6) | (uint32_t)value; + bits += 6; + if (bits >= 8) + { + bits -= 8; + if (target >= CURSOR_BYTES) + { + return false; + } + output[target++] = (unsigned char)(accumulator >> bits); + accumulator &= bits ? (UINT32_C(1) << bits) - 1 : 0; + } + } + return target == CURSOR_BYTES && accumulator == 0; +} + +static bool +cursor_encode(const char *session, const tiny_message_t *message, char output[CURSOR_TEXT_MAX]) +{ + unsigned char data[CURSOR_BYTES] = { 0 }; + data[0] = 1; + tiny_codec_put_u64(data + 1, session_hash(session)); + tiny_codec_put_u64(data + 9, (uint64_t)message->timestamp_ms); + memcpy(data + 17, message->message_id, 26); + tiny_codec_put_u32(data + 43, tiny_codec_crc32(data, 43)); + return base64url_encode(data, sizeof(data), output, CURSOR_TEXT_MAX) == 63; +} + +static bool +cursor_decode(const char *session, const char *value, history_cursor_t *cursor) +{ + unsigned char data[CURSOR_BYTES]; + if (!value || !value[0] || !base64url_decode(value, data) || data[0] != 1 + || tiny_codec_get_u64(data + 1) != session_hash(session) + || tiny_codec_get_u32(data + 43) != tiny_codec_crc32(data, 43)) + { + return false; + } + memcpy(cursor->message_id, data + 17, 26); + cursor->message_id[26] = '\0'; + cursor->timestamp = (int64_t)tiny_codec_get_u64(data + 9); + return cursor->timestamp >= 0 && tiny_message_id_valid(cursor->message_id); +} + +static int +compare_cursor(const tiny_message_t *message, const history_cursor_t *cursor) +{ + if (message->timestamp_ms < cursor->timestamp) { return -1; } - if (a->message.timestamp_ms > b->message.timestamp_ms) + if (message->timestamp_ms > cursor->timestamp) { return 1; } - return strcmp(a->message.message_id, b->message.message_id); + return strcmp(message->message_id, cursor->message_id); +} + +static size_t +escaped_size(const char *value, size_t length) +{ + size_t position, total = 0; + for (position = 0; position < length; ++position) + { + unsigned char byte = (unsigned char)value[position]; + total += byte < 0x20 ? 6 : (byte == '"' || byte == '\\' ? 2 : 1); + } + return total; } static void @@ -29,52 +190,139 @@ call_read_messages(const mcpc_tool_t *tool, mcpc_ucbr_t *result) { tiny_mcp_tools_context_t *context = tiny_mcp_tools_context_current(); char *session = tiny_mcp_tool_get_string(tool, "session_id", TINY_SESSION_ID_MAX); - const tiny_store_entry_t **items = NULL; - size_t index, count = 0; + char *cursor_text = tiny_mcp_tool_get_string(tool, "cursor", CURSOR_TEXT_MAX); + int32_t requested_limit = READ_DEFAULT_LIMIT; + mcpc_errcode_t limit_status + = mcpc_tool_get_tpropval_i32(tool, (const char8_t *)"limit", &requested_limit); + const size_t *entries = NULL; + size_t entry_count = 0, position = 0, emitted = 0, outer_size = 256; + history_cursor_t cursor; + bool has_cursor = cursor_text != NULL, more = false; + const tiny_message_t *last = NULL; tiny_json_buffer_t json; - if (!context || !session || !session[0]) + + if (!context || !session || !session[0] + || (limit_status != MCPC_EC_0 && limit_status != MCPC_EC_ANYDATA_NOTFOUND) + || requested_limit < 1 || requested_limit > READ_MAX_LIMIT + || (has_cursor && !cursor_decode(session, cursor_text, &cursor))) { + tiny_mcp_tool_add_error(result, "VALIDATION_ERROR", "invalid read_messages arguments"); free(session); - tiny_mcp_tool_add_error(result, "VALIDATION_ERROR", "session_id is required"); + free(cursor_text); return; } - items = malloc((tiny_store_entry_count(context->store) + 1) * sizeof(*items)); - if (!items) + if (!tiny_mcp_tools_ensure_history_index(context) + || !tiny_history_index_session(context->history_index, session, &entries, &entry_count)) { + tiny_mcp_tool_add_error(result, "INTERNAL_ERROR", "history index allocation failed"); free(session); - tiny_mcp_tool_add_error(result, "INTERNAL_ERROR", "out of memory"); + free(cursor_text); return; } - for (index = 0; index < tiny_store_entry_count(context->store); ++index) + if (has_cursor) { - const tiny_store_entry_t *entry = tiny_store_entry(context->store, index); - if (!entry->deleted && strcmp(entry->message.session_id, session) == 0) + size_t low = 0, high = entry_count; + while (low < high) { - items[count++] = entry; + size_t middle = low + (high - low) / 2; + const tiny_store_entry_t *entry = tiny_store_entry(context->store, entries[middle]); + if (compare_cursor(&entry->message, &cursor) <= 0) + { + low = middle + 1; + } + else + { + high = middle; + } } + position = low; } - qsort(items, count, sizeof(*items), compare_history); + tiny_json_init(&json); tiny_json_raw(&json, "{\"history\":["); - for (index = 0; index < count; ++index) + while (position < entry_count && emitted < (size_t)requested_limit) { - if (index) + const tiny_store_entry_t *entry = tiny_store_entry(context->store, entries[position]); + tiny_json_buffer_t item; + size_t item_size; + if (!entry || entry->deleted) + { + position++; + continue; + } + tiny_json_init(&item); + tiny_mcp_tool_append_message(&item, &entry->message); + item_size = escaped_size(item.data, item.length) + (emitted ? 2 : 0); + if (item.failed || (emitted && outer_size + item_size > READ_RESPONSE_BUDGET)) + { + tiny_json_free(&item); + if (item.failed) + { + json.failed = true; + } + more = true; + break; + } + if (emitted) { tiny_json_raw(&json, ","); } - tiny_mcp_tool_append_message(&json, &items[index]->message); + tiny_json_raw_n(&json, item.data, item.length); + outer_size += item_size; + emitted++; + last = &entry->message; + position++; + tiny_json_free(&item); + } + if (!more) + { + size_t lookahead = position; + while (lookahead < entry_count) + { + const tiny_store_entry_t *entry = tiny_store_entry(context->store, entries[lookahead++]); + if (entry && !entry->deleted) + { + more = true; + break; + } + } + } + tiny_json_raw(&json, "]"); + if (more && last) + { + char next_cursor[CURSOR_TEXT_MAX]; + if (!cursor_encode(session, last, next_cursor)) + { + json.failed = true; + } + else + { + tiny_json_raw(&json, ",\"nextCursor\":"); + tiny_json_string(&json, next_cursor); + } + } + tiny_json_raw(&json, "}"); + if (json.failed) + { + tiny_mcp_tool_add_error(result, "INTERNAL_ERROR", "response allocation failed"); + } + else + { + tiny_mcp_tool_add_json(result, &json); } - tiny_json_raw(&json, "]}"); - tiny_mcp_tool_add_json(result, &json); tiny_json_free(&json); - free(items); free(session); + free(cursor_text); } bool tiny_mcp_register_read_messages(mcpc_server_t *server) { - mcpc_tool_t *tool = tiny_mcp_tool_new("read_messages", "Read a session", call_read_messages); + mcpc_tool_t *tool = tiny_mcp_tool_new("read_messages", "Read a session page", call_read_messages); return tool && tiny_mcp_tool_add_property(tool, "session_id", "Session ID", MCPC_U8STR, true) + && tiny_mcp_tool_add_property( + tool, "limit", "Optional page limit from 1 to 1000; defaults to 100", MCPC_I32, false) + && tiny_mcp_tool_add_property( + tool, "cursor", "Optional opaque page cursor", MCPC_U8STR, false) && mcpc_server_add_tool(server, tool) == MCPC_EC_0; } |