diff options
Diffstat (limited to 'src/mcp/tools/read_messages.c')
| -rw-r--r-- | src/mcp/tools/read_messages.c | 543 |
1 files changed, 542 insertions, 1 deletions
diff --git a/src/mcp/tools/read_messages.c b/src/mcp/tools/read_messages.c index 44a026f..aa76ce8 100644 --- a/src/mcp/tools/read_messages.c +++ b/src/mcp/tools/read_messages.c @@ -4,11 +4,13 @@ #include "mcp/tools/common/context.h" #include <stdint.h> +#include <stdio.h> #include <stdlib.h> #include <string.h> #include "base/allocator.h" #include "storage/codec.h" +#include "mjson.h" #define READ_DEFAULT_LIMIT 100 #define READ_MAX_LIMIT 1000 @@ -185,10 +187,493 @@ escaped_size(const char *value, size_t length) return total; } +typedef struct context_edge +{ + size_t entry_index; + tiny_relation_type_t type; + size_t depth; + bool incoming; + int64_t timestamp; + char message_id[TINY_MESSAGE_ID_LENGTH + 1]; +} context_edge_t; + +static bool +relation_type_mask(const char *json, uint32_t *mask) +{ + int offset = 0, key_offset, key_length, value_offset, value_length, value_type; + if (!json) + { + *mask = UINT32_MAX; + return true; + } + if (mjson_find(json, (int)strlen(json), "$", NULL, NULL) != MJSON_TOK_ARRAY) + { + return false; + } + *mask = 0; + while ((offset = mjson_next(json, + (int)strlen(json), + offset, + &key_offset, + &key_length, + &value_offset, + &value_length, + &value_type)) + != 0) + { + char value[24]; + int decoded; + tiny_relation_type_t type; + if (value_type != MJSON_TOK_STRING + || (decoded = mjson_get_string( + json + value_offset, value_length, "$", value, sizeof(value))) + <= 0 + || decoded >= (int)sizeof(value)) + { + return false; + } + value[decoded] = '\0'; + if (!tiny_relation_type_parse(value, &type)) + { + return false; + } + *mask |= UINT32_C(1) << (unsigned)type; + } + return true; +} + +static int +compare_context_edges(const void *left, const void *right) +{ + const context_edge_t *a = left, *b = right; + if (a->depth != b->depth) + return a->depth < b->depth ? -1 : 1; + if (a->type != b->type) + return a->type < b->type ? -1 : 1; + if (a->timestamp != b->timestamp) + return a->timestamp < b->timestamp ? -1 : 1; + return strcmp(a->message_id, b->message_id); +} + +static void +call_context_view(tiny_mcp_tools_context_t *context, const mcpc_tool_t *tool, mcpc_ucbr_t *result) +{ + char *message_id = tiny_mcp_tool_get_string(tool, "message_id", TINY_MESSAGE_ID_LENGTH); + char *types_json = tiny_mcp_tool_get_string(tool, "relation_types_json", 256); + int32_t before = 1, after = 1, relation_depth = 1; + mcpc_errcode_t before_status + = mcpc_tool_get_tpropval_i32(tool, (const char8_t *)"before", &before); + mcpc_errcode_t after_status = mcpc_tool_get_tpropval_i32(tool, (const char8_t *)"after", &after); + mcpc_errcode_t depth_status + = mcpc_tool_get_tpropval_i32(tool, (const char8_t *)"relation_depth", &relation_depth); + const tiny_store_entry_t *target = NULL; + const size_t *entries = NULL; + const tiny_relation_reference_t *incoming = NULL; + size_t session_entry_count = 0, position, target_position = SIZE_MAX, emitted, message_count = 1; + size_t incoming_count = 0; + size_t target_index, queue[50], queue_depth[50], queue_head = 0, queue_count = 0; + context_edge_t edges[49]; + size_t edge_count = 0; + bool *visited = NULL, truncated = false; + uint32_t type_mask; + tiny_json_buffer_t json; + if (!message_id || before < 0 || before > 10 || after < 0 || after > 10 + || relation_depth < 0 || relation_depth > 2 + || (before_status != MCPC_EC_0 && before_status != MCPC_EC_ANYDATA_NOTFOUND) + || (after_status != MCPC_EC_0 && after_status != MCPC_EC_ANYDATA_NOTFOUND) + || (depth_status != MCPC_EC_0 && depth_status != MCPC_EC_ANYDATA_NOTFOUND) + || !relation_type_mask(types_json, &type_mask) || !context + || !tiny_mcp_tools_ensure_memory_index(context) + || !tiny_memory_index_find(context->memory_index, message_id, &target_index)) + { + tiny_mcp_tool_add_error(result, "VALIDATION_ERROR", "invalid context view arguments"); + free(message_id); + free(types_json); + return; + } + target = tiny_store_entry(context->store, target_index); + if (!target || target->deleted || !tiny_mcp_tools_ensure_history_index(context) + || !tiny_history_index_session( + context->history_index, target->message.session_id, &entries, &session_entry_count)) + { + tiny_mcp_tool_add_error(result, "NOT_FOUND", "message context was not found"); + free(message_id); + free(types_json); + return; + } + for (position = 0; position < session_entry_count; ++position) + { + if (strcmp(tiny_store_entry(context->store, entries[position])->message.message_id, + message_id) + == 0) + { + target_position = position; + break; + } + } + if (target_position == SIZE_MAX + || !(visited = calloc(tiny_store_entry_count(context->store), sizeof(bool)))) + { + tiny_mcp_tool_add_error(result, "INTERNAL_ERROR", "context index allocation failed"); + free(message_id); + free(types_json); + return; + } + visited[target_index] = true; + queue[queue_count] = target_index; + queue_depth[queue_count++] = 0; + while (queue_head < queue_count && edge_count < 49) + { + size_t current_index = queue[queue_head]; + size_t depth = queue_depth[queue_head++]; + const tiny_store_entry_t *current = tiny_store_entry(context->store, current_index); + size_t relation; + if (depth >= (size_t)relation_depth) + { + continue; + } + for (relation = 0; relation < current->message.metadata.relation_count; ++relation) + { + const tiny_relation_t *value = ¤t->message.metadata.relations[relation]; + size_t related_index; + const tiny_store_entry_t *related; + if (!(type_mask & (UINT32_C(1) << (unsigned)value->type)) + || !tiny_memory_index_find(context->memory_index, value->message_id, &related_index) + || visited[related_index]) + { + continue; + } + related = tiny_store_entry(context->store, related_index); + visited[related_index] = true; + edges[edge_count].entry_index = related_index; + edges[edge_count].type = value->type; + edges[edge_count].depth = depth + 1; + edges[edge_count].incoming = false; + edges[edge_count].timestamp = related->message.timestamp_ms; + memcpy(edges[edge_count].message_id, + related->message.message_id, + sizeof(edges[edge_count].message_id)); + edge_count++; + if (!related->deleted && queue_count < 50) + { + queue[queue_count] = related_index; + queue_depth[queue_count++] = depth + 1; + } + if (edge_count == 49) + { + truncated = true; + break; + } + } + if (edge_count == 49) + { + break; + } + if (!tiny_memory_index_incoming( + context->memory_index, + current->message.message_id, + &incoming, + &incoming_count)) + { + continue; + } + for (relation = 0; relation < incoming_count && edge_count < 49; ++relation) + { + const tiny_store_entry_t *related + = tiny_store_entry(context->store, incoming[relation].source_index); + if (!(type_mask & (UINT32_C(1) << (unsigned)incoming[relation].type)) + || visited[incoming[relation].source_index]) + { + continue; + } + visited[incoming[relation].source_index] = true; + edges[edge_count].entry_index = incoming[relation].source_index; + edges[edge_count].type = incoming[relation].type; + edges[edge_count].depth = depth + 1; + edges[edge_count].incoming = true; + edges[edge_count].timestamp = related->message.timestamp_ms; + memcpy(edges[edge_count].message_id, + related->message.message_id, + sizeof(edges[edge_count].message_id)); + edge_count++; + if (!related->deleted && queue_count < 50) + { + queue[queue_count] = incoming[relation].source_index; + queue_depth[queue_count++] = depth + 1; + } + } + } + qsort(edges, edge_count, sizeof(*edges), compare_context_edges); + tiny_json_init(&json); + tiny_json_raw(&json, "{\"message\":"); + tiny_mcp_tool_append_message(&json, &target->message); + tiny_json_raw(&json, ",\"chronological_context\":{\"before\":["); + emitted = 0; + position = target_position > (size_t)before ? target_position - (size_t)before : 0; + for (; position < target_position; ++position) + { + const tiny_store_entry_t *entry = tiny_store_entry(context->store, entries[position]); + if (entry && !entry->deleted) + { + tiny_json_buffer_t item; + tiny_json_init(&item); + tiny_mcp_tool_append_message(&item, &entry->message); + if (message_count >= 50 || item.failed + || json.length + item.length + 256 > READ_RESPONSE_BUDGET) + { + truncated = true; + tiny_json_free(&item); + break; + } + if (emitted++) + { + tiny_json_raw(&json, ","); + } + tiny_json_raw_n(&json, item.data, item.length); + tiny_json_free(&item); + message_count++; + } + } + tiny_json_raw(&json, "],\"after\":["); + emitted = 0; + for (position = target_position + 1; + position < session_entry_count && emitted < (size_t)after; + ++position) + { + const tiny_store_entry_t *entry = tiny_store_entry(context->store, entries[position]); + if (entry && !entry->deleted) + { + tiny_json_buffer_t item; + tiny_json_init(&item); + tiny_mcp_tool_append_message(&item, &entry->message); + if (message_count >= 50 || item.failed + || json.length + item.length + 256 > READ_RESPONSE_BUDGET) + { + truncated = true; + tiny_json_free(&item); + break; + } + if (emitted++) + { + tiny_json_raw(&json, ","); + } + tiny_json_raw_n(&json, item.data, item.length); + tiny_json_free(&item); + message_count++; + } + } + tiny_json_raw(&json, "]},\"relations\":["); + emitted = 0; + for (position = 0; position < edge_count; ++position) + { + const context_edge_t *edge = &edges[position]; + const tiny_store_entry_t *related = tiny_store_entry(context->store, edge->entry_index); + tiny_json_buffer_t item; + tiny_json_init(&item); + tiny_json_raw(&item, "{\"type\":"); + tiny_json_string(&item, tiny_relation_type_name(edge->type)); + tiny_json_raw(&item, edge->incoming ? ",\"direction\":\"incoming\"" : ",\"direction\":\"outgoing\""); + tiny_json_raw(&item, ",\"depth\":"); + tiny_json_u64(&item, edge->depth); + if (related && !related->deleted) + { + tiny_json_raw(&item, ",\"message\":"); + tiny_mcp_tool_append_message(&item, &related->message); + } + else + { + tiny_json_raw(&item, ",\"message_id\":"); + tiny_json_string(&item, edge->message_id); + tiny_json_raw(&item, ",\"status\":\"deleted\""); + } + tiny_json_raw(&item, "}"); + if (message_count >= 50 || item.failed + || json.length + item.length + 256 > READ_RESPONSE_BUDGET) + { + truncated = true; + tiny_json_free(&item); + break; + } + if (emitted++) + { + tiny_json_raw(&json, ","); + } + tiny_json_raw_n(&json, item.data, item.length); + tiny_json_free(&item); + if (related && !related->deleted) + { + message_count++; + } + } + tiny_json_raw(&json, "],\"session\":{\"session_id\":"); + tiny_json_string(&json, target->message.session_id); + tiny_json_raw(&json, ",\"message_count\":"); + tiny_json_u64(&json, session_entry_count); + tiny_json_raw(&json, "},\"truncated\":"); + tiny_json_raw(&json, truncated ? "true" : "false"); + tiny_json_raw(&json, "}"); + tiny_mcp_tool_add_json(result, &json); + tiny_json_free(&json); + free(visited); + free(message_id); + free(types_json); +} + +static size_t +simple_cursor(const char *cursor, bool *valid) +{ + char *end = NULL; + unsigned long long value; + if (!cursor) + { + *valid = true; + return 0; + } + value = strtoull(cursor, &end, 10); + *valid = cursor[0] && end && !*end && value <= SIZE_MAX; + return *valid ? (size_t)value : 0; +} + +static void +call_sessions_view(tiny_mcp_tools_context_t *context, const mcpc_tool_t *tool, mcpc_ucbr_t *result) +{ + char *cursor = tiny_mcp_tool_get_string(tool, "cursor", CURSOR_TEXT_MAX); + int32_t limit = READ_DEFAULT_LIMIT; + bool valid; + size_t start = simple_cursor(cursor, &valid), index, unique = 0, emitted = 0; + tiny_json_buffer_t json; + mcpc_errcode_t limit_status = mcpc_tool_get_tpropval_i32(tool, (const char8_t *)"limit", &limit); + if (!context || !valid || limit < 1 || limit > READ_MAX_LIMIT + || (limit_status != MCPC_EC_0 && limit_status != MCPC_EC_ANYDATA_NOTFOUND)) + { + tiny_mcp_tool_add_error(result, "VALIDATION_ERROR", "invalid sessions view arguments"); + free(cursor); + return; + } + tiny_json_init(&json); + tiny_json_raw(&json, "{\"sessions\":["); + for (index = 0; index < tiny_store_entry_count(context->store) && emitted < (size_t)limit; + ++index) + { + const tiny_store_entry_t *entry = tiny_store_entry(context->store, index); + size_t prior, scan, count = 0; + int64_t started = INT64_MAX, ended = 0; + if (!entry || entry->deleted) + { + continue; + } + for (prior = 0; prior < index; ++prior) + { + const tiny_store_entry_t *candidate = tiny_store_entry(context->store, prior); + if (candidate && !candidate->deleted + && strcmp(candidate->message.session_id, entry->message.session_id) == 0) + { + break; + } + } + if (prior != index || unique++ < start) + { + continue; + } + for (scan = index; scan < tiny_store_entry_count(context->store); ++scan) + { + const tiny_store_entry_t *candidate = tiny_store_entry(context->store, scan); + if (candidate && !candidate->deleted + && strcmp(candidate->message.session_id, entry->message.session_id) == 0) + { + count++; + started = candidate->message.timestamp_ms < started ? candidate->message.timestamp_ms + : started; + ended = candidate->message.timestamp_ms > ended ? candidate->message.timestamp_ms + : ended; + } + } + if (emitted++) + { + tiny_json_raw(&json, ","); + } + tiny_json_raw(&json, "{\"session_id\":"); + tiny_json_string(&json, entry->message.session_id); + tiny_json_raw(&json, ",\"message_count\":"); + tiny_json_u64(&json, count); + tiny_json_raw(&json, ",\"started_at\":"); + tiny_json_i64(&json, started); + tiny_json_raw(&json, ",\"last_message_at\":"); + tiny_json_i64(&json, ended); + tiny_json_raw(&json, "}"); + } + tiny_json_raw(&json, "]"); + if (emitted == (size_t)limit) + { + char next[32]; + snprintf(next, sizeof(next), "%zu", start + emitted); + tiny_json_raw(&json, ",\"nextCursor\":"); + tiny_json_string(&json, next); + } + tiny_json_raw(&json, "}"); + tiny_mcp_tool_add_json(result, &json); + tiny_json_free(&json); + free(cursor); +} + +static void +call_export_view(tiny_mcp_tools_context_t *context, const mcpc_tool_t *tool, mcpc_ucbr_t *result) +{ + char *session = tiny_mcp_tool_get_string(tool, "session_id", TINY_SESSION_ID_MAX); + char *cursor = tiny_mcp_tool_get_string(tool, "cursor", CURSOR_TEXT_MAX); + int32_t limit = 500; + bool valid; + size_t index = simple_cursor(cursor, &valid), emitted = 0; + tiny_json_buffer_t json, lines; + mcpc_errcode_t limit_status = mcpc_tool_get_tpropval_i32(tool, (const char8_t *)"limit", &limit); + if (!context || !valid || limit < 1 || limit > 1000 + || (limit_status != MCPC_EC_0 && limit_status != MCPC_EC_ANYDATA_NOTFOUND)) + { + tiny_mcp_tool_add_error(result, "VALIDATION_ERROR", "invalid export view arguments"); + free(session); + free(cursor); + return; + } + tiny_json_init(&lines); + if (index == 0) + { + tiny_json_raw(&lines, "{\"type\":\"tiny-convmem-manifest\",\"format_version\":2}\n"); + } + for (; index < tiny_store_entry_count(context->store) && emitted < (size_t)limit; ++index) + { + const tiny_store_entry_t *entry = tiny_store_entry(context->store, index); + if (!entry || entry->deleted + || (session && session[0] && strcmp(session, entry->message.session_id) != 0)) + { + continue; + } + tiny_mcp_tool_append_message(&lines, &entry->message); + tiny_json_raw(&lines, "\n"); + emitted++; + } + tiny_json_init(&json); + tiny_json_raw(&json, "{\"jsonl\":"); + tiny_json_string(&json, lines.data ? lines.data : ""); + if (index < tiny_store_entry_count(context->store)) + { + char next[32]; + snprintf(next, sizeof(next), "%zu", index); + tiny_json_raw(&json, ",\"nextCursor\":"); + tiny_json_string(&json, next); + } + tiny_json_raw(&json, "}"); + tiny_mcp_tool_add_json(result, &json); + tiny_json_free(&json); + tiny_json_free(&lines); + free(session); + free(cursor); +} + static void call_read_messages(const mcpc_tool_t *tool, mcpc_ucbr_t *result) { tiny_mcp_tools_context_t *context = tiny_mcp_tools_context_current(); + char *view = tiny_mcp_tool_get_string(tool, "view", 16); char *session = tiny_mcp_tool_get_string(tool, "session_id", TINY_SESSION_ID_MAX); char *cursor_text = tiny_mcp_tool_get_string(tool, "cursor", CURSOR_TEXT_MAX); int32_t requested_limit = READ_DEFAULT_LIMIT; @@ -201,6 +686,40 @@ call_read_messages(const mcpc_tool_t *tool, mcpc_ucbr_t *result) const tiny_message_t *last = NULL; tiny_json_buffer_t json; + if (view && strcmp(view, "context") == 0) + { + call_context_view(context, tool, result); + free(view); + free(session); + free(cursor_text); + return; + } + if (view && strcmp(view, "sessions") == 0) + { + call_sessions_view(context, tool, result); + free(view); + free(session); + free(cursor_text); + return; + } + if (view && strcmp(view, "export") == 0) + { + call_export_view(context, tool, result); + free(view); + free(session); + free(cursor_text); + return; + } + if (view && strcmp(view, "history") != 0) + { + tiny_mcp_tool_add_error(result, "VALIDATION_ERROR", "unsupported read_messages view"); + free(view); + free(session); + free(cursor_text); + return; + } + free(view); + if (!context || !session || !session[0] || (limit_status != MCPC_EC_0 && limit_status != MCPC_EC_ANYDATA_NOTFOUND) || requested_limit < 1 || requested_limit > READ_MAX_LIMIT @@ -324,11 +843,33 @@ tiny_mcp_register_read_messages(mcpc_server_t *server) "Returns a paginated history for reconstructing that conversation.", call_read_messages); return tool + && tiny_mcp_tool_add_property( + tool, + "view", + "Read view: history (default), context, sessions, or export.", + MCPC_U8STR, + false) && tiny_mcp_tool_add_property(tool, "session_id", "Conversation session whose chronological history to read.", MCPC_U8STR, - true) + false) + && tiny_mcp_tool_add_property( + tool, "message_id", "Message ULID required by the context view.", MCPC_U8STR, false) + && tiny_mcp_tool_add_property( + tool, "before", "Context messages before the target, from 0 to 10.", MCPC_I32, false) + && tiny_mcp_tool_add_property( + tool, "after", "Context messages after the target, from 0 to 10.", MCPC_I32, false) + && tiny_mcp_tool_add_property(tool, + "relation_depth", + "Relation traversal depth from 0 to 2; defaults to 1.", + MCPC_I32, + false) + && tiny_mcp_tool_add_property(tool, + "relation_types_json", + "Optional JSON array string of relation types to follow.", + MCPC_U8STR, + false) && tiny_mcp_tool_add_property( tool, "limit", |