diff options
Diffstat (limited to 'src/mcp/tools/read_messages.c')
| -rw-r--r-- | src/mcp/tools/read_messages.c | 232 |
1 files changed, 180 insertions, 52 deletions
diff --git a/src/mcp/tools/read_messages.c b/src/mcp/tools/read_messages.c index aa76ce8..079e6eb 100644 --- a/src/mcp/tools/read_messages.c +++ b/src/mcp/tools/read_messages.c @@ -225,8 +225,8 @@ relation_type_mask(const char *json, uint32_t *mask) 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))) + || (decoded + = mjson_get_string(json + value_offset, value_length, "$", value, sizeof(value))) <= 0 || decoded >= (int)sizeof(value)) { @@ -247,11 +247,17 @@ 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; + { + return a->depth < b->depth ? -1 : 1; + } if (a->type != b->type) - return a->type < b->type ? -1 : 1; + { + return a->type < b->type ? -1 : 1; + } if (a->timestamp != b->timestamp) - return a->timestamp < b->timestamp ? -1 : 1; + { + return a->timestamp < b->timestamp ? -1 : 1; + } return strcmp(a->message_id, b->message_id); } @@ -277,8 +283,8 @@ call_context_view(tiny_mcp_tools_context_t *context, const mcpc_tool_t *tool, mc 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 + 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) @@ -370,10 +376,7 @@ call_context_view(tiny_mcp_tools_context_t *context, const mcpc_tool_t *tool, mc break; } if (!tiny_memory_index_incoming( - context->memory_index, - current->message.message_id, - &incoming, - &incoming_count)) + context->memory_index, current->message.message_id, &incoming, &incoming_count)) { continue; } @@ -436,8 +439,7 @@ call_context_view(tiny_mcp_tools_context_t *context, const mcpc_tool_t *tool, mc } tiny_json_raw(&json, "],\"after\":["); emitted = 0; - for (position = target_position + 1; - position < session_entry_count && emitted < (size_t)after; + 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]); @@ -472,7 +474,8 @@ call_context_view(tiny_mcp_tools_context_t *context, const mcpc_tool_t *tool, mc 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, + edge->incoming ? ",\"direction\":\"incoming\"" : ",\"direction\":\"outgoing\""); tiny_json_raw(&item, ",\"depth\":"); tiny_json_u64(&item, edge->depth); if (related && !related->deleted) @@ -534,68 +537,173 @@ simple_cursor(const char *cursor, bool *valid) return *valid ? (size_t)value : 0; } +static bool +session_cursor_encode(int64_t last_message_at, const char *session_id, char output[CURSOR_TEXT_MAX]) +{ + unsigned char data[CURSOR_BYTES] = { 0 }; + data[0] = 2; + tiny_codec_put_u64(data + 1, (uint64_t)last_message_at); + tiny_codec_put_u32(data + 9, + tiny_codec_crc32((const unsigned char *)session_id, strlen(session_id))); + memcpy(data + 13, session_id, strlen(session_id)); + tiny_codec_put_u32(data + 43, tiny_codec_crc32(data, 43)); + return base64url_encode(data, sizeof(data), output, CURSOR_TEXT_MAX) == 63; +} + +static bool +session_cursor_decode(const char *cursor, + int64_t *last_message_at, + char session_id[TINY_SESSION_ID_MAX + 1]) +{ + unsigned char data[CURSOR_BYTES]; + size_t length; + if (!cursor || !cursor[0] || !base64url_decode(cursor, data) || data[0] != 2) + { + return false; + } + *last_message_at = (int64_t)tiny_codec_get_u64(data + 1); + length = tiny_codec_get_u32(data + 9); + if (length == 0 || length > TINY_SESSION_ID_MAX + || tiny_codec_crc32(data, 43) != tiny_codec_get_u32(data + 43)) + { + return false; + } + memcpy(session_id, data + 13, length); + session_id[length] = '\0'; + return tiny_utf8_valid(session_id, length); +} + +static int +compare_session_by_last_then_id(const void *left, const void *right) +{ + const tiny_history_index_session_info_t *a = left; + const tiny_history_index_session_info_t *b = right; + if (a->last_message_at != b->last_message_at) + { + return a->last_message_at < b->last_message_at ? -1 : 1; + } + return strcmp(a->session_id, b->session_id); +} + 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)) + int64_t cursor_last = 0; + char cursor_session[TINY_SESSION_ID_MAX + 1] = { 0 }; + bool has_cursor = cursor && cursor[0]; + size_t emitted = 0, index, position; + const tiny_history_index_t *history; + tiny_history_index_session_info_t *sessions = NULL; + size_t session_count = 0; + bool cursor_ok = !has_cursor || session_cursor_decode(cursor, &cursor_last, cursor_session); + + if (!context || !cursor_ok || limit < 1 || limit > READ_MAX_LIMIT + || (limit_status != MCPC_EC_0 && limit_status != MCPC_EC_ANYDATA_NOTFOUND) + || !tiny_mcp_tools_ensure_history_index(context) || !(history = context->history_index)) { 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) + + session_count = tiny_history_index_session_count(history); + if (session_count > 0) { - 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) + sessions = malloc(session_count * sizeof(*sessions)); + if (!sessions) { - continue; + tiny_mcp_tool_add_error(result, "INTERNAL_ERROR", "sessions allocation failed"); + free(cursor); + return; } - for (prior = 0; prior < index; ++prior) + for (index = 0; index < session_count; ++index) { - 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) + const char *session_id; + const size_t *entries; + size_t count; + if (!tiny_history_index_session_at(history, index, &session_id, &entries, &count)) { - break; + free(sessions); + tiny_mcp_tool_add_error(result, "INTERNAL_ERROR", "session enumeration failed"); + free(cursor); + return; } + sessions[index].session_id = session_id; + sessions[index].last_message_at + = count > 0 + ? tiny_store_entry(context->store, entries[count - 1])->message.timestamp_ms + : 0; + } + qsort(sessions, session_count, sizeof(*sessions), compare_session_by_last_then_id); + } + + tiny_json_init(&json); + tiny_json_raw(&json, "{\"sessions\":["); + for (position = 0; position < session_count && emitted < (size_t)limit; ++position) + { + const char *session_id = sessions[position].session_id; + const size_t *entries; + size_t count, representative_index = SIZE_MAX; + int64_t started = INT64_MAX, ended = 0; + int kind_counts[6] = { 0 }; + const tiny_store_entry_t *representative = NULL; + const char *summary = NULL; + if (has_cursor + && (sessions[position].last_message_at < cursor_last + || (sessions[position].last_message_at == cursor_last + && strcmp(session_id, cursor_session) <= 0))) + { + continue; } - if (prior != index || unique++ < start) + if (!tiny_history_index_session(history, session_id, &entries, &count)) { continue; } - for (scan = index; scan < tiny_store_entry_count(context->store); ++scan) + for (index = 0; index < count; ++index) { - 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) + const tiny_store_entry_t *entry = tiny_store_entry(context->store, entries[index]); + if (!entry || entry->deleted) { - count++; - started = candidate->message.timestamp_ms < started ? candidate->message.timestamp_ms - : started; - ended = candidate->message.timestamp_ms > ended ? candidate->message.timestamp_ms - : ended; + continue; } + kind_counts[entry->message.metadata.memory_kind - 1]++; + started = entry->message.timestamp_ms < started ? entry->message.timestamp_ms : started; + ended = entry->message.timestamp_ms > ended ? entry->message.timestamp_ms : ended; + representative_index = entries[index]; + representative = entry; + } + if (representative_index == SIZE_MAX) + { + continue; } if (emitted++) { tiny_json_raw(&json, ","); } + summary = representative->message.metadata.summary ? representative->message.metadata.summary + : representative->message.content; tiny_json_raw(&json, "{\"session_id\":"); - tiny_json_string(&json, entry->message.session_id); + tiny_json_string(&json, session_id); tiny_json_raw(&json, ",\"message_count\":"); tiny_json_u64(&json, count); + tiny_json_raw(&json, ",\"kind_counts\":{\"conversation\":"); + tiny_json_i64(&json, kind_counts[0]); + tiny_json_raw(&json, ",\"episode\":"); + tiny_json_i64(&json, kind_counts[1]); + tiny_json_raw(&json, ",\"semantic\":"); + tiny_json_i64(&json, kind_counts[2]); + tiny_json_raw(&json, ",\"procedure\":"); + tiny_json_i64(&json, kind_counts[3]); + tiny_json_raw(&json, ",\"preference\":"); + tiny_json_i64(&json, kind_counts[4]); + tiny_json_raw(&json, ",\"summary\":"); + tiny_json_i64(&json, kind_counts[5]); + tiny_json_raw(&json, "},\"representative_summary\":"); + tiny_json_string(&json, summary); tiny_json_raw(&json, ",\"started_at\":"); tiny_json_i64(&json, started); tiny_json_raw(&json, ",\"last_message_at\":"); @@ -603,16 +711,20 @@ call_sessions_view(tiny_mcp_tools_context_t *context, const mcpc_tool_t *tool, m tiny_json_raw(&json, "}"); } tiny_json_raw(&json, "]"); - if (emitted == (size_t)limit) + if (emitted == (size_t)limit && position < session_count) { - char next[32]; - snprintf(next, sizeof(next), "%zu", start + emitted); - tiny_json_raw(&json, ",\"nextCursor\":"); - tiny_json_string(&json, next); + char next[CURSOR_TEXT_MAX]; + if (session_cursor_encode( + sessions[position - 1].last_message_at, sessions[position - 1].session_id, next)) + { + 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(sessions); free(cursor); } @@ -623,8 +735,10 @@ call_export_view(tiny_mcp_tools_context_t *context, const mcpc_tool_t *tool, mcp 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; + size_t index = simple_cursor(cursor, &valid), emitted = 0, start_index; tiny_json_buffer_t json, lines; + size_t outer_size = 64; + bool truncated = false; 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)) @@ -634,6 +748,7 @@ call_export_view(tiny_mcp_tools_context_t *context, const mcpc_tool_t *tool, mcp free(cursor); return; } + start_index = index; tiny_json_init(&lines); if (index == 0) { @@ -642,24 +757,37 @@ call_export_view(tiny_mcp_tools_context_t *context, const mcpc_tool_t *tool, mcp 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); + tiny_json_buffer_t item; + size_t line_overhead; 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_init(&item); + tiny_mcp_tool_append_message(&item, &entry->message); + line_overhead = item.length + 1; + if (item.failed || outer_size + lines.length + line_overhead + 2 > READ_RESPONSE_BUDGET) + { + truncated = true; + tiny_json_free(&item); + break; + } + tiny_json_raw_n(&lines, item.data, item.length); tiny_json_raw(&lines, "\n"); emitted++; + tiny_json_free(&item); } 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)) + if (truncated || index < tiny_store_entry_count(context->store)) { char next[32]; - snprintf(next, sizeof(next), "%zu", index); + snprintf(next, sizeof(next), "%zu", index > start_index ? index : start_index + 1); tiny_json_raw(&json, ",\"nextCursor\":"); tiny_json_string(&json, next); + tiny_json_raw(&json, ",\"truncated\":true"); } tiny_json_raw(&json, "}"); tiny_mcp_tool_add_json(result, &json); |