1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
use crate::catalog::backend::{BackendError, Row, SqlValue, TxOptions};
use crate::error::Result;
use crate::model_task::ModelTask;
use super::Catalog;
/// Materialized row from the `models` catalog table.
#[derive(Debug, Clone, serde::Serialize)]
pub struct ModelRecord {
/// Unique model name (e.g., `"sentence-transformers/all-MiniLM-L6-v2"`).
pub model_id: String,
/// Monotonically increasing version number for this model name.
pub version: i32,
/// Model category (e.g., `"embedding"`, `"llm"`, `"lora"`).
pub model_type: String,
/// Parent model this was derived from (fine-tuned or adapted).
pub base_model_id: Option<String>,
/// Inference backend (e.g., `"candle"`, `"vllm"`, `"http"`).
pub backend: String,
/// Task this model performs.
pub task: ModelTask,
/// Filesystem path to model weights or adapter files.
pub artifact_path: Option<String>,
/// Serialized JSON blob with backend-specific configuration.
pub config_json: Option<String>,
/// Lifecycle status (e.g., `"registered"`, `"loaded"`, `"failed"`).
pub status: String,
/// ISO-8601 timestamp of initial registration.
pub created_at: String,
}
/// Input parameters for [`Catalog::register_model`].
#[derive(Debug)]
pub struct RegisterModelParams<'a> {
/// Unique model name.
pub model_id: &'a str,
/// Version number for this registration.
pub version: i32,
/// Model category (e.g., `"embedding"`, `"llm"`).
pub model_type: &'a str,
/// Inference backend identifier.
pub backend: &'a str,
/// Task this model performs.
pub task: ModelTask,
/// Optional parent model ID (for fine-tuned variants).
pub base_model_id: Option<&'a str>,
/// Optional filesystem path to model weights.
pub artifact_path: Option<&'a str>,
/// Optional JSON blob with backend-specific settings.
pub config_json: Option<&'a str>,
}
const SELECT_COLS: &str =
"model_id, name, model_type, task, backend, version, status, metadata, created_at";
impl Catalog {
/// Register or refresh a model in the catalog. The session's bound
/// tenant is written to `tenant_id` and asserted before INSERT.
pub async fn register_model(&self, params: RegisterModelParams<'_>) -> Result<()> {
let pk = format!("{}::{}", params.model_id, params.version);
let metadata = serde_json::json!({
"base_model_id": params.base_model_id,
"artifact_path": params.artifact_path,
"config_json": params.config_json,
})
.to_string();
let model_id = params.model_id.to_string();
let model_type = params.model_type.to_string();
let task = params.task.as_db_str();
let backend = params.backend.to_string();
let version = params.version as i64;
let tenant = self.current_tenant();
self.backend()
.transaction(TxOptions::default(), |tx| {
Box::pin(async move {
tx.set_tenant(tenant);
tx.assert_tenant_matches(tenant, "models")?;
tx.execute(
"INSERT INTO models (model_id, name, model_type, task, backend, version, status, metadata, tenant_id) \
VALUES ($1, $2, $3, $4, $5, $6, 'registered', $7, $8) \
ON CONFLICT(model_id) DO UPDATE SET \
metadata = excluded.metadata, \
backend = excluded.backend, \
task = excluded.task, \
model_type = excluded.model_type, \
updated_at = CAST(CURRENT_TIMESTAMP AS TEXT)",
&[
SqlValue::TextOwned(pk),
SqlValue::TextOwned(model_id),
SqlValue::TextOwned(model_type),
SqlValue::Text(task),
SqlValue::TextOwned(backend),
SqlValue::Int(version),
SqlValue::TextOwned(metadata),
SqlValue::from(tenant.map(|t| t.to_string())),
],
)
.await?;
Ok(())
})
})
.await?;
Ok(())
}
/// Get the latest version of a model by name. Tenant-filtered.
pub async fn get_model(&self, model_id: &str) -> Result<Option<ModelRecord>> {
let sql = format!(
"SELECT {SELECT_COLS} FROM models \
WHERE name = $1 AND (tenant_id = $2 OR tenant_id IS NULL) \
ORDER BY version DESC LIMIT 1"
);
let mid = model_id.to_string();
let tenant = self.current_tenant();
Ok(self
.backend()
.transaction(
TxOptions {
read_only: true,
..Default::default()
},
|tx| {
Box::pin(async move {
tx.query_opt(
&sql,
&[
SqlValue::TextOwned(mid),
SqlValue::from(tenant.map(|t| t.to_string())),
],
parse_model_row,
)
.await
})
},
)
.await?)
}
/// Get a specific version of a model.
pub async fn get_model_version(
&self,
model_id: &str,
version: i32,
) -> Result<Option<ModelRecord>> {
let sql = format!(
"SELECT {SELECT_COLS} FROM models \
WHERE name = $1 AND version = $2 \
AND (tenant_id = $3 OR tenant_id IS NULL)"
);
let mid = model_id.to_string();
let v = version as i64;
let tenant = self.current_tenant();
Ok(self
.backend()
.transaction(
TxOptions {
read_only: true,
..Default::default()
},
|tx| {
Box::pin(async move {
tx.query_opt(
&sql,
&[
SqlValue::TextOwned(mid),
SqlValue::Int(v),
SqlValue::from(tenant.map(|t| t.to_string())),
],
parse_model_row,
)
.await
})
},
)
.await?)
}
/// Update model status. Scoped to the session's tenant.
pub async fn update_model_status(
&self,
model_id: &str,
status: super::status::ModelStatus,
) -> Result<()> {
let status_str = status.to_string();
let mid = model_id.to_string();
let tenant = self.current_tenant();
self.backend()
.transaction(TxOptions::default(), |tx| {
Box::pin(async move {
tx.set_tenant(tenant);
tx.execute(
"UPDATE models SET status = $1, updated_at = CAST(CURRENT_TIMESTAMP AS TEXT) \
WHERE name = $2 AND (tenant_id = $3 OR tenant_id IS NULL)",
&[
SqlValue::TextOwned(status_str),
SqlValue::TextOwned(mid),
SqlValue::from(tenant.map(|t| t.to_string())),
],
)
.await?;
Ok(())
})
})
.await?;
Ok(())
}
/// List models visible to the session's tenant.
pub async fn list_models(&self) -> Result<Vec<ModelRecord>> {
let sql = format!(
"SELECT {SELECT_COLS} FROM models \
WHERE tenant_id = $1 OR tenant_id IS NULL \
ORDER BY created_at"
);
let tenant = self.current_tenant();
Ok(self
.backend()
.transaction(
TxOptions {
read_only: true,
..Default::default()
},
|tx| {
Box::pin(async move {
tx.query(
&sql,
&[SqlValue::from(tenant.map(|t| t.to_string()))],
parse_model_row,
)
.await
})
},
)
.await?)
}
}
/// Parse: model_id, name, model_type, task, backend, version, status, metadata, created_at
fn parse_model_row(row: &Row<'_>) -> std::result::Result<ModelRecord, BackendError> {
let _pk: String = row.get("model_id")?;
let name: String = row.get("name")?;
let model_type: String = row.get("model_type")?;
let task_raw: String = row.get("task")?;
let task = ModelTask::try_from_db_str(&task_raw).map_err(|e| BackendError::TypeConversion {
column: "task".into(),
detail: e.to_string(),
})?;
let backend: String = row.try_get("backend")?.unwrap_or_default();
let version: i64 = row.try_get("version")?.unwrap_or(1);
let status: String = row.try_get("status")?.unwrap_or_default();
let metadata: Option<String> = row.try_get("metadata")?;
let created_at: String = row.try_get("created_at")?.unwrap_or_default();
let (base_model_id, artifact_path, config_json) = metadata
.as_deref()
.and_then(|m| serde_json::from_str::<serde_json::Value>(m).ok())
.map(|v| {
(
v["base_model_id"].as_str().map(String::from),
v["artifact_path"].as_str().map(String::from),
v["config_json"].as_str().map(String::from),
)
})
.unwrap_or((None, None, None));
Ok(ModelRecord {
model_id: name,
version: version as i32,
model_type,
base_model_id,
backend,
task,
artifact_path,
config_json,
status,
created_at,
})
}