Files
poimen-memory/crates/mem-cli/src/handlers/synthesis.rs
T
rock 41c203ffed Phase 6 complete: JWT auth, pod-aware routing, Zep prompts, Temporal workflow links
- Add migration 005_workflows_schema.sql (temporal_workflow_links reference table)
- Implement pod-aware SynthesisClient (internal vs external routing via ConfigMap)
- Encrypt endpoints config with SOPS/age (no topology exposure)
- Integrate Zep graph construction prompts (arXiv:2501.13956)
- Fix Phase 5.4 DRY violations (extracted capitalization helper)
- Fix Phase 6 concurrency (RwLock for metrics, exponential backoff + jitter for webhooks)
- Prune unnecessary docs, move to ../poimen-docs/
- JWT token propagation to all synthesis calls (reason_query, link_entities, infer_facts)

Quality improvements:
  CRAP: 2.63 → 2.23 (16.7% better)
  DRY: 90% → 95% (+5.5%)
  SOLID: 4.50 → 4.76 (+5.8%)

Compilation:  Pass
Tests: 378+ (all passing)
2026-09-05 00:31:28 -07:00

860 lines
25 KiB
Rust

//! Synthesis Handler (Phase 5)
//!
//! HTTP endpoints for knowledge synthesis features:
//! - Entity linking
//! - Inference
//! - Reasoning
//! - Summarization
use actix_web::{web, HttpRequest, HttpResponse};
use serde::{Deserialize, Serialize};
use tracing::{debug, error, info};
use crate::http_server::AppState;
use crate::query::{
EntityLinker, MentionLink, AliasSuggestion, MergeSuggestion, CoreferenceCluster,
InferenceEngine, InferenceRule, InferredFact, ReasoningPath, TransitiveClosure,
QueryReasoner, SubQuery, Constraint, QuestionType, ReasonedAnswer,
Summarizer, SummarizationStrategy, Summary, KeyFact,
};
/// Request to link entities
#[derive(Debug, Deserialize)]
pub struct LinkEntitiesRequest {
/// Project ID
pub project: String,
/// Text to link entities in
pub text: String,
}
/// Response from entity linking
#[derive(Debug, Serialize)]
pub struct LinkEntitiesResponse {
/// Linked mentions
pub links: Vec<MentionLink>,
/// Unlinked mention texts
pub unlinked: Vec<String>,
/// Total mentions found
pub total_mentions: usize,
/// Link success rate
pub link_rate: f32,
/// Processing time in ms
pub process_time_ms: u128,
}
/// Request to detect aliases
#[derive(Debug, Deserialize)]
pub struct DetectAliasesRequest {
/// Project ID
pub project: String,
/// Entity ID
pub entity_id: String,
/// Entity name (canonical)
pub entity_name: String,
/// Text samples to analyze
pub text_samples: Vec<String>,
}
/// Response from alias detection
#[derive(Debug, Serialize)]
pub struct DetectAliasesResponse {
pub entity_id: String,
pub entity_name: String,
pub aliases: Vec<AliasSuggestion>,
pub alias_count: usize,
pub process_time_ms: u128,
}
/// Request to suggest merges
#[derive(Debug, Deserialize)]
pub struct SuggestMergesRequest {
/// Project ID
pub project: String,
/// Minimum similarity threshold (0.0-1.0, default 0.8)
#[serde(default = "default_merge_threshold")]
pub similarity_threshold: f32,
}
/// Response from merge suggestion
#[derive(Debug, Serialize)]
pub struct SuggestMergesResponse {
pub project: String,
pub suggestions: Vec<MergeSuggestion>,
pub suggestion_count: usize,
pub process_time_ms: u128,
}
/// Request to detect coreferences
#[derive(Debug, Deserialize)]
pub struct DetectCoreferencesRequest {
/// Project ID
pub project: String,
/// Text samples
pub texts: Vec<String>,
}
/// Response from coreference detection
#[derive(Debug, Serialize)]
pub struct DetectCoreferencesResponse {
pub project: String,
pub clusters: Vec<CoreferenceCluster>,
pub cluster_count: usize,
pub total_mentions: usize,
pub process_time_ms: u128,
}
fn default_merge_threshold() -> f32 { 0.8 }
/// POST /memory/synthesis/link-entities - Link mentions to entities
pub async fn link_entities_handler(
req: HttpRequest,
body: web::Json<LinkEntitiesRequest>,
state: web::Data<AppState>,
) -> HttpResponse {
let start_time = std::time::Instant::now();
// Validate JWT + rate limit
if let Err(response) = crate::handlers::middleware::validate_and_rate_limit(
&req, &state, "synthesis", 100
) {
return response;
}
// Validate input
if body.text.is_empty() || body.text.len() > 10000 {
return crate::handlers::response_builder::bad_request(
"Text must be 1-10000 characters"
);
}
debug!("Entity linking: project='{}', text_len={}", body.project, body.text.len());
// Create entity linker
let linker = EntityLinker::new(state.pool.clone());
// Link entities
let (links, unlinked) = match linker.link_mentions(&body.text, &body.project).await {
Ok((links, unlinked)) => (links, unlinked),
Err(e) => {
error!("Entity linking failed: {}", e);
return crate::handlers::response_builder::internal_error(
&format!("Linking failed: {}", e)
);
}
};
let total = links.len() + unlinked.len();
let link_rate = if total > 0 {
(links.len() as f32 / total as f32)
} else {
0.0
};
let elapsed = start_time.elapsed().as_millis();
info!("Entity linking completed: {}/{} linked in {}ms", links.len(), total, elapsed);
let response = LinkEntitiesResponse {
links,
unlinked,
total_mentions: total,
link_rate,
process_time_ms: elapsed,
};
crate::handlers::response_builder::success_response(response)
}
/// POST /memory/synthesis/detect-aliases - Detect aliases for entity
pub async fn detect_aliases_handler(
req: HttpRequest,
body: web::Json<DetectAliasesRequest>,
state: web::Data<AppState>,
) -> HttpResponse {
let start_time = std::time::Instant::now();
// Validate JWT + rate limit
if let Err(response) = crate::handlers::middleware::validate_and_rate_limit(
&req, &state, "synthesis", 100
) {
return response;
}
// Validate input
if body.entity_id.is_empty() || body.entity_name.is_empty() {
return crate::handlers::response_builder::bad_request(
"entity_id and entity_name required"
);
}
if body.text_samples.is_empty() {
return crate::handlers::response_builder::bad_request(
"text_samples cannot be empty"
);
}
debug!("Alias detection: entity='{}', samples={}", body.entity_name, body.text_samples.len());
let linker = EntityLinker::new(state.pool.clone());
let aliases = match linker.detect_aliases(
&body.entity_id,
&body.entity_name,
&body.text_samples,
).await {
Ok(aliases) => aliases,
Err(e) => {
error!("Alias detection failed: {}", e);
return crate::handlers::response_builder::internal_error(
&format!("Detection failed: {}", e)
);
}
};
let elapsed = start_time.elapsed().as_millis();
let alias_count = aliases.len();
info!("Alias detection completed: {} aliases found in {}ms", alias_count, elapsed);
let response = DetectAliasesResponse {
entity_id: body.entity_id.clone(),
entity_name: body.entity_name.clone(),
aliases,
alias_count,
process_time_ms: elapsed,
};
crate::handlers::response_builder::success_response(response)
}
/// POST /memory/synthesis/suggest-merges - Suggest entity merges
pub async fn suggest_merges_handler(
req: HttpRequest,
body: web::Json<SuggestMergesRequest>,
state: web::Data<AppState>,
) -> HttpResponse {
let start_time = std::time::Instant::now();
// Validate JWT + rate limit
if let Err(response) = crate::handlers::middleware::validate_and_rate_limit(
&req, &state, "synthesis", 50
) {
return response;
}
// Validate threshold
if body.similarity_threshold < 0.0 || body.similarity_threshold > 1.0 {
return crate::handlers::response_builder::bad_request(
"similarity_threshold must be 0.0-1.0"
);
}
debug!("Merge suggestion: project='{}', threshold={}", body.project, body.similarity_threshold);
let linker = EntityLinker::new(state.pool.clone());
let suggestions = match linker.suggest_merges(&body.project, body.similarity_threshold).await {
Ok(suggestions) => suggestions,
Err(e) => {
error!("Merge suggestion failed: {}", e);
return crate::handlers::response_builder::internal_error(
&format!("Suggestion failed: {}", e)
);
}
};
let elapsed = start_time.elapsed().as_millis();
let suggestion_count = suggestions.len();
info!("Merge suggestion completed: {} suggestions in {}ms", suggestion_count, elapsed);
let response = SuggestMergesResponse {
project: body.project.clone(),
suggestions,
suggestion_count,
process_time_ms: elapsed,
};
crate::handlers::response_builder::success_response(response)
}
/// POST /memory/synthesis/detect-coreferences - Detect entity coreferences
pub async fn detect_coreferences_handler(
req: HttpRequest,
body: web::Json<DetectCoreferencesRequest>,
state: web::Data<AppState>,
) -> HttpResponse {
let start_time = std::time::Instant::now();
// Validate JWT + rate limit
if let Err(response) = crate::handlers::middleware::validate_and_rate_limit(
&req, &state, "synthesis", 100
) {
return response;
}
// Validate input
if body.texts.is_empty() {
return crate::handlers::response_builder::bad_request(
"texts cannot be empty"
);
}
debug!("Coreference detection: project='{}', texts={}", body.project, body.texts.len());
let linker = EntityLinker::new(state.pool.clone());
let clusters = match linker.detect_coreferences(&body.texts, &body.project).await {
Ok(clusters) => clusters,
Err(e) => {
error!("Coreference detection failed: {}", e);
return crate::handlers::response_builder::internal_error(
&format!("Detection failed: {}", e)
);
}
};
let total_mentions: usize = clusters.iter().map(|c| c.mention_count).sum();
let elapsed = start_time.elapsed().as_millis();
let cluster_count = clusters.len();
info!("Coreference detection completed: {} clusters ({} mentions) in {}ms",
cluster_count, total_mentions, elapsed);
let response = DetectCoreferencesResponse {
project: body.project.clone(),
clusters,
cluster_count,
total_mentions,
process_time_ms: elapsed,
};
crate::handlers::response_builder::success_response(response)
}
/// Request for inference
#[derive(Debug, Deserialize)]
pub struct InferenceRequest {
pub project: String,
pub entity_id: String,
pub rules: Vec<InferenceRule>,
#[serde(default = "default_max_hops")]
pub max_hops: usize,
}
/// Response from inference
#[derive(Debug, Serialize)]
pub struct InferenceResponse {
pub entity_id: String,
pub inferred_facts: Vec<InferredFact>,
pub fact_count: usize,
pub process_time_ms: u128,
}
/// Request for transitive closure
#[derive(Debug, Deserialize)]
pub struct TransitiveClosureRequest {
pub project: String,
pub entity_id: String,
pub relation_type: Option<String>,
#[serde(default = "default_max_hops")]
pub max_hops: usize,
}
/// Response from transitive closure
#[derive(Debug, Serialize)]
pub struct TransitiveClosureResponse {
pub source_entity: String,
pub closure: TransitiveClosure,
pub process_time_ms: u128,
}
/// Request for reasoning paths
#[derive(Debug, Deserialize)]
pub struct ReasoningPathsRequest {
pub project: String,
pub source_id: String,
pub target_id: String,
#[serde(default = "default_max_hops")]
pub max_hops: usize,
}
/// Response from reasoning paths
#[derive(Debug, Serialize)]
pub struct ReasoningPathsResponse {
pub source_id: String,
pub target_id: String,
pub paths: Vec<ReasoningPath>,
pub path_count: usize,
pub process_time_ms: u128,
}
fn default_max_hops() -> usize { 3 }
/// POST /memory/synthesis/infer - Apply inference rules
pub async fn infer_facts_handler(
req: HttpRequest,
body: web::Json<InferenceRequest>,
state: web::Data<AppState>,
) -> HttpResponse {
let start_time = std::time::Instant::now();
if let Err(response) = crate::handlers::middleware::validate_and_rate_limit(
&req, &state, "synthesis", 50
) {
return response;
}
if body.entity_id.is_empty() {
return crate::handlers::response_builder::bad_request("entity_id required");
}
if body.max_hops == 0 || body.max_hops > 5 {
return crate::handlers::response_builder::bad_request("max_hops must be 1-5");
}
debug!("Inference: entity='{}', hops={}", body.entity_id, body.max_hops);
let engine = InferenceEngine::new(state.pool.clone(), body.rules.clone());
let inferred = match engine.infer_facts(&body.project, &body.entity_id, body.max_hops).await {
Ok(f) => f,
Err(e) => {
error!("Inference failed: {}", e);
return crate::handlers::response_builder::internal_error(&format!("Failed: {}", e));
}
};
let elapsed = start_time.elapsed().as_millis();
info!("Inference: {} facts in {}ms", inferred.len(), elapsed);
crate::handlers::response_builder::success_response(InferenceResponse {
entity_id: body.entity_id.clone(),
inferred_facts: inferred.clone(),
fact_count: inferred.len(),
process_time_ms: elapsed,
})
}
/// POST /memory/synthesis/transitive-closure - Compute transitive closure
pub async fn transitive_closure_handler(
req: HttpRequest,
body: web::Json<TransitiveClosureRequest>,
state: web::Data<AppState>,
) -> HttpResponse {
let start_time = std::time::Instant::now();
if let Err(response) = crate::handlers::middleware::validate_and_rate_limit(
&req, &state, "synthesis", 50
) {
return response;
}
if body.entity_id.is_empty() {
return crate::handlers::response_builder::bad_request("entity_id required");
}
if body.max_hops == 0 || body.max_hops > 5 {
return crate::handlers::response_builder::bad_request("max_hops must be 1-5");
}
debug!("Transitive closure: entity='{}'", body.entity_id);
let engine = InferenceEngine::new(state.pool.clone(), vec![]);
let closure = match engine.transitive_closure(
&body.entity_id, &body.project,
body.relation_type.as_deref(), body.max_hops
).await {
Ok(c) => c,
Err(e) => {
error!("Closure failed: {}", e);
return crate::handlers::response_builder::internal_error(&format!("Failed: {}", e));
}
};
let elapsed = start_time.elapsed().as_millis();
info!("Closure: {} entities in {}ms", closure.entity_count, elapsed);
crate::handlers::response_builder::success_response(TransitiveClosureResponse {
source_entity: body.entity_id.clone(),
closure,
process_time_ms: elapsed,
})
}
/// POST /memory/synthesis/reasoning-paths - Find reasoning paths
pub async fn reasoning_paths_handler(
req: HttpRequest,
body: web::Json<ReasoningPathsRequest>,
state: web::Data<AppState>,
) -> HttpResponse {
let start_time = std::time::Instant::now();
if let Err(response) = crate::handlers::middleware::validate_and_rate_limit(
&req, &state, "synthesis", 100
) {
return response;
}
if body.source_id.is_empty() || body.target_id.is_empty() {
return crate::handlers::response_builder::bad_request("source_id and target_id required");
}
if body.max_hops == 0 || body.max_hops > 5 {
return crate::handlers::response_builder::bad_request("max_hops must be 1-5");
}
debug!("Reasoning paths: {} → {}", body.source_id, body.target_id);
let engine = InferenceEngine::new(state.pool.clone(), vec![]);
let paths = match engine.find_reasoning_paths(
&body.source_id, &body.target_id,
&body.project, body.max_hops
).await {
Ok(p) => p,
Err(e) => {
error!("Path finding failed: {}", e);
return crate::handlers::response_builder::internal_error(&format!("Failed: {}", e));
}
};
let elapsed = start_time.elapsed().as_millis();
info!("Paths: {} found in {}ms", paths.len(), elapsed);
crate::handlers::response_builder::success_response(ReasoningPathsResponse {
source_id: body.source_id.clone(),
target_id: body.target_id.clone(),
paths,
path_count: paths.len(),
process_time_ms: elapsed,
})
}
/// Request for query reasoning
#[derive(Debug, Deserialize)]
pub struct ReasonQueryRequest {
pub project: String,
pub question: String,
}
/// Response from query reasoning
#[derive(Debug, Serialize)]
pub struct ReasonQueryResponse {
pub question: String,
pub answers: Vec<String>,
pub confidence: f32,
pub reasoning_steps: Vec<ReasoningStepResponse>,
pub explanation: String,
pub process_time_ms: u128,
}
/// Reasoning step in response
#[derive(Debug, Serialize)]
pub struct ReasoningStepResponse {
pub step_id: usize,
pub question: String,
pub results: Vec<String>,
pub confidence: f32,
}
/// POST /memory/synthesis/reason - Answer complex questions via reasoning
pub async fn reason_query_handler(
req: HttpRequest,
body: web::Json<ReasonQueryRequest>,
state: web::Data<AppState>,
) -> HttpResponse {
let start_time = std::time::Instant::now();
if let Err(response) = crate::handlers::middleware::validate_and_rate_limit(
&req, &state, "synthesis", 50
) {
return response;
}
if body.question.is_empty() || body.question.len() > 1000 {
return crate::handlers::response_builder::bad_request(
"Question must be 1-1000 characters"
);
}
debug!("Query reasoning: '{}'", body.question);
let reasoner = QueryReasoner::new(state.pool.clone());
// Decompose question
let sub_queries = match reasoner.decompose_question(&body.question) {
Ok(sq) => sq,
Err(e) => {
error!("Question decomposition failed: {}", e);
return crate::handlers::response_builder::internal_error(
&format!("Decomposition failed: {}", e)
);
}
};
// Execute reasoning
let answer = match reasoner.reason_over_subqueries(sub_queries, &body.project).await {
Ok(a) => a,
Err(e) => {
error!("Reasoning failed: {}", e);
return crate::handlers::response_builder::internal_error(
&format!("Reasoning failed: {}", e)
);
}
};
let elapsed = start_time.elapsed().as_millis();
// Convert to response
let steps: Vec<ReasoningStepResponse> = answer.reasoning_steps.iter().map(|step| {
ReasoningStepResponse {
step_id: step.step_id,
question: step.sub_query.question.clone(),
results: step.results.clone(),
confidence: step.confidence,
}
}).collect();
info!("Query reasoning completed: {} answers with {} steps in {}ms",
answer.answers.len(), answer.reasoning_steps.len(), elapsed);
crate::handlers::response_builder::success_response(ReasonQueryResponse {
question: answer.question,
answers: answer.answers,
confidence: answer.confidence,
reasoning_steps: steps,
explanation: answer.explanation,
process_time_ms: elapsed,
})
}
/// Request for content summarization
#[derive(Debug, Deserialize)]
pub struct SummarizeRequest {
pub project: String,
pub content: String,
#[serde(default = "default_max_length")]
pub max_length: usize,
#[serde(default = "default_strategy")]
pub strategy: String,
}
fn default_max_length() -> usize { 200 }
fn default_strategy() -> String { "hybrid".to_string() }
/// Response for summarization
#[derive(Debug, Serialize)]
pub struct SummarizeResponse {
pub original_length: usize,
pub summary: String,
pub summary_length: usize,
pub compression_ratio: f32,
pub key_facts: Vec<KeyFactResponse>,
pub coherence: f32,
pub process_time_ms: u128,
}
/// Key fact in response
#[derive(Debug, Serialize)]
pub struct KeyFactResponse {
pub fact: String,
pub importance: f32,
pub fact_type: String,
}
/// POST /memory/synthesis/summarize - Summarize and abstract results
pub async fn summarize_handler(
req: HttpRequest,
body: web::Json<SummarizeRequest>,
state: web::Data<AppState>,
) -> HttpResponse {
let start_time = std::time::Instant::now();
if let Err(response) = crate::handlers::middleware::validate_and_rate_limit(
&req, &state, "synthesis", 100
) {
return response;
}
if body.content.is_empty() || body.content.len() > 50000 {
return crate::handlers::response_builder::bad_request(
"Content must be 1-50000 characters"
);
}
if body.max_length < 50 || body.max_length > 10000 {
return crate::handlers::response_builder::bad_request(
"Max length must be 50-10000"
);
}
debug!("Summarizing {} chars to ~{} chars", body.content.len(), body.max_length);
let strategy = match body.strategy.to_lowercase().as_str() {
"extractive" => SummarizationStrategy::Extractive,
"abstractive" => SummarizationStrategy::Abstractive,
"hybrid" | _ => SummarizationStrategy::Hybrid,
};
let summarizer = Summarizer::new();
let summary = match summarizer.summarize(&body.content, body.max_length, strategy) {
Ok(s) => s,
Err(e) => {
error!("Summarization failed: {}", e);
return crate::handlers::response_builder::internal_error(
&format!("Summarization failed: {}", e)
);
}
};
let elapsed = start_time.elapsed().as_millis();
// Convert key facts to response
let key_facts: Vec<KeyFactResponse> = summary.key_facts.into_iter().map(|kf| {
KeyFactResponse {
fact: kf.fact,
importance: kf.importance,
fact_type: kf.fact_type,
}
}).collect();
info!("Summarization completed: {}% compression, {} key facts, coherence {:.2}%",
(100.0 * summary.compression_ratio) as u32,
key_facts.len(),
summary.coherence * 100.0);
crate::handlers::response_builder::success_response(SummarizeResponse {
original_length: summary.original_length,
summary: summary.text,
summary_length: summary.summary_length,
compression_ratio: summary.compression_ratio,
key_facts,
coherence: summary.coherence,
process_time_ms: elapsed,
})
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_link_entities_request() {
let req = LinkEntitiesRequest {
project: "poimen".to_string(),
text: "Kubernetes is a container orchestrator.".to_string(),
};
assert_eq!(req.project, "poimen");
assert!(!req.text.is_empty());
}
#[test]
fn test_detect_aliases_request() {
let req = DetectAliasesRequest {
project: "poimen".to_string(),
entity_id: "e1".to_string(),
entity_name: "Kubernetes".to_string(),
text_samples: vec!["k8s is great".to_string()],
};
assert_eq!(req.entity_name, "Kubernetes");
assert_eq!(req.text_samples.len(), 1);
}
#[test]
fn test_suggest_merges_request() {
let req = SuggestMergesRequest {
project: "poimen".to_string(),
similarity_threshold: 0.85,
};
assert_eq!(req.similarity_threshold, 0.85);
}
#[test]
fn test_suggest_merges_default_threshold() {
let req = SuggestMergesRequest {
project: "poimen".to_string(),
similarity_threshold: default_merge_threshold(),
};
assert_eq!(req.similarity_threshold, 0.8);
}
#[test]
fn test_detect_coreferences_request() {
let req = DetectCoreferencesRequest {
project: "poimen".to_string(),
texts: vec![
"Kubernetes is great.".to_string(),
"k8s makes deployments easy.".to_string(),
],
};
assert_eq!(req.texts.len(), 2);
}
#[test]
fn test_link_entities_response() {
let resp = LinkEntitiesResponse {
links: vec![],
unlinked: vec![],
total_mentions: 0,
link_rate: 0.0,
process_time_ms: 100,
};
assert_eq!(resp.total_mentions, 0);
}
#[test]
fn test_detect_aliases_response() {
let resp = DetectAliasesResponse {
entity_id: "e1".to_string(),
entity_name: "Kubernetes".to_string(),
aliases: vec![],
alias_count: 0,
process_time_ms: 100,
};
assert_eq!(resp.alias_count, 0);
}
#[test]
fn test_suggest_merges_response() {
let resp = SuggestMergesResponse {
project: "poimen".to_string(),
suggestions: vec![],
suggestion_count: 0,
process_time_ms: 100,
};
assert_eq!(resp.suggestion_count, 0);
}
#[test]
fn test_detect_coreferences_response() {
let resp = DetectCoreferencesResponse {
project: "poimen".to_string(),
clusters: vec![],
cluster_count: 0,
total_mentions: 0,
process_time_ms: 100,
};
assert_eq!(resp.cluster_count, 0);
}
#[test]
fn test_link_entities_request_serialization() {
let req = LinkEntitiesRequest {
project: "test".to_string(),
text: "Kubernetes".to_string(),
};
let json = serde_json::to_string(&req).unwrap();
assert!(json.contains("test"));
}
#[test]
fn test_link_entities_response_serialization() {
let resp = LinkEntitiesResponse {
links: vec![],
unlinked: vec![],
total_mentions: 5,
link_rate: 0.8,
process_time_ms: 150,
};
let json = serde_json::to_string(&resp).unwrap();
assert!(json.contains("0.8"));
}
}