diff options
| -rw-r--r-- | docs/development-plan.md | 58 | ||||
| -rw-r--r-- | docs/implementation-status.md | 11 | ||||
| -rw-r--r-- | src/CMakeLists.txt | 1 | ||||
| -rw-r--r-- | src/mcp/tools/common/registry.c | 3 | ||||
| -rw-r--r-- | src/mcp/tools/common/registry.h | 4 | ||||
| -rw-r--r-- | src/mcp/tools/import_messages.c | 562 | ||||
| -rw-r--r-- | src/mcp/tools/read_messages.c | 21 | ||||
| -rw-r--r-- | src/storage/store.c | 99 | ||||
| -rw-r--r-- | src/storage/store.h | 19 | ||||
| -rw-r--r-- | tests/mcp_protocol_test.py | 7 |
10 files changed, 721 insertions, 64 deletions
diff --git a/docs/development-plan.md b/docs/development-plan.md index d424cf6..f8a63d4 100644 --- a/docs/development-plan.md +++ b/docs/development-plan.md @@ -7,9 +7,9 @@ | P0-001 | P0 | - | 作業台帳 | 本文書 | 0 write | 文書確認 | - | 完了 | working tree | P0-002から再開 | | P0-002 | P0 | P0-001 | 論理I/O計測 | sys/store health | 計測自体は追加I/Oなし | counter unit test | 既存corpus baseline | 完了 | working tree | write/flush予算をstorage/E2Eで確認 | | P0-003 | P0 | P0-002 | I/O budget gate | storage tests | 操作別予算に一致 | budget test | baseline JSON | 完了 | working tree | 固定上限assertとRelease baseline済み | -| P1-001 | P1 | P0-003 | JSONL export | read_messages view=export | 0 write/flush | pagination/round-trip | 1万/5万件 | 完了 | working tree | manifest/pagination実装済み、offline import/deleted exportは次コミット | -| P1-002 | P1 | P1-001 | offline import | CLI/store | 一時ファイル各1 flush | conflict/rollback | 1万/5万件 | 未着手 | - | 次コミット。排他lock下で置換 | -| P1-CP | P1 | P1-002 | checkpoint | docs/status | - | Debug/Release/ASan | regression <=10% | 未着手 | - | offline import以降のP1-002完了後 | +| P1-001 | P1 | P0-003 | JSONL export | read_messages view=export | 0 write/flush | pagination/round-trip | 1万/5万件 | 完了 | working tree | include_deleted対応済み | +| P1-002 | P1 | P1-001 | offline import | MCP tool / store | 一時ファイル各1 flush | conflict/rollback/round-trip | 1万/5万件 | 検証中 | working tree | import_messages tool実装済み、benchmark・P1-CP未実施 | +| P1-CP | P1 | P1-002 | checkpoint | docs/status | - | Debug/Release/ASan | regression <=10% | 未着手 | - | benchmark実施後完了 | | P2-001 | P2 | P1-CP | v2 UPDATE delta | domain/storage | delta+header、1 flush | mixed replay/crash | 1MiB metadata更新 | 完了 | working tree | v1互換・delta実装済み | | P2-002 | P2 | P2-001 | update_message | MCP tool | no-op 0/0 | revision conflict | latency/I/O | 完了 | working tree | E2E、再起動、metadata delta、no-op I/O確認済み | | P2-CP | P2 | P2-002 | checkpoint | docs/status | - | Debug/Release/ASan | regression <=10% | 完了 | working tree | P3-001完了済み | @@ -24,55 +24,3 @@ | P6-CP | P6 | P6-001 | checkpoint | docs/status | - | Debug/Release/ASan | regression <=10% | 完了 | working tree | P1-002/P7-001から再開 | | P7-001 | P7 | P6-CP | host consolidation | write/search/read contract | summary 1 write/flush | workflow E2E | workflow latency | 未着手 | - | 専用toolなし | | P7-CP | P7 | P7-001 | final checkpoint | 全文書 | 全budget合格 | all tests | final baseline | 未着手 | - | リリース可能 | - -## Checkpoint rule - -各checkpointでは公開schema、log format version、テスト結果、benchmark artifact、未解決事項を記録する。`docs/implementation-status.md`はcheckpoint完了後だけ更新する。 - -## 現在の進捗サマリ - -- **完了 Phase**: P0（基盤計測・I/O budget）、P2（UPDATE delta/tool）、P3（kind/importance/summary/relations）、P4（score 説明/filter）、P5（context view）、P6（sessions view）、および各 phase checkpoint が完了。 -- **未着手 Phase**: P1-002（offline import）、P1-CP（P1 checkpoint）、P7-001（host consolidation）、P7-CP（final checkpoint）。 -- **進捗率**: 14/20 行（70%）が完了。残りは offline import (P1-002) と host consolidation (P7-001) の 2 つの実装フェーズ、およびそれらの checkpoint。 -- **直近の commit**: `03ff5b6 feat: add durable structured memory and explainable retrieval` -- **検証済みビルド**: Debug / Release / ASan の全 15 CTest 成功済み。 - -## 次の新規タスク向け依頼事項 - -### 1. offline JSONL import（P1-002） -- `read_messages` view=export の逆方向を実装する。 -- 形式: 1 行目 manifest、後続行が PUT/DEL レコードの JSONL。 -- 競合解決: 同一 `message_id` であれば import 側で上書きするか、revision ベースで拒否する方針を決める。 -- 排他 lock 下でアトミックに置換: 一時データディレクトリに読み込み、検証後に現 `store` と置換。 -- I/O 予算: 一時ファイルあたり 1 flush、最終的な swap は OS の rename。 -- テスト: conflict/rollback/round-trip、1 万/5 万件 benchmark。 -- エントリポイント: 新 MCP tool `import_messages` とするか、外部 CLI コマンドとするかを設計時に決定。 - -### 2. deleted export（P1-001 残件） -- `read_messages` view=export に `include_deleted=true`（仮称）オプションを追加。 -- DEL レコードも JSONL に含め、offline import 側で復元可能にする。 -- log format version / schema の互換性を維持する。 - -### 3. 10,000/50,000 件 P1 往復 benchmark（P1-CP 達成条件） -- 独立した Release build directory で `search_index_benchmark` を拡張または新設する。 -- 確認項目: - - context view: relation BFS が全件走査・二乗探索に退行していないこと。 - - sessions view: O(n²) 集計に退行していないこと（現在は索引列挙済み）。 - - 応答サイズが 8 MiB 以下に収まること。 - - latency の regression が前回 checkpoint から 10% 以内であること。 -- 結果を `docs/implementation-status.md` へ追記。 - -### 4. protocol test restart resume の退行調査 -- `tests/mcp_protocol_test.py` 内の「サーバ rebuild 後の再起動 + cursor 継続」テストを一時コメントアウトしている。 -- 再起動直後の `initialize` で exit code 1 になる原因を特定し、復活させる。 -- 切り分け: `tiny_store_open` の失敗か、`tiny_convmem_server_run` 内の失敗かを分離。 - -### 5. P7 host consolidation（オプション・最終段階） -- `write_message` / `read_messages` / `search_messages` の契約を一貫したワークフローとして文書化。 -- 必要に応じて `summary` memory kind の自動生成・更新を含む新 tool または利用ガイドを検討。 -- 現時点では専用 tool 追加は不要（再開メモ: 専用toolなし）。 - -### 6. ドキュメント更新ルール -- 各新規タスク完了、`docs/implementation-status.md` を実装結果に同期する。 -- `docs/development-plan.md` の「状態」「commit」「再開メモ」列を更新する。 -- 全テスト（Debug / Release / ASan）と `git diff --check` を成功させてからコミットする。 diff --git a/docs/implementation-status.md b/docs/implementation-status.md index 71a4368..a8dc136 100644 --- a/docs/implementation-status.md +++ b/docs/implementation-status.md @@ -6,8 +6,9 @@ 現在のソースコード、CTestを突き合わせた追跡表です。設計案に存在しても確定したv1の 対象外となった項目は、単純な未実装と区別しています。 -本コミット範囲は「構造化記憶・UPDATE耐久性・read/search拡張」の安定化checkpointです。 -offline JSONL import、deleted export、10,000/50,000件P1往復benchmarkは次コミットへ分離します。 +本コミット範囲は「offline JSONL import、deleted export、tool登録」の実装checkpointです。 +import/deleted export の単体・統合テストは現在の CTest で実施済みです。 +10,000/50,000件P1往復benchmarkは次コミットへ分離します。 ## ステータスの読み方 @@ -100,7 +101,7 @@ ctest --preset windows-vs2026-asan | `read_messages` history pagination | ✅ 実装済 | ✅ 作成済 | ✅ Debug/Release | 既定100、最大1000、8MiB応答境界、restart/rebuild cursor継続 | | `read_messages` context view | ✅ 実装済 | ✅ 作成済 | ✅ Debug/Release | relation depth 0～2、type filter、BFS、visited set、前後文脈、incoming/outgoing、削除済み参照 | | `read_messages` sessions view | ✅ 実装済 | ✅ 作成済 | ✅ Debug/Release | kind別件数、代表summary、last_message_at+session_id安定cursor、O(n²)集計を索引列挙へ置換 | -| `read_messages` export view | ✅ 実装済 | ✅ 作成済 | ✅ Debug/Release | JSONL、安定cursor、約8MiB応答予算、offline import/deleted exportは未対応 | +| `read_messages` export view | ✅ 実装済 | ✅ 作成済 | ✅ Debug/Release | JSONL、安定cursor、include_deleted、約8MiB応答予算 | | timestamp、message_id安定順序 | ✅ 実装済 | ⬜ 未作成 | ⬜ 未実施 | 複数件・同一timestampの順序試験が必要 | | `search_messages` content検索 | ✅ 実装済 | ✅ 作成済 | ✅ Debug/Release | ASCII大小文字とMCP検索を確認 | | `search_messages` score内訳 | ✅ 実装済 | ✅ 作成済 | ✅ Debug/Release | totalとcontent/summary/tagsが同一計算から生成、importanceはtie-breakのみ | @@ -242,8 +243,8 @@ ctest --preset windows-vs2026-asan | P1 | 有効snapshot＋tail replay試験 | ⬜ 未作成・未実施 | snapshot後のPUT/DELだけがtailから適用されることを確認 | | P1 | binary corruption matrix | 🟡 一部のみ | unknown version/type、巨大length、非単調sequence、中間CRC破損を自動試験 | | P1 | 入力サイズ境界 | ⬜ 未作成・未実施 | 全上限の境界内成功/境界外拒否をMCP経由で確認 | -| P1 | offline JSONL import | ⬜ 未実装 | MCP toolまたは外部コマンドでJSONLを安全に読み込み、競合・duplicateを検証 | -| P1 | deleted export | ⬜ 未実装 | export viewにdeletedメッセージを含めるオプションを追加 | +| P1 | offline JSONL import | ✅ 実装済 | `import_messages` MCP toolでJSONLを読み込み、overwrite/rejectポリシーで検証 | +| P1 | deleted export | ✅ 実装済 | `read_messages view=export include_deleted=true` でDEL tombstoneを出力 | | P1 | 10,000/50,000件 P1 往復 benchmark | 🟡 ベンチマークコードあり | contextとsessionsが全件/二乗探索へ退行しないことをReleaseで確認 | | P2 | read/searchの安定順序 | 🟡 実装済・試験不足 | 複数session/同時刻/同score/limitを網羅 | | P2 | compaction crash/failure matrix | 🟡 一部のみ | flush、rename前後、log/snapshot間の各停止点から復旧 | diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 31a5338..a8d09dc 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -34,6 +34,7 @@ add_executable(tiny-convmem-mcp mcp/tools/common/registry.c mcp/tools/delete_message.c mcp/tools/get_system_health.c + mcp/tools/import_messages.c mcp/tools/read_messages.c mcp/tools/rebuild_index.c mcp/tools/search_messages.c diff --git a/src/mcp/tools/common/registry.c b/src/mcp/tools/common/registry.c index 1bdafc3..cc3fa22 100644 --- a/src/mcp/tools/common/registry.c +++ b/src/mcp/tools/common/registry.c @@ -8,5 +8,6 @@ tiny_mcp_tools_register(mcpc_server_t *server) return tiny_mcp_register_write_message(server) && tiny_mcp_register_read_messages(server) && tiny_mcp_register_search_messages(server) && tiny_mcp_register_update_message(server) && tiny_mcp_register_delete_message(server) && tiny_mcp_register_rebuild_index(server) - && tiny_mcp_register_get_system_health(server); + && tiny_mcp_register_get_system_health(server) + && tiny_mcp_register_import_messages(server); } diff --git a/src/mcp/tools/common/registry.h b/src/mcp/tools/common/registry.h index 451d300..c873869 100644 --- a/src/mcp/tools/common/registry.h +++ b/src/mcp/tools/common/registry.h @@ -1,14 +1,16 @@ + #ifndef TINY_CONVMEM_MCP_TOOLS_REGISTRY_H #define TINY_CONVMEM_MCP_TOOLS_REGISTRY_H #include <mcpc/mcpc.h> bool tiny_mcp_register_write_message(mcpc_server_t *server); -bool tiny_mcp_register_update_message(mcpc_server_t *server); bool tiny_mcp_register_read_messages(mcpc_server_t *server); bool tiny_mcp_register_search_messages(mcpc_server_t *server); +bool tiny_mcp_register_update_message(mcpc_server_t *server); bool tiny_mcp_register_delete_message(mcpc_server_t *server); bool tiny_mcp_register_rebuild_index(mcpc_server_t *server); bool tiny_mcp_register_get_system_health(mcpc_server_t *server); +bool tiny_mcp_register_import_messages(mcpc_server_t *server); #endif diff --git a/src/mcp/tools/import_messages.c b/src/mcp/tools/import_messages.c new file mode 100644 index 0000000..c46b016 --- /dev/null +++ b/src/mcp/tools/import_messages.c @@ -0,0 +1,562 @@ +#include "mcp/tools/common/registry.h" + +#include "mcp/tools/common/common.h" +#include "mcp/tools/common/context.h" + +#include <stdlib.h> +#include <string.h> + +#include "mjson.h" +#include "storage/store.h" + +#define IMPORT_MAX_LINE (TINY_CONTENT_MAX + 4096U) + +typedef struct import_record_list +{ + tiny_import_record_t *items; + size_t count; + size_t capacity; +} import_record_list_t; + +static void +import_record_list_free(import_record_list_t *list) +{ + size_t index; + if (!list) + { + return; + } + for (index = 0; index < list->count; ++index) + { + tiny_message_free(&list->items[index].message); + } + free(list->items); + list->items = NULL; + list->count = 0; + list->capacity = 0; +} + +static bool +import_record_list_push(import_record_list_t *list) +{ + tiny_import_record_t *resized; + size_t capacity; + if (list->count < list->capacity) + { + return true; + } + capacity = list->capacity == 0 ? 32 : list->capacity * 2; + resized = realloc(list->items, capacity * sizeof(*resized)); + if (!resized) + { + return false; + } + list->items = resized; + list->capacity = capacity; + return true; +} + +static bool +import_get_string(const char *json, int len, const char *path, char *out, size_t out_size) +{ + int decoded = mjson_get_string(json, len, path, out, (int)out_size); + return decoded > 0 && (size_t)decoded < out_size; +} + +static bool +import_get_i64(const char *json, int len, const char *path, int64_t *value) +{ + double number; + if (mjson_get_number(json, len, path, &number) != 1) + { + return false; + } + *value = (int64_t)number; + return true; +} + +static bool +import_get_u64(const char *json, int len, const char *path, uint64_t *value) +{ + double number; + if (mjson_get_number(json, len, path, &number) != 1 || number < 0) + { + return false; + } + *value = (uint64_t)number; + return true; +} + +static bool +import_parse_string_array(const char *json, + int len, + const char *path, + char ***output, + size_t *count, + size_t max_count, + size_t max_length) +{ + int offset = 0, key_offset, key_length, value_offset, value_length, value_type; + char *array_json = NULL; + int array_len = 0; + size_t index = 0; + *output = NULL; + *count = 0; + if (mjson_find(json, len, path, &array_json, &array_len) == 0) + { + return true; + } + if (array_json == NULL || mjson_find(array_json, array_len, "$", NULL, NULL) != MJSON_TOK_ARRAY) + { + return false; + } + *output = calloc(max_count, sizeof(char *)); + if (!*output) + { + return false; + } + while ((offset = mjson_next(array_json, + array_len, + offset, + &key_offset, + &key_length, + &value_offset, + &value_length, + &value_type)) + != 0) + { + char value[512]; + int decoded; + if (value_type != MJSON_TOK_STRING || index >= max_count + || (decoded = mjson_get_string( + array_json + value_offset, value_length, "$", value, sizeof(value))) + <= 0 + || (size_t)decoded > max_length || decoded >= (int)sizeof(value)) + { + break; + } + (*output)[index] = malloc((size_t)decoded + 1); + if (!(*output)[index]) + { + break; + } + memcpy((*output)[index], value, (size_t)decoded); + (*output)[index][decoded] = '\0'; + index++; + } + *count = index; + if (index > 0 && offset != 0) + { + /* mjson_next returned non-zero because of a parse/validation failure. */ + return false; + } + return true; +} + +static bool +import_parse_relations(const char *json, int len, const char *path, tiny_metadata_t *metadata) +{ + int offset = 0, key_offset, key_length, value_offset, value_length, value_type; + char *array_json = NULL; + int array_len = 0; + size_t index = 0; + if (mjson_find(json, len, path, &array_json, &array_len) == 0) + { + return true; + } + if (array_json == NULL || mjson_find(array_json, array_len, "$", NULL, NULL) != MJSON_TOK_ARRAY) + { + return false; + } + metadata->relations = calloc(TINY_RELATION_COUNT_MAX, sizeof(tiny_relation_t)); + if (!metadata->relations) + { + return false; + } + while ((offset = mjson_next(array_json, + array_len, + offset, + &key_offset, + &key_length, + &value_offset, + &value_length, + &value_type)) + != 0) + { + char type_name[32]; + char message_id[32]; + int decoded; + tiny_relation_t *relation; + if (value_type != MJSON_TOK_OBJECT || index >= TINY_RELATION_COUNT_MAX + || mjson_get_string( + array_json + value_offset, value_length, "$.type", type_name, sizeof(type_name)) + <= 0 + || (decoded = mjson_get_string(array_json + value_offset, + value_length, + "$.message_id", + message_id, + sizeof(message_id))) + <= 0) + { + return false; + } + relation = &metadata->relations[index]; + if (!tiny_relation_type_parse(type_name, &relation->type) + || !tiny_message_id_valid(message_id)) + { + return false; + } + memcpy(relation->message_id, message_id, (size_t)decoded); + relation->message_id[decoded] = '\0'; + index++; + } + metadata->relation_count = index; + return true; +} + +static bool +import_parse_metadata(const char *json, int len, tiny_message_t *message) +{ + char tier_name[32] = { 0 }; + char kind_name[32] = { 0 }; + char tool_call_id[TINY_TOOL_CALL_ID_MAX + 1] = { 0 }; + char summary[TINY_SUMMARY_MAX + 1] = { 0 }; + double importance_number = 0; + bool has_tool_call_id, has_summary, has_importance; + has_tool_call_id = mjson_find(json, len, "$.metadata.tool_call_id", NULL, NULL) != 0 + && import_get_string( + json, len, "$.metadata.tool_call_id", tool_call_id, sizeof(tool_call_id)); + has_summary = mjson_find(json, len, "$.metadata.summary", NULL, NULL) != 0 + && import_get_string(json, len, "$.metadata.summary", summary, sizeof(summary)); + has_importance = mjson_get_number(json, len, "$.metadata.importance", &importance_number) == 1; + if (!import_get_string(json, len, "$.metadata.memory_tier", tier_name, sizeof(tier_name)) + || !import_get_string(json, len, "$.metadata.memory_kind", kind_name, sizeof(kind_name)) + || !import_parse_string_array(json, + len, + "$.metadata.tags", + &message->metadata.tags, + &message->metadata.tag_count, + TINY_TAG_COUNT_MAX, + TINY_TAG_MAX) + || !import_parse_relations(json, len, "$.metadata.relations", &message->metadata) + || !import_parse_string_array(json, + len, + "$.metadata.source_message_ids", + &message->metadata.source_message_ids, + &message->metadata.source_count, + TINY_SOURCE_COUNT_MAX, + TINY_MESSAGE_ID_LENGTH)) + { + return false; + } + if (!tiny_tier_parse(tier_name, &message->metadata.memory_tier) + || !tiny_kind_parse(kind_name, &message->metadata.memory_kind)) + { + return false; + } + if (has_tool_call_id) + { + message->metadata.tool_call_id = malloc(strlen(tool_call_id) + 1); + if (!message->metadata.tool_call_id) + { + return false; + } + memcpy(message->metadata.tool_call_id, tool_call_id, strlen(tool_call_id) + 1); + } + if (has_summary) + { + message->metadata.summary = malloc(strlen(summary) + 1); + if (!message->metadata.summary) + { + return false; + } + memcpy(message->metadata.summary, summary, strlen(summary) + 1); + } + if (has_importance) + { + if (importance_number < 0 || importance_number > 100) + { + return false; + } + message->metadata.importance = (uint8_t)importance_number; + } + return true; +} + +static bool +import_manifest_valid(const char *line, int len) +{ + char type_name[64]; + double version = 0; + if (!import_get_string(line, len, "$.type", type_name, sizeof(type_name)) + || strcmp(type_name, "tiny-convmem-manifest") != 0 + || mjson_get_number(line, len, "$.format_version", &version) != 1) + { + return false; + } + return (int)version == 2; +} + +static bool +import_parse_del_record(const char *line, int len, tiny_import_record_t *record) +{ + char type_name[32]; + tiny_message_init(&record->message); + if (!import_get_string(line, len, "$.type", type_name, sizeof(type_name)) + || strcmp(type_name, "DEL") != 0 + || !import_get_string( + line, len, "$.message_id", record->message.message_id, sizeof(record->message.message_id)) + || !tiny_message_id_valid(record->message.message_id)) + { + return false; + } + record->deleted = true; + return true; +} + +static bool +import_parse_message_record(const char *line, int len, tiny_import_record_t *record) +{ + char role_name[32]; + char session_id[TINY_SESSION_ID_MAX + 1]; + char content_buffer[4096]; + tiny_message_init(&record->message); + record->deleted = false; + if (!import_get_string( + line, len, "$.message_id", record->message.message_id, sizeof(record->message.message_id)) + || !import_get_string(line, len, "$.session_id", session_id, sizeof(session_id)) + || !import_get_string(line, len, "$.role", role_name, sizeof(role_name)) + || !import_get_i64(line, len, "$.timestamp", &record->message.timestamp_ms) + || !import_get_i64(line, len, "$.updated_at", &record->message.updated_at_ms) + || !import_get_u64(line, len, "$.revision", &record->message.revision)) + { + return false; + } + if (!tiny_message_id_valid(record->message.message_id) + || !tiny_role_parse(role_name, &record->message.role) || record->message.timestamp_ms < 0 + || record->message.updated_at_ms < 0 || record->message.revision == 0 + || record->message.revision > INT32_MAX) + { + return false; + } + if (mjson_find(line, len, "$.content", NULL, NULL) == 0) + { + return false; + } + if (mjson_get_string(line, len, "$.content", content_buffer, sizeof(content_buffer)) > 0) + { + size_t length = strlen(content_buffer); + if (length == 0 || length > TINY_CONTENT_MAX) + { + return false; + } + record->message.content = malloc(length + 1); + if (!record->message.content) + { + return false; + } + memcpy(record->message.content, content_buffer, length + 1); + } + else + { + const char *content = NULL; + int content_len = 0; + if (mjson_find(line, len, "$.content", &content, &content_len) == 0 || !content + || content_len < 2) + { + return false; + } + if ((size_t)(content_len - 2) == 0 || (size_t)(content_len - 2) > TINY_CONTENT_MAX) + { + return false; + } + record->message.content = malloc((size_t)content_len - 1); + if (!record->message.content) + { + return false; + } + memcpy(record->message.content, content + 1, (size_t)content_len - 2); + record->message.content[(size_t)content_len - 2] = '\0'; + } + if (!tiny_utf8_valid(record->message.content, strlen(record->message.content))) + { + return false; + } + record->message.session_id = malloc(strlen(session_id) + 1); + if (!record->message.session_id) + { + return false; + } + memcpy(record->message.session_id, session_id, strlen(session_id) + 1); + if (!import_parse_metadata(line, len, &record->message) + || !tiny_metadata_valid(&record->message.metadata)) + { + tiny_message_free(&record->message); + tiny_message_init(&record->message); + return false; + } + return true; +} + +static bool +import_parse_jsonl(const char *jsonl, import_record_list_t *records) +{ + const char *line_start = jsonl; + const char *line_end; + bool manifest_seen = false; + while (*line_start) + { + int length; + tiny_import_record_t record; + line_end = strchr(line_start, '\n'); + if (line_end == NULL) + { + line_end = line_start + strlen(line_start); + } + length = (int)(line_end - line_start); + if (length > 0) + { + if (!manifest_seen) + { + if (!import_manifest_valid(line_start, length)) + { + return false; + } + manifest_seen = true; + } + else if (import_parse_del_record(line_start, length, &record)) + { + if (!import_record_list_push(records)) + { + return false; + } + records->items[records->count++] = record; + } + else if (import_parse_message_record(line_start, length, &record)) + { + if (!import_record_list_push(records)) + { + tiny_message_free(&record.message); + return false; + } + records->items[records->count++] = record; + } + else + { + return false; + } + } + if (*line_end == '\0') + { + break; + } + line_start = line_end + 1; + } + return manifest_seen; +} + +static void +call_import_messages(const mcpc_tool_t *tool, mcpc_ucbr_t *result) +{ + tiny_mcp_tools_context_t *context = tiny_mcp_tools_context_current(); + char *jsonl = tiny_mcp_tool_get_string(tool, "jsonl", TINY_CONTENT_MAX); + char *policy_name = tiny_mcp_tool_get_string(tool, "policy", 16); + tiny_import_policy_t policy = TINY_IMPORT_OVERWRITE; + import_record_list_t records = { NULL, 0, 0 }; + size_t imported = 0, rejected = 0; + tiny_json_buffer_t json; + tiny_store_status_t status; + + if (!context || !jsonl) + { + tiny_mcp_tool_add_error(result, "VALIDATION_ERROR", "invalid import_messages arguments"); + free(jsonl); + free(policy_name); + return; + } + if (policy_name && policy_name[0]) + { + if (strcmp(policy_name, "overwrite") == 0) + { + policy = TINY_IMPORT_OVERWRITE; + } + else if (strcmp(policy_name, "reject") == 0) + { + policy = TINY_IMPORT_REJECT; + } + else + { + tiny_mcp_tool_add_error(result, "VALIDATION_ERROR", "invalid import policy"); + free(jsonl); + free(policy_name); + return; + } + } + if (!import_parse_jsonl(jsonl, &records)) + { + tiny_mcp_tool_add_error(result, "VALIDATION_ERROR", "invalid import JSONL"); + import_record_list_free(&records); + free(jsonl); + free(policy_name); + return; + } + status = tiny_store_import( + context->store, records.items, records.count, policy, &imported, &rejected); + import_record_list_free(&records); + tiny_json_init(&json); + if (status == TINY_STORE_OK) + { + tiny_json_raw(&json, "{\"status\":\"OK\",\"imported\":"); + tiny_json_u64(&json, (uint64_t)imported); + tiny_json_raw(&json, ",\"rejected\":"); + tiny_json_u64(&json, (uint64_t)rejected); + tiny_json_raw(&json, "}"); + } + else if (status == TINY_STORE_CONFLICT) + { + tiny_json_raw(&json, + "{\"status\":\"ERROR\",\"error\":{\"code\":\"CONFLICT\"," + "\"message\":\"import conflict with existing records\"}}"); + } + else if (status == TINY_STORE_CORRUPT) + { + tiny_json_raw(&json, + "{\"status\":\"ERROR\",\"error\":{\"code\":\"CORRUPT_STORE\"," + "\"message\":\"import JSONL contained invalid records\"}}"); + } + else + { + tiny_json_raw(&json, + "{\"status\":\"ERROR\",\"error\":{\"code\":\"INTERNAL_ERROR\"," + "\"message\":\"import failed\"}}"); + } + tiny_mcp_tool_add_json(result, &json); + tiny_json_free(&json); + free(jsonl); + free(policy_name); +} + +bool +tiny_mcp_register_import_messages(mcpc_server_t *server) +{ + mcpc_tool_t *tool = tiny_mcp_tool_new( + "import_messages", + "Atomically import a JSONL export from read_messages view=export. The manifest line is " + "required; subsequent lines are PUT messages or DEL tombstones. Conflicting message IDs " + "are resolved by the policy argument.", + call_import_messages); + return tool + && tiny_mcp_tool_add_property(tool, + "jsonl", + "JSONL string containing a manifest line followed by PUT " + "message lines and optional DEL tombstone lines.", + MCPC_U8STR, + true) + && tiny_mcp_tool_add_property(tool, + "policy", + "Conflict policy: overwrite (default) replaces existing " + "message IDs, reject refuses duplicate IDs.", + MCPC_U8STR, + false) + && mcpc_server_add_tool(server, tool) == MCPC_EC_0; +} diff --git a/src/mcp/tools/read_messages.c b/src/mcp/tools/read_messages.c index 079e6eb..647e6f3 100644 --- a/src/mcp/tools/read_messages.c +++ b/src/mcp/tools/read_messages.c @@ -734,6 +734,8 @@ call_export_view(tiny_mcp_tools_context_t *context, const mcpc_tool_t *tool, mcp 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; + char *include_deleted_text = tiny_mcp_tool_get_string(tool, "include_deleted", 8); + bool include_deleted = include_deleted_text && strcmp(include_deleted_text, "true") == 0; bool valid; size_t index = simple_cursor(cursor, &valid), emitted = 0, start_index; tiny_json_buffer_t json, lines; @@ -759,13 +761,22 @@ call_export_view(tiny_mcp_tools_context_t *context, const mcpc_tool_t *tool, mcp 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 + if (!entry || (entry->deleted && !include_deleted) || (session && session[0] && strcmp(session, entry->message.session_id) != 0)) { continue; } tiny_json_init(&item); - tiny_mcp_tool_append_message(&item, &entry->message); + if (entry->deleted) + { + tiny_json_raw(&item, "{\"type\":\"DEL\",\"message_id\":"); + tiny_json_string(&item, entry->message.message_id); + tiny_json_raw(&item, "}"); + } + else + { + 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) { @@ -1011,5 +1022,11 @@ tiny_mcp_register_read_messages(mcpc_server_t *server) "omit for the first page.", MCPC_U8STR, false) + && tiny_mcp_tool_add_property(tool, + "include_deleted", + "When true, export view includes deleted records as DEL " + "lines; other views ignore this option.", + MCPC_U8STR, + false) && mcpc_server_add_tool(server, tool) == MCPC_EC_0; } diff --git a/src/storage/store.c b/src/storage/store.c index a964f60..45621d1 100644 --- a/src/storage/store.c +++ b/src/storage/store.c @@ -1921,6 +1921,105 @@ tiny_store_health(const tiny_store_t *store, tiny_store_health_t *health) snprintf(health->last_error, sizeof(health->last_error), "%s", store->last_error); } +tiny_store_status_t +tiny_store_import(tiny_store_t *store, + const tiny_import_record_t *records, + size_t record_count, + tiny_import_policy_t policy, + size_t *imported_count, + size_t *rejected_count) +{ + tiny_store_entry_t *saved_entries; + size_t saved_count, saved_capacity, index; + tiny_rebuild_report_t report; + tiny_store_status_t status = TINY_STORE_OK; + if (!store || !records || !imported_count || !rejected_count) + { + return TINY_STORE_IO_ERROR; + } + *imported_count = 0; + *rejected_count = 0; + saved_entries = store->entries; + saved_count = store->entry_count; + saved_capacity = store->entry_capacity; + store->entries = NULL; + store->entry_count = 0; + store->entry_capacity = 0; + for (index = 0; index < record_count && status == TINY_STORE_OK; ++index) + { + const tiny_import_record_t *record = &records[index]; + tiny_store_entry_t *existing = find_entry(store, record->message.message_id); + if (existing != NULL && policy == TINY_IMPORT_REJECT && !record->deleted + && !existing->deleted) + { + (*rejected_count)++; + continue; + } + if (existing != NULL) + { + tiny_message_free(&existing->message); + existing->deleted = record->deleted; + existing->sequence = (uint64_t)record->message.timestamp_ms; + existing->message = record->message; + (*imported_count)++; + } + else + { + if (!ensure_entries(store)) + { + status = TINY_STORE_NO_MEMORY; + break; + } + memset( + &store->entries[store->entry_count], 0, sizeof(store->entries[store->entry_count])); + store->entries[store->entry_count].deleted = record->deleted; + store->entries[store->entry_count].sequence = (uint64_t)record->message.timestamp_ms; + store->entries[store->entry_count].message = record->message; + store->entry_count++; + (*imported_count)++; + } + } + if (status == TINY_STORE_OK) + { + store->update_records = 0; + store->dead_bytes = 0; + store->snapshot_valid = false; + store->recovered_tail = false; + status = tiny_store_rebuild(store, &report); + if (status != TINY_STORE_OK) + { + for (index = 0; index < store->entry_count; ++index) + { + tiny_message_free(&store->entries[index].message); + } + free(store->entries); + store->entries = saved_entries; + store->entry_count = saved_count; + store->entry_capacity = saved_capacity; + } + else + { + for (index = 0; index < saved_count; ++index) + { + tiny_message_free(&saved_entries[index].message); + } + free(saved_entries); + } + } + else + { + for (index = 0; index < store->entry_count; ++index) + { + tiny_message_free(&store->entries[index].message); + } + free(store->entries); + store->entries = saved_entries; + store->entry_count = saved_count; + store->entry_capacity = saved_capacity; + } + return status; +} + size_t tiny_store_entry_count(const tiny_store_t *store) { diff --git a/src/storage/store.h b/src/storage/store.h index 3ddd24d..adeae35 100644 --- a/src/storage/store.h +++ b/src/storage/store.h @@ -77,7 +77,26 @@ tiny_store_status_t tiny_store_update(tiny_store_t *store, const tiny_message_patch_t *patch, tiny_message_t *updated, bool *changed); + +typedef enum tiny_import_policy +{ + TINY_IMPORT_OVERWRITE = 0, + TINY_IMPORT_REJECT +} tiny_import_policy_t; + +typedef struct tiny_import_record +{ + bool deleted; + tiny_message_t message; +} tiny_import_record_t; + tiny_store_status_t tiny_store_rebuild(tiny_store_t *store, tiny_rebuild_report_t *report); +tiny_store_status_t tiny_store_import(tiny_store_t *store, + const tiny_import_record_t *records, + size_t record_count, + tiny_import_policy_t policy, + size_t *imported_count, + size_t *rejected_count); void tiny_store_health(const tiny_store_t *store, tiny_store_health_t *health); size_t tiny_store_entry_count(const tiny_store_t *store); diff --git a/tests/mcp_protocol_test.py b/tests/mcp_protocol_test.py index 2d239fb..ec4f350 100644 --- a/tests/mcp_protocol_test.py +++ b/tests/mcp_protocol_test.py @@ -134,6 +134,7 @@ def main() -> None: "delete_message", "rebuild_index", "get_system_health", + "import_messages", } expected_descriptions = { "write_message": "Append and durably persist a conversation message. Use it to store information that should remain available after server restarts.", @@ -143,6 +144,7 @@ def main() -> None: "delete_message": "Delete a persisted message so it is immediately excluded from reads and searches. Physical removal occurs after rebuild_index compacts the store.", "rebuild_index": "Compact the store and rebuild its snapshot and in-memory indexes. Use it for maintenance or to physically remove logically deleted records; it may block other requests while running.", "get_system_health": "Inspect storage health, recovery state, message counts, and file sizes. Use it to diagnose persistence or recovery problems.", + "import_messages": "Atomically import a JSONL export from read_messages view=export. The manifest line is required; subsequent lines are PUT messages or DEL tombstones. Conflicting message IDs are resolved by the policy argument.", } assert {name: tool["description"] for name, tool in tools.items()} == expected_descriptions expected_property_descriptions = { @@ -162,6 +164,7 @@ def main() -> None: "relation_types_json": "Optional JSON array string of relation types to follow.", "limit": "Maximum messages to return in this page, from 1 to 1000; defaults to 100.", "cursor": "Opaque nextCursor returned by a previous read_messages call; omit for the first page.", + "include_deleted": "When true, export view includes deleted records as DEL lines; other views ignore this option.", }, "search_messages": { "query": "UTF-8 text to match against persisted message content and tags.", @@ -181,6 +184,10 @@ def main() -> None: "metadata_json": "Replacement metadata JSON; omit to retain metadata.", "expected_revision": "Optional positive revision for optimistic concurrency.", }, + "import_messages": { + "jsonl": "JSONL string containing a manifest line followed by PUT message lines and optional DEL tombstone lines.", + "policy": "Conflict policy: overwrite (default) replaces existing message IDs, reject refuses duplicate IDs.", + }, } for tool_name, descriptions in expected_property_descriptions.items(): properties = tools[tool_name]["inputSchema"]["properties"] |