[FEAT] agentic-core conversation and responses database CRUD.

[maralbahari]

Summary

This will be part of agentic-core after #29 merged will open for review

Integrated complete SQLx-based storage layer for persistent conversation and response management. Implements trait-based CRUD operations with connection pooling supporting SQLite, PostgreSQL, and MySQL. Two main stores (ConversationStore and ResponseStore) handle persisting input/output items with metadata and rehydrating full conversation/response history from database.

Features:

• Database connection pooling with optional URL configuration (defaults to SQLite)
• ConversationStore: persist turns with items and responses, rehydrate full conversation history
• ResponseStore: persist responses with items and metadata, rehydrate by response ID
• Schema migrations and initialization via SchemaManager
• Type-safe database models with sqlx FromRow derive

Test Plan

Unit tests (39 passing):

• Model CRUD operations, schema initialization, type conversions, error handling
• IO conversion from data types resulted from database models to Rust data types.

Integration tests (18 passing):

• Store-level CRUD, error handling for disabled/missing resources
• Multi-turn conversations with response chaining
• Concurrent operations, foreign key constraints, edge cases

Benchmarks:

• conversation_persist: measure insertion of items and response records
• response_persist: measure response creation with item tracking
• conversation_rehydrate: measure full conversation retrieval from database
• response_rehydrate: measure response history lookup by ID

All tests pass. No clippy warnings.

Running Benchmarks

To run the storage CRUD benchmarks:

cargo bench -p agentic-core --bench storage_crud 

Or run with specific sample count and target time:

cargo bench -p agentic-core --bench storage_crud -- --sample-size 100 --warm-up-time 3.000 s

Benchmark Results

Benchmarking conversation_persist
conversation_persist    time:   [1.4119 ms 1.4390 ms 1.4663 ms]

Benchmarking response_persist
response_persist        time:   [893.67 µs 908.53 µs 923.55 µs]

Benchmarking conversation_rehydrate
conversation_rehydrate  time:   [1.4377 ms 1.4701 ms 1.5037 ms]

Benchmarking response_rehydrate
response_rehydrate      time:   [1.0061 ms 1.0270 ms 1.0488 ms]

Performance Notes

Response persist is ~1.5x faster than conversation persist because:

• ResponseStore.persist() stores history_item_ids directly in the responses table (single lookup)
• ConversationStore.persist() must query the items table to determine the next sequence number for conversation items (additional query)

[franciscojavierarceo]

This is great!

Can we handle a few items to before merging?

1. Add tenant_id TEXT placeholder columns to all three tables with indexes on (tenant_id, id) — avoids a schema migration when multi-tenancy lands.

2. Add raw_tokens TEXT placeholder column on items and/or responses — so we can store the unmodified token sequence and pass it back to vLLM for faster inference (prefix caching / KV-cache reuse) without a migration later.

3. Add a fallback variant for unknown item types — InputItem and OutputItem only handle message and function_call/function_call_output. For reference, ogx's ConversationItem handles ~10 types (web search, file search, MCP approval, MCP call, reasoning, compaction, etc.). Unknown types silently disappear via filter_map in as_inout(). At minimum, add an Unknown(serde_json::Value) variant so data is preserved round-trip.

4. Add metadata to the Conversation struct — the migration SQL has a metadata TEXT column but the FromRow struct in models/conversation.rs only has id and created_at. The field is silently dropped on read.

Note on crate structure

PR #35 (move gateway to agentic-server) was reverted in PR #40, so this PR's crates/agentic-core/ paths will need to be rebased accordingly.

[franciscojavierarceo]

created_at should be INTEGER NOT NULL (Unix epoch), not TEXT. The OpenAI API spec returns created_at as an int. TEXT means range queries do lexicographic comparison (slower), and converting back to the API response requires parsing. Applies to all three tables.

[maralbahari]

@franciscojavierarceo reverted the type to integer

[franciscojavierarceo]

Use MAX(seq) instead of COUNT(*) for the next sequence number. If items are ever deleted, COUNT(*) returns a lower number than the actual max seq, causing collisions. Use SELECT MAX(seq) FROM items WHERE conversation_id = ? then max_seq.unwrap_or(-1) + 1.

[maralbahari]

@franciscojavierarceo done

[franciscojavierarceo]

SCHEMA_READY should be per-pool, not a process-global static. Two pools pointing at different databases means the second skips migrations. The ensure_ready_for_test workaround confirms this is a known pain point. Suggestion: bundle the pool and readiness flag into a wrapper struct (e.g., PoolWithSchema) — eliminates the need for ensure_ready_for_test entirely.

[maralbahari]

