fn memory_promote_metadata(
metadata: Option<&Value>,
request: &MemoryPromoteRequest,
promoted_at_ms: u64,
) -> Option<Value> {
let mut obj = metadata
.and_then(Value::as_object)
.cloned()
.unwrap_or_default();
obj.insert(
"promotion".to_string(),
json!({
"promoted_at_ms": promoted_at_ms,
"promote_run_id": request.run_id,
"source_memory_id": request.source_memory_id,
"from_tier": request.from_tier,
"to_tier": request.to_tier,
"reason": request.reason,
"review": {
"required": request.review.required,
"reviewer_id": request.review.reviewer_id,
"approval_id": request.review.approval_id,
},
}),
);
Some(Value::Object(obj))
}
fn memory_promote_provenance(
provenance: Option<&Value>,
request: &MemoryPromoteRequest,
partition_key: &str,
promoted_at_ms: u64,
tenant_context: &TenantContext,
) -> Value {
let mut obj = provenance
.and_then(Value::as_object)
.cloned()
.unwrap_or_default();
obj.insert(
"promotion".to_string(),
json!({
"promoted_at_ms": promoted_at_ms,
"promote_run_id": request.run_id,
"source_memory_id": request.source_memory_id,
"partition_key": partition_key,
"to_tier": request.to_tier,
"reviewer_id": request.review.reviewer_id,
"approval_id": request.review.approval_id,
"tenant_context": tenant_context,
}),
);
Value::Object(obj)
}
fn memory_linkage(record: &GlobalMemoryRecord) -> Value {
memory_linkage_from_parts(
&record.run_id,
record.project_tag.as_deref(),
record.metadata.as_ref(),
record.provenance.as_ref(),
)
}
fn memory_linkage_from_parts(
run_id: &str,
project_id: Option<&str>,
metadata: Option<&Value>,
provenance: Option<&Value>,
) -> Value {
let artifact_refs = memory_artifact_refs(metadata);
json!({
"run_id": run_id,
"project_id": project_id,
"origin_event_type": provenance
.and_then(|row| row.get("origin_event_type"))
.and_then(Value::as_str),
"origin_run_id": provenance
.and_then(|row| row.get("origin_run_id"))
.and_then(Value::as_str)
.or(Some(run_id)),
"origin_session_id": provenance
.and_then(|row| row.get("origin_session_id"))
.and_then(Value::as_str),
"origin_message_id": provenance
.and_then(|row| row.get("origin_message_id"))
.and_then(Value::as_str),
"partition_key": provenance
.and_then(|row| row.get("partition_key"))
.and_then(Value::as_str),
"promote_run_id": provenance
.and_then(|row| row.get("promotion"))
.and_then(|row| row.get("promote_run_id"))
.and_then(Value::as_str),
"approval_id": provenance
.and_then(|row| row.get("promotion"))
.and_then(|row| row.get("approval_id"))
.and_then(Value::as_str),
"artifact_refs": artifact_refs,
})
}
fn memory_kind_label(source_type: &str) -> &str {
match source_type {
"solution_capsule" => "solution_capsule",
"note" => "note",
"fact" => "fact",
other => other,
}
}
fn memory_linkage_detail(linkage: &Value) -> String {
let origin_run_id = linkage
.get("origin_run_id")
.and_then(Value::as_str)
.unwrap_or_default();
let project_id = linkage
.get("project_id")
.and_then(Value::as_str)
.unwrap_or_default();
let promote_run_id = linkage
.get("promote_run_id")
.and_then(Value::as_str)
.unwrap_or_default();
format!(
" origin_run_id={} project_id={} promote_run_id={}",
origin_run_id, project_id, promote_run_id
)
}
fn memory_kind_for_request(kind: tandem_memory::MemoryContentKind) -> &'static str {
match kind {
tandem_memory::MemoryContentKind::SolutionCapsule => "solution_capsule",
tandem_memory::MemoryContentKind::Note => "note",
tandem_memory::MemoryContentKind::Fact => "fact",
}
}
fn memory_tier_for_visibility(visibility: &str) -> tandem_memory::GovernedMemoryTier {
if visibility.eq_ignore_ascii_case("shared") {
tandem_memory::GovernedMemoryTier::Project
} else {
tandem_memory::GovernedMemoryTier::Session
}
}
fn memory_classification_label(metadata: Option<&Value>) -> &str {
metadata
.and_then(|row| row.get("classification"))
.and_then(Value::as_str)
.filter(|value| !value.trim().is_empty())
.unwrap_or("internal")
}
pub(super) fn scrub_content(input: &str) -> ScrubReport {
let mut redactions = 0u32;
let mut blocked = false;
let lower = input.to_lowercase();
let redact_markers = [
"api_key",
"secret=",
"authorization: bearer",
"x-api-key",
"token=",
];
for marker in redact_markers {
if lower.contains(marker) {
redactions = redactions.saturating_add(1);
}
}
let block_markers = [
"-----begin private key-----",
"aws_secret_access_key",
"sk-ant-",
"ghp_",
];
for marker in block_markers {
if lower.contains(marker) {
blocked = true;
break;
}
}
if blocked {
ScrubReport {
status: ScrubStatus::Blocked,
redactions,
block_reason: Some("sensitive secret marker detected".to_string()),
}
} else if redactions > 0 {
ScrubReport {
status: ScrubStatus::Redacted,
redactions,
block_reason: None,
}
} else {
ScrubReport {
status: ScrubStatus::Passed,
redactions: 0,
block_reason: None,
}
}
}
pub(super) fn scrub_content_for_memory(input: &str) -> (String, ScrubReport) {
let mut scrubbed = input.to_string();
let mut redactions = 0u32;
let mut blocked = false;
let redact_patterns = [
r"(?i)authorization:\s*bearer\s+[a-z0-9\.\-_]+",
r"(?i)(api[_-]?key|token|secret)\s*[:=]\s*[a-z0-9\-_]{8,}",
r"(?i)x-api-key\s*:\s*[a-z0-9\-_]{8,}",
r"(?i)sk-[a-z0-9]{12,}",
r"(?i)ghp_[a-z0-9]{12,}",
];
for pattern in redact_patterns {
if let Ok(re) = Regex::new(pattern) {
let matches = re.find_iter(&scrubbed).count() as u32;
if matches > 0 {
redactions = redactions.saturating_add(matches);
scrubbed = re.replace_all(&scrubbed, "[REDACTED]").to_string();
}
}
}
let block_markers = [
"-----begin private key-----",
"aws_secret_access_key",
"-----begin rsa private key-----",
];
let lowered = input.to_lowercase();
for marker in block_markers {
if lowered.contains(marker) {
blocked = true;
break;
}
}
if blocked {
(
String::new(),
ScrubReport {
status: ScrubStatus::Blocked,
redactions,
block_reason: Some("sensitive secret marker detected".to_string()),
},
)
} else if redactions > 0 {
(
scrubbed,
ScrubReport {
status: ScrubStatus::Redacted,
redactions,
block_reason: None,
},
)
} else {
(
scrubbed,
ScrubReport {
status: ScrubStatus::Passed,
redactions: 0,
block_reason: None,
},
)
}
}
pub(super) fn hash_text(input: &str) -> String {
let mut hasher = Sha256::new();
hasher.update(input.as_bytes());
format!("{:x}", hasher.finalize())
}
pub(super) async fn append_memory_audit(
state: &AppState,
tenant_context: &TenantContext,
mut event: crate::MemoryAuditEvent,
) -> Result<(), StatusCode> {
event.tenant_context = tenant_context.clone();
if let Some(parent) = state.memory_audit_path.parent() {
tokio::fs::create_dir_all(parent)
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
}
let line = serde_json::to_string(&event).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
let mut file = tokio::fs::OpenOptions::new()
.create(true)
.append(true)
.open(&state.memory_audit_path)
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
tokio::io::AsyncWriteExt::write_all(&mut file, line.as_bytes())
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
tokio::io::AsyncWriteExt::write_all(&mut file, b"\n")
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
file.sync_data()
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
let mut audit = state.memory_audit_log.write().await;
audit.push(event);
Ok(())
}
async fn load_memory_audit_events(path: &std::path::Path) -> Vec<crate::MemoryAuditEvent> {
let Ok(content) = tokio::fs::read_to_string(path).await else {
return Vec::new();
};
content
.lines()
.filter_map(|line| {
let trimmed = line.trim();
if trimmed.is_empty() {
return None;
}
serde_json::from_str::<crate::MemoryAuditEvent>(trimmed).ok()
})
.collect()
}
#[derive(Debug, Clone)]
pub(super) struct RunMemoryContext {
run_id: String,
user_id: String,
started_at_ms: u64,
host_tag: Option<String>,
tenant_context: TenantContext,
}
pub(super) async fn open_global_memory_db() -> Option<MemoryDatabase> {
let paths = tandem_core::resolve_shared_paths().ok()?;
if let Some(parent) = paths.memory_db_path.parent() {
let _ = tokio::fs::create_dir_all(parent).await;
}
MemoryDatabase::new(&paths.memory_db_path).await.ok()
}
pub(super) async fn open_global_memory_db_for_state(state: &AppState) -> Option<MemoryDatabase> {
if let Some(parent) = state.memory_db_path.parent() {
let _ = tokio::fs::create_dir_all(parent).await;
}
MemoryDatabase::new(&state.memory_db_path).await.ok()
}
pub(super) async fn open_memory_manager() -> Option<tandem_memory::MemoryManager> {
let paths = tandem_core::resolve_shared_paths().ok()?;
if let Some(parent) = paths.memory_db_path.parent() {
let _ = tokio::fs::create_dir_all(parent).await;
}
tandem_memory::MemoryManager::new(&paths.memory_db_path)
.await
.ok()
}
pub(super) async fn open_memory_manager_for_state(
state: &AppState,
) -> Option<tandem_memory::MemoryManager> {
if let Some(parent) = state.memory_db_path.parent() {
let _ = tokio::fs::create_dir_all(parent).await;
}
tandem_memory::MemoryManager::new(&state.memory_db_path)
.await
.ok()
}
pub(super) fn event_run_id(event: &EngineEvent) -> Option<String> {
event
.properties
.get("runID")
.or_else(|| event.properties.get("run_id"))
.and_then(|v| v.as_str())
.map(ToString::to_string)
}
pub(super) fn event_session_id(event: &EngineEvent) -> Option<String> {
event
.properties
.get("sessionID")
.or_else(|| event.properties.get("sessionId"))
.and_then(|v| v.as_str())
.map(ToString::to_string)
}
pub(super) fn summarize_value(value: &Value, limit: usize) -> String {
let text = if value.is_string() {
value.as_str().unwrap_or_default().to_string()
} else {
value.to_string()
};
truncate_text(&text, limit)
}
pub(super) async fn persist_global_memory_record(
state: &AppState,
db: &MemoryDatabase,
mut record: GlobalMemoryRecord,
) {
let tenant_context = record_tenant_context(&record);
publish_tenant_event(
state,
&tenant_context,
"memory.write.attempted",
json!({
"runID": record.run_id,
"sourceType": record.source_type,
"sessionID": record.session_id,
"messageID": record.message_id,
}),
);
let (scrubbed, scrub) = scrub_content_for_memory(&record.content);
if scrub.status == ScrubStatus::Blocked || scrubbed.trim().is_empty() {
publish_tenant_event(
state,
&tenant_context,
"memory.write.skipped",
json!({
"runID": record.run_id,
"sourceType": record.source_type,
"reason": scrub.block_reason.unwrap_or_else(|| "scrub_blocked".to_string()),
"sessionID": record.session_id,
"messageID": record.message_id,
}),
);
return;
}
record.content = scrubbed;
record.redaction_count = scrub.redactions;
record.redaction_status = match scrub.status {
ScrubStatus::Passed => "passed".to_string(),
ScrubStatus::Redacted => "redacted".to_string(),
ScrubStatus::Blocked => "blocked".to_string(),
};
record.content_hash = hash_text(&record.content);
match db.put_global_memory_record(&record).await {
Ok(write) => {
let event_name = if write.deduped {
"memory.write.skipped"
} else {
"memory.write.succeeded"
};
publish_tenant_event(
state,
&tenant_context,
event_name,
json!({
"runID": record.run_id,
"memoryID": write.id,
"sourceType": record.source_type,
"deduped": write.deduped,
"redactionStatus": record.redaction_status,
"redactionCount": record.redaction_count,
"sessionID": record.session_id,
"messageID": record.message_id,
}),
);
}
Err(err) => {
publish_tenant_event(
state,
&tenant_context,
"memory.write.skipped",
json!({
"runID": record.run_id,
"sourceType": record.source_type,
"reason": format!("db_error:{err}"),
"sessionID": record.session_id,
"messageID": record.message_id,
}),
);
}
}
}
pub(super) async fn ingest_run_messages(
state: &AppState,
db: &MemoryDatabase,
session_id: &str,
ctx: &RunMemoryContext,
) {
let Some(session) = state.storage.get_session(session_id).await else {
return;
};
for message in session.messages {
let created_ms = message.created_at.timestamp_millis() as u64;
if created_ms + 1_000 < ctx.started_at_ms {
continue;
}
for part in message.parts {
match (message.role.clone(), part) {
(MessageRole::User, MessagePart::Text { text }) => {
let now = crate::now_ms();
persist_global_memory_record(
state,
db,
GlobalMemoryRecord {
id: Uuid::new_v4().to_string(),
user_id: ctx.user_id.clone(),
source_type: "user_message".to_string(),
content: text,
content_hash: String::new(),
run_id: ctx.run_id.clone(),
session_id: Some(session_id.to_string()),
message_id: Some(message.id.clone()),
tool_name: None,
project_tag: session.project_id.clone(),
channel_tag: None,
host_tag: ctx.host_tag.clone(),
metadata: Some(json!({"role": "user"})),
provenance: Some(json!({"origin_event_type": "session.run.finished", "origin_message_id": message.id, "origin_session_id": session_id})),
redaction_status: "passed".to_string(),
redaction_count: 0,
visibility: "private".to_string(),
demoted: false,
score_boost: 0.0,
created_at_ms: now,
updated_at_ms: now,
expires_at_ms: None,
},
)
.await;
}
(MessageRole::Assistant, MessagePart::Text { text }) => {
let now = crate::now_ms();
persist_global_memory_record(
state,
db,
GlobalMemoryRecord {
id: Uuid::new_v4().to_string(),
user_id: ctx.user_id.clone(),
source_type: "assistant_final".to_string(),
content: text,
content_hash: String::new(),
run_id: ctx.run_id.clone(),
session_id: Some(session_id.to_string()),
message_id: Some(message.id.clone()),
tool_name: None,
project_tag: session.project_id.clone(),
channel_tag: None,
host_tag: ctx.host_tag.clone(),
metadata: Some(json!({"role": "assistant"})),
provenance: Some(json!({"origin_event_type": "session.run.finished", "origin_message_id": message.id, "origin_session_id": session_id})),
redaction_status: "passed".to_string(),
redaction_count: 0,
visibility: "private".to_string(),
demoted: false,
score_boost: 0.0,
created_at_ms: now,
updated_at_ms: now,
expires_at_ms: None,
},
)
.await;
}
(
MessageRole::Assistant | MessageRole::Tool,
MessagePart::ToolInvocation {
tool,
args,
result,
error,
},
) => {
let now = crate::now_ms();
let tool_input = summarize_value(&args, 1200);
persist_global_memory_record(
state,
db,
GlobalMemoryRecord {
id: Uuid::new_v4().to_string(),
user_id: ctx.user_id.clone(),
source_type: "tool_input".to_string(),
content: format!("tool={} args={}", tool, tool_input),
content_hash: String::new(),
run_id: ctx.run_id.clone(),
session_id: Some(session_id.to_string()),
message_id: Some(message.id.clone()),
tool_name: Some(tool.clone()),
project_tag: session.project_id.clone(),
channel_tag: None,
host_tag: ctx.host_tag.clone(),
metadata: None,
provenance: Some(json!({
"origin_event_type": "session.run.finished",
"tenant_context": ctx.tenant_context,
})),
redaction_status: "passed".to_string(),
redaction_count: 0,
visibility: "private".to_string(),
demoted: false,
score_boost: 0.0,
created_at_ms: now,
updated_at_ms: now,
expires_at_ms: Some(now + 30 * 24 * 60 * 60 * 1000),
},
)
.await;
let tool_output = result
.as_ref()
.map(|v| summarize_value(v, 1500))
.or(error)
.unwrap_or_default();
if !tool_output.trim().is_empty() {
let now = crate::now_ms();
persist_global_memory_record(
state,
db,
GlobalMemoryRecord {
id: Uuid::new_v4().to_string(),
user_id: ctx.user_id.clone(),
source_type: "tool_output".to_string(),
content: format!("tool={} output={}", tool, tool_output),
content_hash: String::new(),
run_id: ctx.run_id.clone(),
session_id: Some(session_id.to_string()),
message_id: Some(message.id.clone()),
tool_name: Some(tool),
project_tag: session.project_id.clone(),
channel_tag: None,
host_tag: ctx.host_tag.clone(),
metadata: None,
provenance: Some(
json!({"origin_event_type": "session.run.finished"}),
),
redaction_status: "passed".to_string(),
redaction_count: 0,
visibility: "private".to_string(),
demoted: false,
score_boost: 0.0,
created_at_ms: now,
updated_at_ms: now,
expires_at_ms: Some(now + 30 * 24 * 60 * 60 * 1000),
},
)
.await;
}
}
_ => {}
}
}
}
}
pub(super) async fn ingest_event_memory_records(
state: &AppState,
db: &MemoryDatabase,
event: &EngineEvent,
ctx_by_session: &HashMap<String, RunMemoryContext>,
) {
let session_id = event_session_id(event);
let session_ctx = session_id
.as_ref()
.and_then(|sid| ctx_by_session.get(sid))
.cloned();
let run_id = event_run_id(event)
.or_else(|| session_ctx.as_ref().map(|c| c.run_id.clone()))
.unwrap_or_else(|| "unknown".to_string());
let user_id = session_ctx
.as_ref()
.map(|c| c.user_id.clone())
.unwrap_or_else(|| "default".to_string());
let host_tag = session_ctx.as_ref().and_then(|c| c.host_tag.clone());
let tenant_context = event_tenant_context(event)
.or_else(|| session_ctx.as_ref().map(|c| c.tenant_context.clone()))
.unwrap_or_default();
let (source_type, content, ttl_ms): (&str, String, Option<u64>) =
match event.event_type.as_str() {
"permission.asked" => (
"approval_request",
format!(
"permission requested tool={} query={}",
event
.properties
.get("tool")
.and_then(|v| v.as_str())
.unwrap_or("unknown"),
event
.properties
.get("query")
.and_then(|v| v.as_str())
.unwrap_or("")
),
Some(14 * 24 * 60 * 60 * 1000),
),
"permission.replied" => (
"approval_decision",
format!(
"permission reply requestID={} reply={}",
event
.properties
.get("requestID")
.and_then(|v| v.as_str())
.unwrap_or(""),
event
.properties
.get("reply")
.and_then(|v| v.as_str())
.unwrap_or(""),
),
Some(14 * 24 * 60 * 60 * 1000),
),
"mcp.auth.required" | "mcp.auth.pending" => (
"auth_challenge",
format!(
"mcp auth tool={} server={} status={} message={}",
event
.properties
.get("tool")
.and_then(|v| v.as_str())
.unwrap_or(""),
event
.properties
.get("server")
.and_then(|v| v.as_str())
.unwrap_or(""),
event.event_type,
event
.properties
.get("message")
.and_then(|v| v.as_str())
.unwrap_or(""),
),
Some(7 * 24 * 60 * 60 * 1000),
),
"todo.updated" => (
"plan_todos",
format!(
"todo updated: {}",
summarize_value(event.properties.get("todos").unwrap_or(&Value::Null), 1200)
),
Some(60 * 24 * 60 * 60 * 1000),
),
"question.asked" => (
"question_prompt",
format!(
"question asked: {}",
summarize_value(
event.properties.get("questions").unwrap_or(&Value::Null),
1200
)
),
Some(60 * 24 * 60 * 60 * 1000),
),
_ => return,
};
let now = crate::now_ms();
persist_global_memory_record(
state,
db,
GlobalMemoryRecord {
id: Uuid::new_v4().to_string(),
user_id,
source_type: source_type.to_string(),
content,
content_hash: String::new(),
run_id,
session_id,
message_id: event
.properties
.get("messageID")
.and_then(|v| v.as_str())
.map(ToString::to_string),
tool_name: event
.properties
.get("tool")
.and_then(|v| v.as_str())
.map(ToString::to_string),
project_tag: None,
channel_tag: event
.properties
.get("channel")
.and_then(|v| v.as_str())
.map(ToString::to_string),
host_tag,
metadata: None,
provenance: Some(json!({
"origin_event_type": event.event_type,
"tenant_context": tenant_context,
})),
redaction_status: "passed".to_string(),
redaction_count: 0,
visibility: "private".to_string(),
demoted: false,
score_boost: 0.0,
created_at_ms: now,
updated_at_ms: now,
expires_at_ms: ttl_ms.map(|ttl| now + ttl),
},
)
.await;
}
pub(super) async fn run_global_memory_ingestor(state: AppState) {
if !state.wait_until_ready_or_failed(120, 250).await {
tracing::warn!("global memory ingestor: skipped because runtime did not become ready");
return;
}
let mut rx = state.event_bus.subscribe();
let Some(db) = open_global_memory_db_for_state(&state).await else {
tracing::warn!("global memory ingestor disabled: could not open memory database");
return;
};
let mut by_session: HashMap<String, RunMemoryContext> = HashMap::new();
loop {
match rx.recv().await {
Ok(event) => match event.event_type.as_str() {
"session.run.started" => {
let session_id = event_session_id(&event);
let run_id = event_run_id(&event);
if let (Some(session_id), Some(run_id)) = (session_id, run_id) {
let started_at_ms = event
.properties
.get("startedAtMs")
.and_then(|v| v.as_u64())
.unwrap_or_else(crate::now_ms);
let user_id = event
.properties
.get("clientID")
.and_then(|v| v.as_str())
.filter(|v| !v.trim().is_empty())
.unwrap_or("default")
.to_string();
let host_tag = event
.properties
.get("environment")
.and_then(|v| v.get("os"))
.and_then(|v| v.as_str())
.map(ToString::to_string);
let tenant_context = event_tenant_context(&event).unwrap_or_default();
by_session.insert(
session_id,
RunMemoryContext {
run_id,
user_id,
started_at_ms,
host_tag,
tenant_context,
},
);
}
}
"session.run.finished" => {
if let Some(session_id) = event_session_id(&event) {
if let Some(ctx) = by_session.remove(&session_id) {
ingest_run_messages(&state, &db, &session_id, &ctx).await;
}
}
}
"permission.asked" | "permission.replied" | "mcp.auth.required"
| "mcp.auth.pending" | "todo.updated" | "question.asked" => {
ingest_event_memory_records(&state, &db, &event, &by_session).await;
}
_ => {}
},
Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
}
}
}
pub(super) async fn memory_put(
State(state): State<AppState>,
Extension(tenant_context): Extension<TenantContext>,
Json(input): Json<MemoryPutInput>,
) -> Result<Json<MemoryPutResponse>, StatusCode> {
let response =
memory_put_impl(&state, &tenant_context, input.request, input.capability).await?;
Ok(Json(response))
}
pub(super) async fn memory_import(
State(state): State<AppState>,
Extension(tenant_context): Extension<TenantContext>,
Extension(request_principal): Extension<RequestPrincipal>,
verified_tenant_context: Option<Extension<VerifiedTenantContext>>,
Json(input): Json<MemoryImportInput>,
) -> Result<Json<MemoryImportResponse>, (StatusCode, Json<ErrorEnvelope>)> {
let source_kind = input.source.kind.trim().to_ascii_lowercase();
if source_kind != "path" {
return Err(skill_error(
StatusCode::BAD_REQUEST,
"source.kind must be `path`",
));
}
let path = input.source.path.trim().to_string();
if path.is_empty() {
return Err(skill_error(
StatusCode::BAD_REQUEST,
"source.path is required for path imports",
));
}
validate_memory_import_path(&path)?;
let project_id = normalize_optional_memory_import_id(input.project_id);
let session_id = normalize_optional_memory_import_id(input.session_id);
let source_binding_id = normalize_optional_memory_import_id(input.source_binding_id);
match input.tier {
MemoryTier::Project if project_id.is_none() => {
return Err(skill_error(
StatusCode::BAD_REQUEST,
"tier=project requires project_id",
));
}
MemoryTier::Session if session_id.is_none() => {
return Err(skill_error(
StatusCode::BAD_REQUEST,
"tier=session requires session_id",
));
}
_ => {}
}
let source_binding = resolve_memory_import_source_binding(
&state,
&tenant_context,
&request_principal,
verified_tenant_context.as_deref(),
source_binding_id.as_deref(),
)
.await?;
let source_binding_for_job = source_binding.clone();
let job_started_at_ms = crate::util::time::now_ms();
let ingestion_job_id = source_binding_for_job.as_ref().map(|binding| {
format!(
"manual-import-{}-{}",
job_started_at_ms,
uuid::Uuid::new_v4()
)
});
if let (Some(job_id), Some(binding)) = (&ingestion_job_id, source_binding_for_job.as_ref()) {
if let Err(err) = record_enterprise_ingestion_job(
&state,
IngestionJob {
job_id: job_id.clone(),
tenant_context: tenant_context.clone(),
connector_id: binding.connector_id.clone(),
binding_id: binding.binding_id.clone(),
state: IngestionJobState::Running,
source_object_ids: Vec::new(),
started_at_ms: Some(job_started_at_ms),
finished_at_ms: None,
quarantine_id: None,
},
)
.await
{
tracing::warn!(
error = %err,
"failed to record enterprise ingestion job start"
);
}
}
publish_tenant_event(
&state,
&tenant_context,
"memory.import.started",
json!({
"source": {"kind": "path", "path": path},
"format": input.format,
"tier": input.tier,
"project_id": project_id.clone(),
"session_id": session_id.clone(),
"source_binding_id": source_binding_id.clone(),
"sync_deletes": input.sync_deletes,
}),
);
let Some(manager) = open_memory_manager_for_state(&state).await else {
if let (Some(job_id), Some(binding)) = (&ingestion_job_id, source_binding_for_job.as_ref())
{
if let Err(err) = record_enterprise_ingestion_job(
&state,
IngestionJob {
job_id: job_id.clone(),
tenant_context: tenant_context.clone(),
connector_id: binding.connector_id.clone(),
binding_id: binding.binding_id.clone(),
state: IngestionJobState::Failed,
source_object_ids: Vec::new(),
started_at_ms: Some(job_started_at_ms),
finished_at_ms: Some(crate::util::time::now_ms()),
quarantine_id: None,
},
)
.await
{
tracing::warn!(
error = %err,
"failed to record enterprise ingestion job failure"
);
}
}
publish_tenant_event(
&state,
&tenant_context,
"memory.import.failed",
json!({
"source": {"kind": "path", "path": path},
"format": input.format,
"tier": input.tier,
"source_binding_id": source_binding_id.clone(),
"error": "failed to open memory manager",
}),
);
return Err(skill_error(
StatusCode::INTERNAL_SERVER_ERROR,
"failed to open memory manager",
));
};
let request = TandemMemoryImportRequest {
root_path: path.clone(),
format: input.format,
tier: input.tier,
session_id: session_id.clone(),
project_id: project_id.clone(),
tenant_scope: MemoryTenantScope {
org_id: tenant_context.org_id.clone(),
workspace_id: tenant_context.workspace_id.clone(),
deployment_id: tenant_context.deployment_id.clone(),
},
source_binding,
sync_deletes: input.sync_deletes,
import_namespace: None,
};
let stats = match import_files(&manager, &request, None::<fn(&MemoryImportProgress)>).await {
Ok(stats) => stats,
Err(err) => {
if let (Some(job_id), Some(binding)) =
(&ingestion_job_id, source_binding_for_job.as_ref())
{
if let Err(record_err) = record_enterprise_ingestion_job(
&state,
IngestionJob {
job_id: job_id.clone(),
tenant_context: tenant_context.clone(),
connector_id: binding.connector_id.clone(),
binding_id: binding.binding_id.clone(),
state: IngestionJobState::Failed,
source_object_ids: Vec::new(),
started_at_ms: Some(job_started_at_ms),
finished_at_ms: Some(crate::util::time::now_ms()),
quarantine_id: None,
},
)
.await
{
tracing::warn!(
error = %record_err,
"failed to record enterprise ingestion job failure"
);
}
}
publish_tenant_event(
&state,
&tenant_context,
"memory.import.failed",
json!({
"source": {"kind": "path", "path": path},
"format": input.format,
"tier": input.tier,
"project_id": project_id.clone(),
"session_id": session_id.clone(),
"source_binding_id": source_binding_id.clone(),
"sync_deletes": input.sync_deletes,
"error": err.to_string(),
}),
);
return Err(skill_error(
StatusCode::INTERNAL_SERVER_ERROR,
format!("memory import failed: {err}"),
));
}
};
if let (Some(job_id), Some(binding)) = (&ingestion_job_id, source_binding_for_job.as_ref()) {
let source_objects = source_objects_seen_since(
&manager,
&request.tenant_scope,
&binding.binding_id,
job_started_at_ms,
)
.await
.unwrap_or_default();
let source_object_ids = source_objects
.iter()
.map(|record| record.source_object_id.clone())
.collect::<Vec<_>>();
let quarantine_id = if binding.require_review {
let quarantine_id =
format!("quarantine-{}-{}", job_started_at_ms, uuid::Uuid::new_v4());
if let Err(err) = quarantine_source_bound_import(
&manager,
&request.tenant_scope,
&binding.binding_id,
&source_objects,
job_started_at_ms,
)
.await
{
tracing::warn!(
error = %err,
"failed to quarantine enterprise source-bound import output"
);
}
if let Err(err) = record_enterprise_ingestion_quarantine(
&state,
IngestionQuarantine {
quarantine_id: quarantine_id.clone(),
tenant_context: tenant_context.clone(),
connector_id: binding.connector_id.clone(),
binding_id: binding.binding_id.clone(),
source_object_ids: source_object_ids.clone(),
reason: "source binding requires ingestion review".to_string(),
created_at_ms: crate::util::time::now_ms(),
reviewed_by: None,
reviewed_at_ms: None,
disposition: None,
},
)
.await
{
tracing::warn!(
error = %err,
"failed to record enterprise ingestion quarantine"
);
}
Some(quarantine_id)
} else {
None
};
let job_state = if binding.require_review {
IngestionJobState::Quarantined
} else if stats.errors > 0 {
IngestionJobState::Failed
} else {
IngestionJobState::Completed
};
if let Err(err) = record_enterprise_ingestion_job(
&state,
IngestionJob {
job_id: job_id.clone(),
tenant_context: tenant_context.clone(),
connector_id: binding.connector_id.clone(),
binding_id: binding.binding_id.clone(),
state: job_state,
source_object_ids,
started_at_ms: Some(job_started_at_ms),
finished_at_ms: Some(crate::util::time::now_ms()),
quarantine_id,
},
)
.await
{
tracing::warn!(
error = %err,
"failed to record enterprise ingestion job completion"
);
}
}
publish_tenant_event(
&state,
&tenant_context,
"memory.import.succeeded",
json!({
"source": {"kind": "path", "path": path},
"format": input.format,
"tier": input.tier,
"project_id": project_id.clone(),
"session_id": session_id.clone(),
"source_binding_id": source_binding_id.clone(),
"sync_deletes": input.sync_deletes,
"stats": {
"discovered_files": stats.discovered_files,
"files_processed": stats.files_processed,
"indexed_files": stats.indexed_files,
"skipped_files": stats.skipped_files,
"deleted_files": stats.deleted_files,
"chunks_created": stats.chunks_created,
"errors": stats.errors,
},
}),
);
Ok(Json(memory_import_response(
path,
input.format,
input.tier,
project_id,
session_id,
source_binding_id,
input.sync_deletes,
stats,
)))
}
async fn resolve_memory_import_source_binding(
state: &AppState,
tenant_context: &TenantContext,
request_principal: &RequestPrincipal,
verified_tenant_context: Option<&VerifiedTenantContext>,
source_binding_id: Option<&str>,
) -> Result<Option<MemoryImportSourceBinding>, (StatusCode, Json<ErrorEnvelope>)> {
let Some(source_binding_id) = source_binding_id else {
if memory_import_requires_source_binding(
tenant_context,
request_principal,
verified_tenant_context,
) {
return Err(skill_error(
StatusCode::BAD_REQUEST,
"hosted/enterprise memory imports require source_binding_id",
));
}
return Ok(None);
};
if source_binding_id == DEFAULT_LOCAL_MANUAL_SOURCE_BINDING_ID
&& !memory_import_requires_source_binding(
tenant_context,
request_principal,
verified_tenant_context,
)
{
return Ok(Some(default_local_manual_source_binding(tenant_context)));
}
let registry = state.enterprise_source_bindings.read().await;
let Some(binding) = registry.values().find(|binding| {
binding.binding_id == source_binding_id && binding.tenant_matches(tenant_context)
}) else {
return Err(skill_error(
StatusCode::BAD_REQUEST,
"source_binding_id does not reference an enabled binding for this tenant",
));
};
if !binding.state.allows_ingestion() || !binding.ingestion_policy.allow_indexing {
return Err(skill_error(
StatusCode::BAD_REQUEST,
"source binding does not allow memory import indexing",
));
}
let registry = state.enterprise_connectors.read().await;
let Some(connector) = registry.values().find(|connector| {
connector.connector_id == binding.connector_id && connector.tenant_matches(tenant_context)
}) else {
return Err(skill_error(
StatusCode::BAD_REQUEST,
"source binding connector is not registered for this tenant",
));
};
if !connector.state.allows_ingestion() {
return Err(skill_error(
StatusCode::BAD_REQUEST,
format!(
"source binding connector does not allow memory import indexing: {}",
connector_lifecycle_state_label(connector.state)
),
));
}
Ok(Some(MemoryImportSourceBinding {
binding_id: binding.binding_id.clone(),
connector_id: binding.connector_id.clone(),
resource_ref: serde_json::to_value(&binding.resource_ref).map_err(|_| {
skill_error(
StatusCode::INTERNAL_SERVER_ERROR,
"failed to serialize source binding resource scope",
)
})?,
data_class: serde_json::to_value(binding.data_class)
.ok()
.and_then(|value| value.as_str().map(ToOwned::to_owned))
.unwrap_or_else(|| format!("{:?}", binding.data_class)),
require_review: binding.ingestion_policy.require_review,
}))
}
const DEFAULT_LOCAL_MANUAL_SOURCE_BINDING_ID: &str = "local_manual_upload";
fn default_local_manual_source_binding(
tenant_context: &TenantContext,
) -> MemoryImportSourceBinding {
MemoryImportSourceBinding {
binding_id: DEFAULT_LOCAL_MANUAL_SOURCE_BINDING_ID.to_string(),
connector_id: "manual_upload".to_string(),
resource_ref: json!({
"organization_id": tenant_context.org_id.clone(),
"workspace_id": tenant_context.workspace_id.clone(),
"resource_kind": "document_collection",
"resource_id": "local-manual-uploads",
}),
data_class: "internal".to_string(),
require_review: false,
}
}
async fn record_enterprise_ingestion_job(
state: &AppState,
job: IngestionJob,
) -> Result<(), std::io::Error> {
let mut registry = state.enterprise_ingestion_jobs.write().await;
let key = enterprise_ingestion_job_key(&job);
registry.insert(key, job);
persist_enterprise_ingestion_jobs(&state.enterprise_ingestion_jobs_path, ®istry).await
}
async fn record_enterprise_ingestion_quarantine(
state: &AppState,
quarantine: IngestionQuarantine,
) -> Result<(), std::io::Error> {
let mut registry = state.enterprise_ingestion_quarantines.write().await;
let key = enterprise_ingestion_quarantine_key(&quarantine);
registry.insert(key, quarantine);
persist_enterprise_ingestion_quarantines(
&state.enterprise_ingestion_quarantines_path,
®istry,
)
.await
}
fn enterprise_ingestion_job_key(job: &IngestionJob) -> String {
let deployment = job
.tenant_context
.deployment_id
.as_deref()
.unwrap_or("local");
format!(
"{}::{}::{}::{}",
job.tenant_context.org_id, job.tenant_context.workspace_id, deployment, job.job_id
)
}
fn enterprise_ingestion_quarantine_key(quarantine: &IngestionQuarantine) -> String {
let deployment = quarantine
.tenant_context
.deployment_id
.as_deref()
.unwrap_or("local");
format!(
"{}::{}::{}::{}",
quarantine.tenant_context.org_id,
quarantine.tenant_context.workspace_id,
deployment,
quarantine.quarantine_id
)
}
async fn persist_enterprise_ingestion_jobs(
path: &std::path::Path,
registry: &std::collections::HashMap<String, IngestionJob>,
) -> Result<(), std::io::Error> {
if let Some(parent) = path.parent() {
tokio::fs::create_dir_all(parent).await?;
}
let payload = serde_json::to_vec_pretty(registry).map_err(std::io::Error::other)?;
tokio::fs::write(path, payload).await
}
async fn persist_enterprise_ingestion_quarantines(
path: &std::path::Path,
registry: &std::collections::HashMap<String, IngestionQuarantine>,
) -> Result<(), std::io::Error> {
if let Some(parent) = path.parent() {
tokio::fs::create_dir_all(parent).await?;
}
let payload = serde_json::to_vec_pretty(registry).map_err(std::io::Error::other)?;
tokio::fs::write(path, payload).await
}
async fn source_objects_seen_since(
manager: &tandem_memory::MemoryManager,
tenant_scope: &MemoryTenantScope,
binding_id: &str,
started_at_ms: u64,
) -> Result<Vec<SourceObjectLifecycleRecord>, tandem_memory::types::MemoryError> {
let mut records: Vec<_> = manager
.db()
.list_source_object_lifecycle_for_binding_for_tenant(tenant_scope, binding_id)
.await?
.into_iter()
.filter(|record| record.last_seen_at_ms >= started_at_ms)
.collect();
records.sort_by(|left, right| left.source_object_id.cmp(&right.source_object_id));
records.dedup_by(|left, right| left.source_object_id == right.source_object_id);
Ok(records)
}
async fn quarantine_source_bound_import(
manager: &tandem_memory::MemoryManager,
tenant_scope: &MemoryTenantScope,
binding_id: &str,
source_objects: &[SourceObjectLifecycleRecord],
changed_at_ms: u64,
) -> Result<(), tandem_memory::types::MemoryError> {
for record in source_objects {
manager
.db()
.delete_file_chunks_by_path_for_tenant(
record.tier,
record.session_id.as_deref(),
record.project_id.as_deref(),
&record.indexed_path,
tenant_scope,
)
.await?;
manager
.db()
.delete_import_index_entry_for_tenant(
record.tier,
record.session_id.as_deref(),
record.project_id.as_deref(),
&record.indexed_path,
tenant_scope,
)
.await?;
manager
.db()
.mark_source_object_lifecycle_state_for_tenant(
tenant_scope,
binding_id,
&record.source_object_id,
SourceObjectLifecycleState::Quarantined,
changed_at_ms,
)
.await?;
}
Ok(())
}
fn connector_lifecycle_state_label(state: ConnectorLifecycleState) -> &'static str {
match state {
ConnectorLifecycleState::Active => "active",
ConnectorLifecycleState::Paused => "paused",
ConnectorLifecycleState::Revoked => "revoked",
ConnectorLifecycleState::Quarantined => "quarantined",
}
}
fn memory_import_requires_source_binding(
tenant_context: &TenantContext,
request_principal: &RequestPrincipal,
verified_tenant_context: Option<&VerifiedTenantContext>,
) -> bool {
verified_tenant_context.is_some()
|| tenant_context.deployment_id.is_some()
|| request_principal.source == "tandem-web"
}
fn memory_import_response(
path: String,
format: MemoryImportFormat,
tier: MemoryTier,
project_id: Option<String>,
session_id: Option<String>,
source_binding_id: Option<String>,
sync_deletes: bool,
stats: MemoryImportStats,
) -> MemoryImportResponse {
MemoryImportResponse {
ok: true,
source: MemoryImportPathSourceResponse { kind: "path", path },
format,
tier,
project_id,
session_id,
source_binding_id,
sync_deletes,
discovered_files: stats.discovered_files,
files_processed: stats.files_processed,
indexed_files: stats.indexed_files,
skipped_files: stats.skipped_files,
deleted_files: stats.deleted_files,
chunks_created: stats.chunks_created,
errors: stats.errors,
}
}
fn normalize_optional_memory_import_id(value: Option<String>) -> Option<String> {
value
.map(|raw| raw.trim().to_string())
.filter(|trimmed| !trimmed.is_empty())
}
fn validate_memory_import_path(path: &str) -> Result<(), (StatusCode, Json<ErrorEnvelope>)> {
let metadata = std::fs::metadata(path).map_err(|err| {
skill_error(
StatusCode::BAD_REQUEST,
format!("source.path must exist and be readable: {err}"),
)
})?;
let readable = if metadata.is_dir() {
std::fs::read_dir(path).map(|_| ())
} else {
std::fs::File::open(path).map(|_| ())
};
readable.map_err(|err| {
skill_error(
StatusCode::BAD_REQUEST,
format!("source.path must be readable: {err}"),
)
})
}
pub(super) async fn memory_put_impl(
state: &AppState,
tenant_context: &TenantContext,
request: MemoryPutRequest,
capability: Option<MemoryCapabilityToken>,
) -> Result<MemoryPutResponse, StatusCode> {
let capability =
validate_memory_put_capability_with_guardrail(state, tenant_context, &request, capability)
.await?;
if !capability
.memory
.write_tiers
.contains(&request.partition.tier)
{
emit_blocked_memory_put_guardrail(
state,
tenant_context,
&request,
capability.subject.clone(),
"write tier not allowed by capability",
)
.await?;
return Err(StatusCode::FORBIDDEN);
}
let id = Uuid::new_v4().to_string();
let partition_key = request.partition.key();
let kind = memory_kind_for_request(request.kind.clone());
let now = crate::now_ms();
let audit_id = Uuid::new_v4().to_string();
let db = open_global_memory_db_for_state(state)
.await
.ok_or(StatusCode::INTERNAL_SERVER_ERROR)?;
let artifact_refs = request.artifact_refs.clone();
let artifact_ref_labels = artifact_refs
.iter()
.map(String::as_str)
.collect::<Vec<_>>()
.join(",");
let source_type = match request.kind {
tandem_memory::MemoryContentKind::SolutionCapsule => "solution_capsule",
tandem_memory::MemoryContentKind::Note => "note",
tandem_memory::MemoryContentKind::Fact => "fact",
}
.to_string();
let user_id = capability.subject.clone();
let record = GlobalMemoryRecord {
id: id.clone(),
user_id,
source_type,
content: request.content.clone(),
content_hash: String::new(),
run_id: request.run_id.clone(),
session_id: None,
message_id: None,
tool_name: None,
project_tag: Some(request.partition.project_id.clone()),
channel_tag: None,
host_tag: None,
metadata: memory_metadata_with_storage_fields(
request.metadata.clone(),
&artifact_refs,
request.classification,
),
provenance: Some(memory_put_provenance(
&request,
&partition_key,
&artifact_refs,
tenant_context,
)),
redaction_status: "passed".to_string(),
redaction_count: 0,
visibility: "private".to_string(),
demoted: false,
score_boost: 0.0,
created_at_ms: now,
updated_at_ms: now,
expires_at_ms: None,
};
let memory_linkage_value = memory_linkage_from_parts(
&request.run_id,
Some(&request.partition.project_id),
record.metadata.as_ref(),
record.provenance.as_ref(),
);
let put_detail = format!(
"kind={} classification={} artifact_refs={} visibility=private tier={} partition_key={}{}",
kind,
memory_classification_label(record.metadata.as_ref()),
artifact_ref_labels,
request.partition.tier,
partition_key,
memory_linkage_detail(&memory_linkage_value)
);
persist_global_memory_record(&state, &db, record).await;
append_memory_audit(
&state,
tenant_context,
crate::MemoryAuditEvent {
audit_id: audit_id.clone(),
action: "memory_put".to_string(),
run_id: request.run_id.clone(),
tenant_context: tenant_context.clone(),
memory_id: Some(id.clone()),
source_memory_id: None,
to_tier: Some(request.partition.tier),
partition_key: partition_key.clone(),
actor: capability.subject,
status: "ok".to_string(),
detail: Some(put_detail),
created_at_ms: now,
},
)
.await?;
publish_tenant_event(
state,
tenant_context,
"memory.put",
json!({
"runID": request.run_id,
"memoryID": id,
"kind": kind,
"classification": request.classification,
"artifactRefs": artifact_refs,
"visibility": "private",
"tier": request.partition.tier,
"partitionKey": partition_key,
"linkage": memory_linkage_value.clone(),
"auditID": audit_id,
}),
);
publish_tenant_event(
state,
tenant_context,
"memory.updated",
json!({
"memoryID": id,
"runID": request.run_id,
"action": "put",
"kind": kind,
"classification": request.classification,
"artifactRefs": artifact_refs,
"visibility": "private",
"tier": request.partition.tier,
"partitionKey": partition_key,
"linkage": memory_linkage_value,
"auditID": audit_id,
}),
);
Ok(MemoryPutResponse {
id,
stored: true,
tier: request.partition.tier,
partition_key,
audit_id,
})
}
pub(super) async fn memory_promote(
State(state): State<AppState>,
Extension(tenant_context): Extension<TenantContext>,
Json(input): Json<MemoryPromoteInput>,
) -> Result<Json<MemoryPromoteResponse>, StatusCode> {
let response =
memory_promote_impl(&state, &tenant_context, input.request, input.capability).await?;
Ok(Json(response))
}
pub(super) async fn memory_promote_impl(
state: &AppState,
tenant_context: &TenantContext,
request: MemoryPromoteRequest,
capability: Option<MemoryCapabilityToken>,
) -> Result<MemoryPromoteResponse, StatusCode> {
let source_memory_id = request.source_memory_id.clone();
let capability = validate_memory_promote_capability_with_guardrail(
state,
tenant_context,
&request,
capability,
)
.await?;
if !capability.memory.promote_targets.contains(&request.to_tier) {
emit_blocked_memory_promote_guardrail(
state,
tenant_context,
&request,
capability.subject.clone(),
"promotion target not allowed by capability",
)
.await?;
return Err(StatusCode::FORBIDDEN);
}
if capability.memory.require_review_for_promote
&& (request.review.approval_id.is_none() || request.review.reviewer_id.is_none())
{
emit_blocked_memory_promote_guardrail(
state,
tenant_context,
&request,
capability.subject.clone(),
"review approval required for promote",
)
.await?;
return Err(StatusCode::FORBIDDEN);
}
let db = open_global_memory_db_for_state(state)
.await
.ok_or(StatusCode::INTERNAL_SERVER_ERROR)?;
let source = db
.get_global_memory_for_tenant(
&request.source_memory_id,
&tenant_context.org_id,
&tenant_context.workspace_id,
tenant_context.deployment_id.as_deref(),
)
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
let Some(source) = source else {
let scrub_report = ScrubReport {
status: ScrubStatus::Blocked,
redactions: 0,
block_reason: Some("source memory missing or previously blocked".to_string()),
};
let audit_id = Uuid::new_v4().to_string();
let partition_key = format!(
"{}/{}/{}/{}",
request.partition.org_id,
request.partition.workspace_id,
request.partition.project_id,
request.to_tier
);
let linkage = json!({
"run_id": request.run_id,
"project_id": request.partition.project_id,
"origin_event_type": Value::Null,
"origin_run_id": request.run_id,
"origin_session_id": Value::Null,
"origin_message_id": Value::Null,
"partition_key": partition_key,
"promote_run_id": Value::Null,
"approval_id": request.review.approval_id,
"artifact_refs": [],
});
append_memory_audit(
&state,
tenant_context,
crate::MemoryAuditEvent {
audit_id: audit_id.clone(),
action: "memory_promote".to_string(),
run_id: request.run_id.clone(),
tenant_context: tenant_context.clone(),
memory_id: None,
source_memory_id: Some(source_memory_id.clone()),
to_tier: Some(request.to_tier),
partition_key: partition_key.clone(),
actor: capability.subject,
status: "blocked".to_string(),
detail: scrub_report
.block_reason
.as_ref()
.map(|detail| format!("{detail}{}", memory_linkage_detail(&linkage))),
created_at_ms: crate::now_ms(),
},
)
.await?;
publish_tenant_event(
state,
tenant_context,
"memory.promote",
json!({
"runID": request.run_id,
"sourceMemoryID": source_memory_id,
"toTier": request.to_tier,
"partitionKey": partition_key,
"status": "blocked",
"kind": Value::Null,
"classification": Value::Null,
"artifactRefs": [],
"visibility": Value::Null,
"scrubStatus": scrub_report.status,
"linkage": linkage,
"detail": scrub_report.block_reason.clone(),
"auditID": audit_id,
}),
);
return Ok(MemoryPromoteResponse {
promoted: false,
new_memory_id: None,
to_tier: request.to_tier,
scrub_report,
audit_id,
});
};
let scrub_report = scrub_content(&source.content);
let audit_id = Uuid::new_v4().to_string();
let now = crate::now_ms();
let partition_key = format!(
"{}/{}/{}/{}",
request.partition.org_id,
request.partition.workspace_id,
request.partition.project_id,
request.to_tier
);
let linkage = memory_linkage(&source);
if scrub_report.status == ScrubStatus::Blocked {
append_memory_audit(
&state,
tenant_context,
crate::MemoryAuditEvent {
audit_id: audit_id.clone(),
action: "memory_promote".to_string(),
run_id: request.run_id.clone(),
tenant_context: tenant_context.clone(),
memory_id: None,
source_memory_id: Some(source_memory_id.clone()),
to_tier: Some(request.to_tier),
partition_key: partition_key.clone(),
actor: capability.subject,
status: "blocked".to_string(),
detail: scrub_report
.block_reason
.as_ref()
.map(|detail| format!("{detail}{}", memory_linkage_detail(&linkage))),
created_at_ms: now,
},
)
.await?;
publish_tenant_event(
state,
tenant_context,
"memory.promote",
json!({
"runID": request.run_id,
"sourceMemoryID": source_memory_id,
"toTier": request.to_tier,
"partitionKey": partition_key,
"status": "blocked",
"kind": memory_kind_label(&source.source_type),
"classification": memory_classification_label(source.metadata.as_ref()),
"artifactRefs": memory_artifact_refs(source.metadata.as_ref()),
"visibility": source.visibility,
"scrubStatus": scrub_report.status,
"linkage": linkage,
"detail": scrub_report.block_reason.clone(),
"auditID": audit_id,
}),
);
return Ok(MemoryPromoteResponse {
promoted: false,
new_memory_id: None,
to_tier: request.to_tier,
scrub_report,
audit_id,
});
}
let new_id = source.id.clone();
let next_metadata = memory_promote_metadata(source.metadata.as_ref(), &request, now);
let next_provenance = memory_promote_provenance(
source.provenance.as_ref(),
&request,
&partition_key,
now,
tenant_context,
);
let classification = memory_classification_label(next_metadata.as_ref());
let artifact_refs = memory_artifact_refs(next_metadata.as_ref());
let artifact_ref_labels = artifact_refs
.iter()
.map(String::as_str)
.collect::<Vec<_>>()
.join(",");
let kind = memory_kind_label(&source.source_type);
let promote_detail = format!(
"kind={} classification={} artifact_refs={} visibility=shared tier={} partition_key={} source_memory_id={} approval_id={}{}",
kind,
classification,
artifact_ref_labels,
request.to_tier,
partition_key,
source_memory_id,
request.review.approval_id.clone().unwrap_or_default(),
memory_linkage_detail(&memory_linkage_from_parts(
&source.run_id,
source.project_tag.as_deref(),
next_metadata.as_ref(),
Some(&next_provenance),
))
);
db.update_global_memory_context_for_tenant(
&new_id,
&tenant_context.org_id,
&tenant_context.workspace_id,
tenant_context.deployment_id.as_deref(),
"shared",
false,
next_metadata.as_ref(),
Some(&next_provenance),
)
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
append_memory_audit(
&state,
tenant_context,
crate::MemoryAuditEvent {
audit_id: audit_id.clone(),
action: "memory_promote".to_string(),
run_id: request.run_id.clone(),
tenant_context: tenant_context.clone(),
memory_id: Some(new_id.clone()),
source_memory_id: Some(source_memory_id.clone()),
to_tier: Some(request.to_tier),
partition_key: format!(
"{}/{}/{}/{}",
request.partition.org_id,
request.partition.workspace_id,
request.partition.project_id,
request.to_tier
),
actor: capability.subject,
status: "ok".to_string(),
detail: Some(promote_detail),
created_at_ms: now,
},
)
.await?;
publish_tenant_event(
state,
tenant_context,
"memory.promote",
json!({
"runID": request.run_id,
"sourceMemoryID": source_memory_id,
"memoryID": new_id,
"kind": kind,
"classification": classification,
"artifactRefs": artifact_refs,
"visibility": "shared",
"toTier": request.to_tier,
"partitionKey": partition_key,
"linkage": memory_linkage_from_parts(
&source.run_id,
source.project_tag.as_deref(),
next_metadata.as_ref(),
Some(&next_provenance),
),
"approvalID": request.review.approval_id,
"auditID": audit_id,
"scrubStatus": scrub_report.status,
}),
);
publish_tenant_event(
state,
tenant_context,
"memory.updated",
json!({
"memoryID": new_id,
"runID": request.run_id,
"action": "promote",
"kind": kind,
"classification": classification,
"artifactRefs": artifact_refs,
"visibility": "shared",
"tier": request.to_tier,
"partitionKey": partition_key,
"linkage": memory_linkage_from_parts(
&source.run_id,
source.project_tag.as_deref(),
next_metadata.as_ref(),
Some(&next_provenance),
),
"sourceMemoryID": source_memory_id,
"approvalID": request.review.approval_id,
"auditID": audit_id,
}),
);
Ok(MemoryPromoteResponse {
promoted: true,
new_memory_id: Some(new_id),
to_tier: request.to_tier,
scrub_report,
audit_id,
})
}
pub(super) async fn memory_search(
State(state): State<AppState>,
Extension(tenant_context): Extension<TenantContext>,
verified_tenant_context: Option<Extension<VerifiedTenantContext>>,
Json(input): Json<MemorySearchInput>,
) -> Result<Json<MemorySearchResponse>, StatusCode> {
let request = input.request;
let capability = validate_memory_search_capability_with_guardrail(
&state,
&tenant_context,
&request,
input.capability,
)
.await?;
let requested_scopes = if request.read_scopes.is_empty() {
capability.memory.read_tiers.clone()
} else {
request.read_scopes.clone()
};
let mut scopes_used = Vec::new();
let mut blocked_scopes = Vec::new();
for scope in &requested_scopes {
if capability.memory.read_tiers.contains(scope) {
scopes_used.push(scope.clone());
} else {
blocked_scopes.push(scope.clone());
}
}
let allow_private_results = scopes_used
.iter()
.any(|scope| matches!(scope, tandem_memory::GovernedMemoryTier::Session));
let limit = request.limit.unwrap_or(8).clamp(1, 100);
let source_access_filter = verified_tenant_context
.as_ref()
.and_then(|context| context.strict_projection.clone())
.map(|strict_context| MemoryAccessFilter::strict(strict_context, crate::now_ms()));
let hits = if scopes_used.is_empty() {
Vec::new()
} else {
let db = open_global_memory_db_for_state(&state)
.await
.ok_or(StatusCode::INTERNAL_SERVER_ERROR)?;
db.search_global_memory_for_tenant(
&tenant_context.org_id,
&tenant_context.workspace_id,
tenant_context.deployment_id.as_deref(),
&capability.subject,
&request.query,
limit,
Some(&request.partition.project_id),
None,
None,
)
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
.into_iter()
.filter(|hit| allow_private_results || hit.record.visibility.eq_ignore_ascii_case("shared"))
.filter(|hit| {
global_memory_record_visible_to_access_filter(
&hit.record,
source_access_filter.as_ref(),
)
})
.collect::<Vec<_>>()
};
let results = hits
.into_iter()
.map(|hit| {
json!({
"id": hit.record.id,
"tier": memory_tier_for_visibility(&hit.record.visibility),
"classification": memory_classification_label(hit.record.metadata.as_ref()),
"kind": memory_kind_label(&hit.record.source_type),
"source_type": hit.record.source_type,
"created_at_ms": hit.record.created_at_ms,
"content": hit.record.content,
"score": hit.score,
"run_id": hit.record.run_id,
"visibility": hit.record.visibility,
"artifact_refs": memory_artifact_refs(hit.record.metadata.as_ref()),
"linkage": memory_linkage(&hit.record),
"metadata": hit.record.metadata,
"provenance": hit.record.provenance,
})
})
.collect::<Vec<_>>();
let result_ids = results
.iter()
.filter_map(|row| row.get("id").and_then(Value::as_str))
.map(ToString::to_string)
.collect::<Vec<_>>();
let result_kinds = results
.iter()
.filter_map(|row| row.get("kind").and_then(Value::as_str))
.map(ToString::to_string)
.collect::<Vec<_>>();
let linkage = json!({
"run_id": request.run_id,
"project_id": request.partition.project_id,
"origin_event_type": "memory.search",
"origin_run_id": request.run_id,
"origin_session_id": Value::Null,
"origin_message_id": Value::Null,
"partition_key": request.partition.key(),
"promote_run_id": Value::Null,
"approval_id": Value::Null,
"artifact_refs": [],
});
let audit_id = Uuid::new_v4().to_string();
let now = crate::now_ms();
let search_status = if scopes_used.is_empty() && !blocked_scopes.is_empty() {
"blocked"
} else {
"ok"
};
let search_detail = format!(
"query={} result_count={} result_ids={} result_kinds={} requested_scopes={} scopes_used={} blocked_scopes={}{}",
request.query,
results.len(),
result_ids.join(","),
result_kinds.join(","),
requested_scopes
.iter()
.map(|scope| scope.to_string())
.collect::<Vec<_>>()
.join(","),
scopes_used
.iter()
.map(|scope| scope.to_string())
.collect::<Vec<_>>()
.join(","),
blocked_scopes
.iter()
.map(|scope| scope.to_string())
.collect::<Vec<_>>()
.join(","),
memory_linkage_detail(&linkage)
);
append_memory_audit(
&state,
&tenant_context,
crate::MemoryAuditEvent {
audit_id: audit_id.clone(),
action: "memory_search".to_string(),
run_id: request.run_id.clone(),
tenant_context: tenant_context.clone(),
memory_id: None,
source_memory_id: None,
to_tier: None,
partition_key: request.partition.key(),
actor: capability.subject,
status: search_status.to_string(),
detail: Some(search_detail),
created_at_ms: now,
},
)
.await?;
publish_tenant_event(
&state,
&tenant_context,
"memory.search",
json!({
"runID": request.run_id,
"query": request.query,
"partitionKey": request.partition.key(),
"resultCount": results.len(),
"resultIDs": result_ids,
"resultKinds": result_kinds,
"requestedScopes": requested_scopes,
"scopesUsed": scopes_used.clone(),
"blockedScopes": blocked_scopes.clone(),
"linkage": linkage,
"status": search_status,
"auditID": audit_id,
}),
);
Ok(Json(MemorySearchResponse {
results,
scopes_used,
blocked_scopes,
audit_id,
}))
}
fn global_memory_record_visible_to_access_filter(
record: &GlobalMemoryRecord,
access_filter: Option<&MemoryAccessFilter>,
) -> bool {
let Some(target) = MemorySourceAccessTarget::from_metadata(record.metadata.as_ref()) else {
return true;
};
access_filter
.map(|filter| filter.allows_source_target(&target))
.unwrap_or(false)
}
pub(super) async fn memory_demote(
State(state): State<AppState>,
Extension(tenant_context): Extension<TenantContext>,
Json(input): Json<MemoryDemoteInput>,
) -> Result<Json<Value>, StatusCode> {
let db = open_global_memory_db_for_state(&state)
.await
.ok_or(StatusCode::INTERNAL_SERVER_ERROR)?;
let record = db
.get_global_memory_for_tenant(
&input.id,
&tenant_context.org_id,
&tenant_context.workspace_id,
tenant_context.deployment_id.as_deref(),
)
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
let Some(record) = record else {
emit_missing_memory_demote_audit(
&state,
&tenant_context,
&input.run_id,
&input.id,
"memory not found",
)
.await?;
return Err(StatusCode::NOT_FOUND);
};
let changed = db
.set_global_memory_visibility_for_tenant(
&input.id,
&tenant_context.org_id,
&tenant_context.workspace_id,
tenant_context.deployment_id.as_deref(),
"private",
true,
)
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
if !changed {
emit_missing_memory_demote_audit(
&state,
&tenant_context,
&input.run_id,
&input.id,
"memory not found",
)
.await?;
return Err(StatusCode::NOT_FOUND);
}
let partition_key = memory_linkage(&record)
.get("partition_key")
.and_then(Value::as_str)
.unwrap_or("demoted")
.to_string();
let demote_detail = format!(
"kind={} classification={} artifact_refs={} visibility=private tier={} partition_key={} demoted=true{}",
memory_kind_label(&record.source_type),
memory_classification_label(record.metadata.as_ref()),
memory_artifact_refs(record.metadata.as_ref())
.iter()
.map(String::as_str)
.collect::<Vec<_>>()
.join(","),
tandem_memory::GovernedMemoryTier::Session,
partition_key,
memory_linkage_detail(&memory_linkage(&record))
);
let audit_id = Uuid::new_v4().to_string();
append_memory_audit(
&state,
&tenant_context,
crate::MemoryAuditEvent {
audit_id: audit_id.clone(),
action: "memory_demote".to_string(),
run_id: input.run_id.clone(),
tenant_context: tenant_context.clone(),
memory_id: Some(input.id.clone()),
source_memory_id: None,
to_tier: None,
partition_key: partition_key.clone(),
actor: "system".to_string(),
status: "ok".to_string(),
detail: Some(demote_detail),
created_at_ms: crate::now_ms(),
},
)
.await?;
publish_tenant_event(
&state,
&tenant_context,
"memory.updated",
json!({
"memoryID": input.id,
"runID": input.run_id,
"action": "demote",
"kind": memory_kind_label(&record.source_type),
"classification": memory_classification_label(record.metadata.as_ref()),
"artifactRefs": memory_artifact_refs(record.metadata.as_ref()),
"visibility": "private",
"tier": tandem_memory::GovernedMemoryTier::Session,
"partitionKey": partition_key,
"demoted": true,
"linkage": memory_linkage(&record),
"auditID": audit_id,
}),
);
Ok(Json(json!({
"ok": true,
"audit_id": audit_id,
})))
}
#[derive(Debug, Deserialize)]
pub(super) struct ContextResolveUriRequest {
uri: String,
}
#[derive(Debug, Deserialize)]
pub(super) struct ContextTreeQuery {
uri: String,
max_depth: Option<usize>,
}
#[derive(Debug, Deserialize)]
pub(super) struct ContextGenerateLayersRequest {
node_id: String,
}
#[derive(Debug, Deserialize)]
pub(super) struct ContextDistillRequest {
session_id: String,
conversation: Vec<String>,
#[serde(default)]
run_id: Option<String>,
#[serde(default)]
workflow_id: Option<String>,
#[serde(default)]
project_id: Option<String>,
#[serde(default)]
artifact_refs: Vec<String>,
#[serde(default)]
subject: Option<String>,
#[serde(default)]
importance_threshold: Option<f64>,
}
pub(super) async fn context_resolve_uri(
State(state): State<AppState>,
Json(input): Json<ContextResolveUriRequest>,
) -> Result<Json<Value>, StatusCode> {
let manager = open_memory_manager_for_state(&state)
.await
.ok_or(StatusCode::INTERNAL_SERVER_ERROR)?;
let node = manager
.resolve_uri(&input.uri)
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
Ok(Json(json!({ "node": node })))
}
pub(super) async fn context_tree(
State(state): State<AppState>,
Query(query): Query<ContextTreeQuery>,
) -> Result<Json<Value>, StatusCode> {
let manager = open_memory_manager_for_state(&state)
.await
.ok_or(StatusCode::INTERNAL_SERVER_ERROR)?;
let max_depth = query.max_depth.unwrap_or(3);
let tree = manager
.tree(&query.uri, max_depth)
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
Ok(Json(json!({ "tree": tree })))
}
pub(super) async fn context_generate_layers(
State(state): State<AppState>,
Json(input): Json<ContextGenerateLayersRequest>,
) -> Result<Json<Value>, StatusCode> {
let runtime_state = state.runtime.wait();
let providers = runtime_state.providers.clone();
let manager = open_memory_manager_for_state(&state)
.await
.ok_or(StatusCode::INTERNAL_SERVER_ERROR)?;
manager
.generate_layers_for_node(&input.node_id, &providers)
.await
.map_err(|e| {
tracing::warn!("Failed to generate layers: {}", e);
StatusCode::INTERNAL_SERVER_ERROR
})?;
Ok(Json(json!({ "ok": true })))
}
pub(super) async fn context_distill(
State(state): State<AppState>,
Extension(tenant_context): Extension<TenantContext>,
Json(input): Json<ContextDistillRequest>,
) -> Result<Json<Value>, StatusCode> {
let runtime_state = state.runtime.wait();
let providers = runtime_state.providers.clone();
let run_id = input
.run_id
.clone()
.unwrap_or_else(|| format!("distill-{}", input.session_id));
let project_id = input
.project_id
.clone()
.or_else(|| input.workflow_id.clone())
.unwrap_or_else(|| input.session_id.clone());
let subject = run_memory_subject(
input
.subject
.as_deref()
.or(tenant_context.actor_id.as_deref()),
);
let partition = tandem_memory::MemoryPartition {
org_id: tenant_context.org_id.clone(),
workspace_id: tenant_context.workspace_id.clone(),
project_id,
tier: tandem_memory::GovernedMemoryTier::Session,
};
let capability = issue_run_memory_capability(
&run_id,
Some(subject.as_str()),
&partition,
RunMemoryCapabilityPolicy::CoderWorkflow,
);
let writer = GovernedDistillationWriter {
state: state.clone(),
tenant_context: tenant_context.clone(),
partition,
capability,
run_id,
workflow_id: input.workflow_id.clone(),
artifact_refs: input.artifact_refs.clone(),
subject,
};
let threshold = input.importance_threshold.unwrap_or(0.5).clamp(0.0, 1.0);
let distiller = tandem_memory::SessionDistiller::with_threshold(Arc::new(providers), threshold);
let report = distiller
.distill_with_writer(&input.session_id, &input.conversation, &writer)
.await
.map_err(|e| {
tracing::warn!("Failed to distill session: {}", e);
StatusCode::INTERNAL_SERVER_ERROR
})?;
let distillation_id = report.distillation_id.clone();
let session_id = report.session_id.clone();
let facts_extracted = report.facts_extracted;
let stored_count = report.stored_count;
let deduped_count = report.deduped_count;
let memory_ids = report.memory_ids.clone();
let candidate_ids = report.candidate_ids.clone();
let status = report.status.clone();
Ok(Json(json!({
"ok": true,
"distillation_id": distillation_id,
"session_id": session_id,
"facts_extracted": facts_extracted,
"stored_count": stored_count,
"deduped_count": deduped_count,
"memory_ids": memory_ids,
"candidate_ids": candidate_ids,
"status": status,
"report": report,
})))
}
pub(super) async fn workflow_learning_candidates_list(
State(state): State<AppState>,
Query(query): Query<WorkflowLearningCandidateListQuery>,
) -> Result<Json<Value>, StatusCode> {
let status = match query.status.as_deref() {
Some(value) => {
Some(workflow_learning_status_from_str(value).ok_or(StatusCode::BAD_REQUEST)?)
}
None => None,
};
let kind = match query.kind.as_deref() {
Some(value) => Some(workflow_learning_kind_from_str(value).ok_or(StatusCode::BAD_REQUEST)?),
None => None,
};
let mut candidates = state
.list_workflow_learning_candidates(query.workflow_id.as_deref(), status, kind)
.await;
if let Some(project_id) = query.project_id.as_deref() {
candidates.retain(|candidate| candidate.project_id == project_id);
}
let count = candidates.len();
Ok(Json(json!({
"candidates": candidates,
"count": count,
})))
}
pub(super) async fn workflow_learning_candidate_review(
State(state): State<AppState>,
Path(candidate_id): Path<String>,
Json(input): Json<WorkflowLearningCandidateReviewRequest>,
) -> Result<Json<Value>, StatusCode> {
let Some(candidate) = state.get_workflow_learning_candidate(&candidate_id).await else {
return Err(StatusCode::NOT_FOUND);
};
let action = input
.action
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or("approve")
.to_ascii_lowercase();
let next_status = match action.as_str() {
"approve" | "approved" => WorkflowLearningCandidateStatus::Approved,
"reject" | "rejected" => WorkflowLearningCandidateStatus::Rejected,
"applied" => WorkflowLearningCandidateStatus::Applied,
"supersede" | "superseded" => WorkflowLearningCandidateStatus::Superseded,
"regress" | "regressed" => WorkflowLearningCandidateStatus::Regressed,
_ => return Err(StatusCode::BAD_REQUEST),
};
let baseline = if matches!(
next_status,
WorkflowLearningCandidateStatus::Approved | WorkflowLearningCandidateStatus::Applied
) {
Some(
state
.workflow_learning_metrics_for_workflow(&candidate.workflow_id)
.await,
)
} else {
None
};
let reviewed_at_ms = crate::now_ms();
let updated = state
.update_workflow_learning_candidate(&candidate_id, |candidate| {
candidate.status = next_status;
if candidate.baseline_before.is_none() {
candidate.baseline_before = baseline.clone();
}
if let Some(note) = input
.note
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
{
candidate.evidence_refs.push(json!({
"review_note": note,
"reviewer_id": input.reviewer_id,
"reviewed_at_ms": reviewed_at_ms,
"action": action,
}));
}
})
.await
.ok_or(StatusCode::NOT_FOUND)?;
Ok(Json(json!({
"ok": true,
"candidate": updated,
})))
}