lean_ctx/gateway_server/mcp/
store.rs

1//! `mcp_events` + `mcp_tool_inventory` Postgres store (GL#102/#103).
2//!
3//! Deliberately **separate tables** from `usage_events` (a documented
4//! deviation from Doc 15 §7's "new dimension" sketch): LLM spend and tool
5//! context cost are different currencies, and folding MCP rows into
6//! `usage_events` would silently inflate every existing spend report,
7//! projection and evidence export. Attribution still joins on `person`/
8//! `team`/`project` — the identity plane is shared.
9//!
10//! Lifecycle parity with the LLM channel is non-negotiable:
11//! - Retention: `[gateway_server].usage_retention_days` purges both tables.
12//! - GDPR: `gateway ln export|delete` covers both tables (Art. 15/17).
13//! - Fail-open: inserts are queued by `metering::spawn_writer`; errors are
14//!   logged and counted, never propagated to the tool-traffic path.
15//!
16//! Schema management follows the repo rule: idempotent `CREATE … IF NOT
17//! EXISTS` DDL run on every start, no migration files.
18
19use deadpool_postgres::Pool;
20
21/// Idempotent DDL, run on every `gateway serve` start (same contract as
22/// `USAGE_EVENTS_DDL`).
23const MCP_DDL: &str = r"
24CREATE TABLE IF NOT EXISTS mcp_events (
25  id               BIGSERIAL PRIMARY KEY,
26  ts               TIMESTAMPTZ      NOT NULL DEFAULT now(),
27  person           TEXT             NOT NULL,
28  team             TEXT,
29  project          TEXT             NOT NULL,
30  server_id        TEXT             NOT NULL,
31  method           TEXT             NOT NULL,
32  tool             TEXT,
33  status           TEXT             NOT NULL,
34  duration_ms      BIGINT           NOT NULL DEFAULT 0,
35  result_bytes     BIGINT           NOT NULL DEFAULT 0,
36  result_tokens    BIGINT           NOT NULL DEFAULT 0,
37  context_cost_usd DOUBLE PRECISION NOT NULL DEFAULT 0,
38  reference_model  TEXT
39);
40CREATE INDEX IF NOT EXISTS idx_mcp_events_person_ts ON mcp_events (person, ts);
41CREATE INDEX IF NOT EXISTS idx_mcp_events_server_ts ON mcp_events (server_id, ts);
42CREATE INDEX IF NOT EXISTS idx_mcp_events_tool_ts   ON mcp_events (server_id, tool, ts);
43CREATE TABLE IF NOT EXISTS mcp_tool_inventory (
44  server_id       TEXT        NOT NULL,
45  tool            TEXT        NOT NULL,
46  schema_sha256   TEXT        NOT NULL,
47  previous_sha256 TEXT,
48  first_seen      TIMESTAMPTZ NOT NULL DEFAULT now(),
49  last_seen       TIMESTAMPTZ NOT NULL DEFAULT now(),
50  change_count    BIGINT      NOT NULL DEFAULT 0,
51  PRIMARY KEY (server_id, tool)
52);
53";
54
55/// Applies the MCP-store DDL. Safe to run on every start (idempotent).
56pub async fn init_schema(pool: &Pool) -> anyhow::Result<()> {
57    let client = pool.get().await?;
58    client.batch_execute(MCP_DDL).await?;
59    Ok(())
60}
61
62/// One measured MCP exchange, ready for insertion.
63#[derive(Debug, Clone, PartialEq)]
64pub struct McpEvent {
65    pub person: String,
66    pub team: Option<String>,
67    pub project: String,
68    pub server_id: String,
69    /// JSON-RPC method label (`tools/call`, `tools/list`, `passthrough`, …).
70    pub method: String,
71    /// Tool name for `tools/call`; `None` otherwise.
72    pub tool: Option<String>,
73    /// `ok` | `error` (JSON-RPC error frame) | `upstream_error` (transport).
74    pub status: String,
75    pub duration_ms: i64,
76    pub result_bytes: i64,
77    pub result_tokens: i64,
78    /// `result_tokens` priced at the reference model's input rate — what this
79    /// tool context costs every time it is sent on to an LLM. `0.0` when no
80    /// `[proxy.baseline].reference_model` is configured (never invented).
81    pub context_cost_usd: f64,
82    pub reference_model: Option<String>,
83}
84
85/// Inserts one event. Errors bubble to the writer loop, which logs and moves on.
86pub async fn insert_event(client: &deadpool_postgres::Client, e: &McpEvent) -> anyhow::Result<()> {
87    client
88        .execute(
89            "INSERT INTO mcp_events \
90             (person, team, project, server_id, method, tool, status, \
91              duration_ms, result_bytes, result_tokens, context_cost_usd, reference_model) \
92             VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12)",
93            &[
94                &e.person,
95                &e.team,
96                &e.project,
97                &e.server_id,
98                &e.method,
99                &e.tool,
100                &e.status,
101                &e.duration_ms,
102                &e.result_bytes,
103                &e.result_tokens,
104                &e.context_cost_usd,
105                &e.reference_model,
106            ],
107        )
108        .await?;
109    Ok(())
110}
111
112/// Records a `tools/list` snapshot for one server: upsert per tool, bumping
113/// `change_count` and remembering the previous hash whenever the definition
114/// fingerprint moved (the rug-pull trail, GL#103). Tools that disappeared
115/// from the listing keep their last row — an inventory is also a history.
116pub async fn upsert_inventory(
117    client: &deadpool_postgres::Client,
118    server_id: &str,
119    tools: &[super::frames::ToolDef],
120) -> anyhow::Result<()> {
121    let stmt = client
122        .prepare_cached(
123            "INSERT INTO mcp_tool_inventory (server_id, tool, schema_sha256) \
124             VALUES ($1, $2, $3) \
125             ON CONFLICT (server_id, tool) DO UPDATE SET \
126               last_seen       = now(), \
127               previous_sha256 = CASE WHEN mcp_tool_inventory.schema_sha256 <> EXCLUDED.schema_sha256 \
128                                      THEN mcp_tool_inventory.schema_sha256 \
129                                      ELSE mcp_tool_inventory.previous_sha256 END, \
130               change_count    = mcp_tool_inventory.change_count + \
131                                 CASE WHEN mcp_tool_inventory.schema_sha256 <> EXCLUDED.schema_sha256 \
132                                      THEN 1 ELSE 0 END, \
133               schema_sha256   = EXCLUDED.schema_sha256",
134        )
135        .await?;
136    for t in tools {
137        client
138            .execute(&stmt, &[&server_id, &t.name, &t.schema_sha256])
139            .await?;
140    }
141    Ok(())
142}
143
144/// Deletes `mcp_events` rows older than `days` (retention parity with
145/// `usage_events`, enterprise#36). The inventory is config-scale metadata,
146/// not per-person telemetry — it is never purged by retention.
147pub async fn purge_events_older_than(pool: &Pool, days: u32) -> anyhow::Result<u64> {
148    let client = pool.get().await?;
149    let purged = client
150        .execute(
151            "DELETE FROM mcp_events WHERE ts < now() - make_interval(days => $1)",
152            &[&i32::try_from(days).unwrap_or(i32::MAX)],
153        )
154        .await?;
155    Ok(purged)
156}
157
158/// All MCP events attributed to one of `person_keys` (raw + pseudonym) —
159/// GDPR Art. 15 export, same contract as `store::person_events`.
160pub async fn person_events(
161    pool: &Pool,
162    person_keys: &[String],
163) -> anyhow::Result<Vec<serde_json::Value>> {
164    let client = pool.get().await?;
165    let rows = client
166        .query(
167            "SELECT to_jsonb(mcp_events) FROM mcp_events \
168             WHERE person = ANY($1) ORDER BY ts",
169            &[&person_keys],
170        )
171        .await?;
172    Ok(rows
173        .into_iter()
174        .map(|r| r.get::<_, serde_json::Value>(0))
175        .collect())
176}
177
178/// Deletes all MCP events of `person_keys` (GDPR Art. 17). Returns rows removed.
179pub async fn delete_person_events(pool: &Pool, person_keys: &[String]) -> anyhow::Result<u64> {
180    let client = pool.get().await?;
181    let deleted = client
182        .execute(
183            "DELETE FROM mcp_events WHERE person = ANY($1)",
184            &[&person_keys],
185        )
186        .await?;
187    Ok(deleted)
188}
189
190/// Aggregated per-server × tool activity for the admin window (console
191/// "Tools" section). Stable ordering: cost desc, then name — deterministic
192/// output for identical database contents (#498).
193pub const TOOL_BREAKDOWN_SQL: &str = "
194SELECT server_id,
195       coalesce(tool, method)        AS tool,
196       count(*)                      AS calls,
197       count(*) FILTER (WHERE status <> 'ok') AS errors,
198       count(DISTINCT person)        AS persons,
199       sum(result_tokens)::BIGINT    AS result_tokens,
200       sum(context_cost_usd)         AS context_cost_usd,
201       max(duration_ms)              AS max_duration_ms,
202       percentile_cont(0.5) WITHIN GROUP (ORDER BY duration_ms) AS p50_duration_ms
203FROM mcp_events
204WHERE ts >= $1 AND ts <= $2
205GROUP BY server_id, coalesce(tool, method)
206ORDER BY context_cost_usd DESC, tool";
207
208/// Per-person MCP totals for `/me` ("your tools").
209pub const ME_TOOLS_SQL: &str = "
210SELECT server_id,
211       coalesce(tool, method)     AS tool,
212       count(*)                   AS calls,
213       sum(result_tokens)::BIGINT AS result_tokens,
214       sum(context_cost_usd)      AS context_cost_usd
215FROM mcp_events
216WHERE ts >= $1 AND ts <= $2 AND person = $3
217GROUP BY server_id, coalesce(tool, method)
218ORDER BY context_cost_usd DESC, tool
219LIMIT 50";
220
221/// Window totals for the admin summary strip.
222pub const TOTALS_SQL: &str = "
223SELECT count(*)                                AS calls,
224       count(*) FILTER (WHERE status <> 'ok')  AS errors,
225       count(DISTINCT person)                  AS persons,
226       coalesce(sum(result_tokens), 0)::BIGINT AS result_tokens,
227       coalesce(sum(context_cost_usd), 0)      AS context_cost_usd
228FROM mcp_events
229WHERE ts >= $1 AND ts <= $2";
230
231/// Inventory listing with live hash status. `changed` surfaces every tool
232/// whose definition fingerprint moved at least once — the observe-stage
233/// rug-pull signal (enforcement pins hashes in M4).
234pub const INVENTORY_SQL: &str = "
235SELECT server_id, tool, schema_sha256, previous_sha256,
236       change_count,
237       to_char(first_seen AT TIME ZONE 'utc', 'YYYY-MM-DD') AS first_seen,
238       to_char(last_seen  AT TIME ZONE 'utc', 'YYYY-MM-DD') AS last_seen
239FROM mcp_tool_inventory
240ORDER BY server_id, tool";
241
242#[cfg(test)]
243mod tests {
244    use super::*;
245
246    #[test]
247    fn ddl_is_idempotent_by_construction() {
248        for stmt in ["CREATE TABLE", "CREATE INDEX"] {
249            for (i, _) in MCP_DDL.match_indices(stmt) {
250                let tail = &MCP_DDL[i..(i + stmt.len() + 14).min(MCP_DDL.len())];
251                assert!(
252                    tail.contains("IF NOT EXISTS"),
253                    "non-idempotent DDL statement: {tail}"
254                );
255            }
256        }
257    }
258
259    #[test]
260    fn schema_carries_the_observe_columns() {
261        // The columns the observe stage's queries and the M4 enforce stage's
262        // pinning depend on — a rename here is a breaking change.
263        for col in [
264            "server_id",
265            "method",
266            "tool",
267            "status",
268            "result_tokens",
269            "context_cost_usd",
270            "schema_sha256",
271            "previous_sha256",
272            "change_count",
273        ] {
274            assert!(MCP_DDL.contains(col), "column {col} missing from DDL");
275        }
276    }
277
278    #[test]
279    fn aggregate_sql_is_window_bounded_and_deterministically_ordered() {
280        for sql in [TOOL_BREAKDOWN_SQL, ME_TOOLS_SQL, TOTALS_SQL] {
281            assert!(
282                sql.contains("ts >= $1 AND ts <= $2"),
283                "window bounds: {sql}"
284            );
285        }
286        for sql in [TOOL_BREAKDOWN_SQL, ME_TOOLS_SQL, INVENTORY_SQL] {
287            assert!(sql.contains("ORDER BY"), "stable ordering required: {sql}");
288        }
289    }
290}
lean_ctx/gateway_server/mcp/store.rs

lean_ctx/gateway_server/mcp/
store.rs