@franciscojavierarceo changed to per instance pool creation over static global. my initial though was that it might be expensive and cause pref regression to have the pool created per instance. but as long as we make sure it is called correctly one it should not be a concern.

[franciscojavierarceo]

new_test() is identical to disabled() — drop one. Both do Self { pool: None }.

[franciscojavierarceo]

unwrap_or_default() silently swallows serialization errors, writing empty strings to the database. Use TryFrom or return Result<String, ...> to propagate the error.

[maralbahari]

@franciscojavierarceo done

[ashwing]

Few comments

get_items fetches via WHERE id IN (...) but SQL doesn't guarantee result order matches the input list. ResponseStore::rehydrate passes the ordered history_item_ids, so items can come back shuffled — breaking conversation ordering. A post-fetch reindex against the input would fix it:

let mut map: HashMap<String, Item> = rows.into_iter().map(|r| (r.id.clone(), r)).collect();
let ordered: Vec<Item> = ids.iter().filter_map(|id| map.remove(id)).collect();

Separately — rehydrate uses filter_map(|row| row.as_inout()) which silently drops items that can't deserialize as either InputItem or OutputItem. Combined with francisco's point about needing an Unknown variant, this means any item type the enum doesn't cover today (reasoning, web_search_call, mcp_approval, etc.) will vanish from the rehydrated history. Might be worth at least a tracing::warn! on the None path so operators notice when items are being dropped.

[franciscojavierarceo]

FYI — we'll also want to add a raw_tokens TEXT placeholder column (on items and/or responses) so we can store the unmodified token sequence and pass it back to vLLM for prefix cache reuse. This will require adding token pass-through support to the /v1/responses endpoint upstream in vllm-project/vllm, since it currently only renders messages to tokens internally. The /v1/completions endpoint already supports prompt: list[int] so the plumbing exists — we just need to expose it through the Responses API.

[maralbahari]

@franciscojavierarceo I think we should add those columns when the feature is actually supported than a placeholder. cause it would be out of the scope of the PR ?

[maralbahari]

Few comments

get_items fetches via WHERE id IN (...) but SQL doesn't guarantee result order matches the input list. ResponseStore::rehydrate passes the ordered history_item_ids, so items can come back shuffled — breaking conversation ordering. A post-fetch reindex against the input would fix it:

let mut map: HashMap<String, Item> = rows.into_iter().map(|r| (r.id.clone(), r)).collect();
let ordered: Vec<Item> = ids.iter().filter_map(|id| map.remove(id)).collect();

Separately — rehydrate uses filter_map(|row| row.as_inout()) which silently drops items that can't deserialize as either InputItem or OutputItem. Combined with francisco's point about needing an Unknown variant, this means any item type the enum doesn't cover today (reasoning, web_search_call, mcp_approval, etc.) will vanish from the rehydrated history. Might be worth at least a tracing::warn! on the None path so operators notice when items are being dropped.

@ashwing Thanks for the suggestion. added suggestion for the preserving the order of item ids.
as for the serde error handling. now all the serde utility functions in common.rs throw the error and if need to gracefully return default are separated. for the care of input/output items are wrapped around TryFrom trait to handle the error.

[maralbahari]

This is great!

Can we handle a few items to before merging?

1. Add tenant_id TEXT placeholder columns to all three tables with indexes on (tenant_id, id) — avoids a schema migration when multi-tenancy lands.
2. Add raw_tokens TEXT placeholder column on items and/or responses — so we can store the unmodified token sequence and pass it back to vLLM for faster inference (prefix caching / KV-cache reuse) without a migration later.
3. Add a fallback variant for unknown item types — InputItem and OutputItem only handle message and function_call/function_call_output. For reference, ogx's ConversationItem handles ~10 types (web search, file search, MCP approval, MCP call, reasoning, compaction, etc.). Unknown types silently disappear via filter_map in as_inout(). At minimum, add an Unknown(serde_json::Value) variant so data is preserved round-trip.
4. Add metadata to the Conversation struct — the migration SQL has a metadata TEXT column but the FromRow struct in models/conversation.rs only has id and created_at. The field is silently dropped on read.

Note on crate structure

PR #35 (move gateway to agentic-server) was reverted in PR #40, so this PR's crates/agentic-core/ paths will need to be rebased accordingly.

1-2: I think we can add those columns when the feature is supported just placeholder doesnt seem to be doing anything.

3: just for the simplicity input output text message added in this PR and the serialize and deserialize errors are handled as you mentioned in another comment. other types can be added as the tool calling features are developed. while in this PR there are some structs for other types but not utilized yet. we can implement the rust native version of types handle on ogx here in core too overtime as development goes forward.

4: the metadata column of conversation table in migration is removed now we dont really need it as the metadata is stored in response table already.