diff --git a/crates/temper-server/src/observe/evolution.rs b/crates/temper-server/src/observe/evolution.rs index 5342d9cd0..b4d7566e8 100644 --- a/crates/temper-server/src/observe/evolution.rs +++ b/crates/temper-server/src/observe/evolution.rs @@ -2,6 +2,8 @@ pub(crate) mod insight_generator; mod operations; +#[cfg(test)] +pub(crate) use operations::{materialize_feature_requests_for_test, stable_feature_request_id}; mod records_detail; mod records_list; mod trajectories; diff --git a/crates/temper-server/src/observe/evolution/insight_generator/gap_analysis.rs b/crates/temper-server/src/observe/evolution/insight_generator/gap_analysis.rs index bcd6a998b..964f27995 100644 --- a/crates/temper-server/src/observe/evolution/insight_generator/gap_analysis.rs +++ b/crates/temper-server/src/observe/evolution/insight_generator/gap_analysis.rs @@ -182,14 +182,21 @@ pub(crate) fn generate_feature_requests( }); accum.count += 1; accum.timestamps.push(entry.timestamp.clone()); + if let Some(error) = entry.error.as_ref() + && error < &accum.description + { + accum.description.clone_from(error); + } } let mut feature_requests = Vec::new(); - for accum in groups.into_values() { + for mut accum in groups.into_values() { if accum.count < FEATURE_REQUEST_THRESHOLD { continue; } + accum.timestamps.sort(); + let category = match accum.error_pattern.as_str() { "EntitySetNotFound" => PlatformGapCategory::MissingCapability, "ActionNotFound" => PlatformGapCategory::MissingMethod, diff --git a/crates/temper-server/src/observe/evolution/operations.rs b/crates/temper-server/src/observe/evolution/operations.rs index b6e06a4ed..6cf4ce299 100644 --- a/crates/temper-server/src/observe/evolution/operations.rs +++ b/crates/temper-server/src/observe/evolution/operations.rs @@ -5,6 +5,7 @@ use axum::extract::{Path, Query, State}; use axum::http::{HeaderMap, StatusCode}; use axum::response::Json; use axum::response::sse::{Event, KeepAlive, Sse}; +use sha2::{Digest, Sha256}; use temper_evolution::FeatureRequestDisposition; use temper_runtime::tenant::TenantId; use tokio_stream::StreamExt; @@ -19,15 +20,19 @@ use crate::sentinel; use crate::state::{ObserveRefreshHint, ServerState}; mod materialize; +mod reconcile; mod support; pub(crate) use materialize::{handle_evolution_analyze, handle_evolution_materialize}; +use reconcile::reconcile_legacy_feature_requests; use support::{ - create_system_entity_logged, emit_refresh_hints, next_system_entity_id, persist_alerts, - persist_insights, spawn_intent_discovery, + dispatch_system_action_idempotent, emit_refresh_hints, persist_alerts, persist_insights, + spawn_intent_discovery, }; +const FEATURE_REQUEST_GENERATOR_VERSION: &str = "v1"; + /// POST /api/evolution/sentinel/check -- trigger sentinel rule evaluation. /// /// Evaluates all default sentinel rules against current server state. @@ -106,6 +111,8 @@ pub(crate) async fn handle_sentinel_check( tracing::Span::current().record("insights_count", insights.len()); tracing::info!(insights_count = insights.len(), "evolution.insight"); let insight_results = persist_insights(&state, &insights).await; + let feature_request_ids = + materialize_feature_requests(&state, &analysis_tenant, &trajectory_entries).await?; emit_refresh_hints( &state, @@ -123,9 +130,133 @@ pub(crate) async fn handle_sentinel_check( "intent_discoveries": discovery_results, "insights_count": insights.len(), "insights": insight_results, + "feature_requests_count": feature_request_ids.len(), + "feature_request_ids": feature_request_ids, }))) } +pub(crate) fn stable_feature_request_id( + tenant: &TenantId, + generator_version: &str, + feature_request: &temper_evolution::FeatureRequestRecord, +) -> String { + let mut trajectory_refs = feature_request.trajectory_refs.clone(); + trajectory_refs.sort(); + let category = match feature_request.category { + temper_evolution::PlatformGapCategory::MissingMethod => "missing_method", + temper_evolution::PlatformGapCategory::GovernanceBlocked => "governance_blocked", + temper_evolution::PlatformGapCategory::UnsupportedIntegration => "unsupported_integration", + temper_evolution::PlatformGapCategory::MissingCapability => "missing_capability", + }; + let identity = serde_json::json!({ + "tenant": tenant.as_str(), + "generator_version": generator_version, + "category": category, + "description": feature_request.description, + "trajectory_refs": trajectory_refs, + }); + format!("FR-{:x}", Sha256::digest(identity.to_string())) +} + +async fn materialize_feature_requests( + state: &ServerState, + tenant: &TenantId, + trajectory_entries: &[crate::state::TrajectoryEntry], +) -> Result, StatusCode> { + let tenant_entries = trajectory_entries + .iter() + .filter(|entry| entry.tenant == tenant.as_str()) + .cloned() + .collect::>(); + let generated = insight_generator::generate_feature_requests(&tenant_entries); + if generated.is_empty() { + return Ok(Vec::new()); + } + let store = state + .platform_metadata_store() + .ok_or(StatusCode::SERVICE_UNAVAILABLE)?; + let existing_rows = store + .list_feature_requests(None) + .await + .map_err(|error| { + tracing::error!(error = %error, backend = store.backend_name(), "failed to load feature requests for reconciliation"); + StatusCode::INTERNAL_SERVER_ERROR + })?; + let mut materialized_ids = Vec::with_capacity(generated.len()); + + for feature_request in &generated { + let stable_id = + stable_feature_request_id(tenant, FEATURE_REQUEST_GENERATOR_VERSION, feature_request); + let category = format!("{:?}", feature_request.category); + let refs_json = serde_json::to_string(&feature_request.trajectory_refs) + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + let disposition = match feature_request.disposition { + FeatureRequestDisposition::Open => "Open", + FeatureRequestDisposition::Acknowledged => "Acknowledged", + FeatureRequestDisposition::Planned => "Planned", + FeatureRequestDisposition::WontFix => "WontFix", + FeatureRequestDisposition::Resolved => "Resolved", + }; + let params = serde_json::json!({ + "category": category, + "description": feature_request.description, + "frequency": feature_request.frequency.to_string(), + "developer_notes": feature_request.developer_notes.clone().unwrap_or_default(), + }); + + dispatch_system_action_idempotent( + state, + "FeatureRequest", + &stable_id, + "CreateFeatureRequest", + params, + &format!("feature-request:{stable_id}"), + ) + .await + .map_err(|error| { + tracing::error!(error = %error, feature_request_id = %stable_id, "failed to materialize feature request entity"); + StatusCode::INTERNAL_SERVER_ERROR + })?; + + store + .upsert_feature_request( + &stable_id, + &category, + &feature_request.description, + feature_request.frequency as i64, + &refs_json, + disposition, + feature_request.developer_notes.as_deref(), + ) + .await + .map_err(|error| { + tracing::error!(error = %error, backend = store.backend_name(), feature_request_id = %stable_id, "failed to project feature request"); + StatusCode::INTERNAL_SERVER_ERROR + })?; + reconcile_legacy_feature_requests( + store.as_ref(), + &existing_rows, + tenant, + trajectory_entries, + &stable_id, + feature_request, + ) + .await?; + materialized_ids.push(stable_id); + } + + Ok(materialized_ids) +} + +#[cfg(test)] +pub(crate) async fn materialize_feature_requests_for_test( + state: &ServerState, + tenant: &TenantId, + trajectory_entries: &[crate::state::TrajectoryEntry], +) -> Result, StatusCode> { + materialize_feature_requests(state, tenant, trajectory_entries).await +} + /// GET /observe/evolution/unmet-intents -- grouped unmet intents from trajectories. /// /// Uses a SQL GROUP BY aggregation instead of loading raw trajectory rows to @@ -228,51 +359,7 @@ pub(crate) async fn handle_feature_requests( require_observe_auth(&state, &headers, "read_evolution", "Evolution")?; let disposition_filter = params.get("disposition").map(String::as_str); - let trajectory_entries = state.load_trajectory_entries(1_000).await; - if let Some(store) = state.platform_metadata_store() { - let generated = insight_generator::generate_feature_requests(&trajectory_entries); - for feature_request in &generated { - let refs_json = serde_json::to_string(&feature_request.trajectory_refs) - .unwrap_or_else(|_| "[]".to_string()); - let disposition = match feature_request.disposition { - FeatureRequestDisposition::Open => "Open", - FeatureRequestDisposition::Acknowledged => "Acknowledged", - FeatureRequestDisposition::Planned => "Planned", - FeatureRequestDisposition::WontFix => "WontFix", - FeatureRequestDisposition::Resolved => "Resolved", - }; - if let Err(error) = store - .upsert_feature_request( - &feature_request.header.id, - &format!("{:?}", feature_request.category), - &feature_request.description, - feature_request.frequency as i64, - &refs_json, - disposition, - feature_request.developer_notes.as_deref(), - ) - .await - { - tracing::warn!(error = %error, backend = store.backend_name(), "failed to upsert feature request"); - } - - create_system_entity_logged( - &state, - "FeatureRequest", - &next_system_entity_id("FR"), - "CreateFeatureRequest", - serde_json::json!({ - "category": format!("{:?}", feature_request.category), - "description": feature_request.description, - "frequency": feature_request.frequency.to_string(), - "developer_notes": feature_request.developer_notes.clone().unwrap_or_default(), - "legacy_record_id": feature_request.header.id, - }), - ) - .await; - } - return match store.list_feature_requests(disposition_filter).await { Ok(rows) => { let feature_requests = rows diff --git a/crates/temper-server/src/observe/evolution/operations/reconcile.rs b/crates/temper-server/src/observe/evolution/operations/reconcile.rs new file mode 100644 index 000000000..88029ae69 --- /dev/null +++ b/crates/temper-server/src/observe/evolution/operations/reconcile.rs @@ -0,0 +1,190 @@ +//! Reconciliation of legacy feature-request projections into stable revisions. + +use std::collections::BTreeSet; + +use axum::http::StatusCode; +use temper_evolution::FeatureRequestRecord; +use temper_runtime::tenant::TenantId; +use temper_store_turso::FeatureRequestRow; + +use crate::state::TrajectoryEntry; + +fn normalized_trajectory_refs(raw: &str) -> Option> { + let mut refs = serde_json::from_str::>(raw).ok()?; + refs.sort(); + Some(refs) +} + +fn evidence_description_prefix(description: &str) -> &str { + description + .split_once(" — ") + .map_or(description, |(prefix, _)| prefix) +} + +fn is_same_evidence_revision( + row: &FeatureRequestRow, + category: &str, + feature_request: &FeatureRequestRecord, +) -> bool { + let mut generated_refs = feature_request.trajectory_refs.clone(); + generated_refs.sort(); + row.category == category + && row.frequency == feature_request.frequency as i64 + && evidence_description_prefix(&row.description) + == evidence_description_prefix(&feature_request.description) + && normalized_trajectory_refs(&row.trajectory_refs).as_ref() == Some(&generated_refs) +} + +fn is_legacy_feature_request_id(id: &str) -> bool { + let mut parts = id.split('-'); + matches!( + (parts.next(), parts.next(), parts.next(), parts.next()), + (Some("FR"), Some(year), Some(suffix), None) + if year.len() == 4 + && year.chars().all(|character| character.is_ascii_digit()) + && suffix.len() == 12 + && suffix.chars().all(|character| character.is_ascii_hexdigit()) + ) +} + +fn legacy_evidence_belongs_unambiguously_to_tenant( + row: &FeatureRequestRow, + tenant: &TenantId, + trajectory_entries: &[TrajectoryEntry], +) -> bool { + let Some(refs) = normalized_trajectory_refs(&row.trajectory_refs) else { + return false; + }; + let expected_refs = refs.iter().map(String::as_str).collect::>(); + if expected_refs.is_empty() { + return false; + } + + let mut matched_refs = BTreeSet::new(); + let mut source_tenants = BTreeSet::new(); + for entry in trajectory_entries { + if expected_refs.contains(entry.timestamp.as_str()) { + matched_refs.insert(entry.timestamp.as_str()); + source_tenants.insert(entry.tenant.as_str()); + } + } + matched_refs == expected_refs + && source_tenants.len() == 1 + && source_tenants.contains(tenant.as_str()) +} + +fn merged_review_state(rows: &[&FeatureRequestRow], stable_id: &str) -> (String, Option) { + let reviewed = rows + .iter() + .copied() + .filter(|row| { + row.disposition != "Open" + || row + .developer_notes + .as_deref() + .is_some_and(|notes| !notes.trim().is_empty()) + }) + .max_by(|left, right| (&left.updated_at, &left.id).cmp(&(&right.updated_at, &right.id))); + let disposition = reviewed + .map(|row| row.disposition.clone()) + .unwrap_or_else(|| "Open".to_string()); + + let canonical_notes = rows + .iter() + .find(|row| row.id == stable_id) + .and_then(|row| row.developer_notes.clone()); + let canonical_components = canonical_notes + .iter() + .flat_map(|notes| notes.split("\n\n")) + .map(str::trim) + .filter(|note| !note.is_empty()) + .collect::>(); + let mut legacy_components = BTreeSet::new(); + for row in rows.iter().filter(|row| row.id != stable_id) { + if let Some(row_notes) = row.developer_notes.as_deref() { + for note in row_notes.split("\n\n").map(str::trim) { + if !note.is_empty() && !canonical_components.contains(note) { + legacy_components.insert(note.to_string()); + } + } + } + } + let developer_notes = match (canonical_notes, legacy_components.is_empty()) { + (Some(notes), true) => Some(notes), + (Some(mut notes), false) => { + for component in legacy_components { + notes.push_str("\n\n"); + notes.push_str(&component); + } + Some(notes) + } + (None, false) => Some( + legacy_components + .into_iter() + .collect::>() + .join("\n\n"), + ), + (None, true) => None, + }; + (disposition, developer_notes) +} + +pub(super) async fn reconcile_legacy_feature_requests( + store: &dyn crate::storage::MetadataStore, + existing_rows: &[FeatureRequestRow], + tenant: &TenantId, + trajectory_entries: &[TrajectoryEntry], + stable_id: &str, + feature_request: &FeatureRequestRecord, +) -> Result<(), StatusCode> { + let category = format!("{:?}", feature_request.category); + let matching_rows = existing_rows + .iter() + .filter(|row| { + let is_owned_legacy = is_legacy_feature_request_id(&row.id) + && legacy_evidence_belongs_unambiguously_to_tenant(row, tenant, trajectory_entries); + (row.id == stable_id || is_owned_legacy) + && is_same_evidence_revision(row, &category, feature_request) + }) + .collect::>(); + let obsolete_ids = matching_rows + .iter() + .filter(|row| row.id != stable_id && is_legacy_feature_request_id(&row.id)) + .map(|row| row.id.clone()) + .collect::>(); + if obsolete_ids.is_empty() { + return Ok(()); + } + + let (disposition, developer_notes) = merged_review_state(&matching_rows, stable_id); + let canonical_updated = store + .update_feature_request(stable_id, &disposition, developer_notes.as_deref()) + .await + .map_err(|error| { + tracing::error!(error = %error, backend = store.backend_name(), feature_request_id = stable_id, "failed to preserve legacy feature-request review state"); + StatusCode::INTERNAL_SERVER_ERROR + })?; + if !canonical_updated { + tracing::error!( + backend = store.backend_name(), + feature_request_id = stable_id, + "canonical feature request disappeared during reconciliation" + ); + return Err(StatusCode::INTERNAL_SERVER_ERROR); + } + for obsolete_id in &obsolete_ids { + store + .delete_feature_request(obsolete_id) + .await + .map_err(|error| { + tracing::error!(error = %error, backend = store.backend_name(), feature_request_id = obsolete_id, "failed to remove reconciled legacy feature request"); + StatusCode::INTERNAL_SERVER_ERROR + })?; + } + tracing::info!( + feature_request_id = stable_id, + reconciled_count = obsolete_ids.len(), + "reconciled legacy feature-request projections" + ); + Ok(()) +} diff --git a/crates/temper-server/src/observe/evolution/operations/support.rs b/crates/temper-server/src/observe/evolution/operations/support.rs index 81c84e52b..40a8de2a5 100644 --- a/crates/temper-server/src/observe/evolution/operations/support.rs +++ b/crates/temper-server/src/observe/evolution/operations/support.rs @@ -109,6 +109,29 @@ pub(super) async fn dispatch_system_action( .await } +pub(super) async fn dispatch_system_action_idempotent( + state: &ServerState, + entity_type: &str, + entity_id: &str, + action: &str, + params: serde_json::Value, + idempotency_key: &str, +) -> Result { + let system_tenant = TenantId::new("temper-system"); + let mut agent_ctx = AgentContext::for_service("evolution-engine"); + agent_ctx.idempotency_key = Some(idempotency_key.to_string()); + state + .dispatch_tenant_action( + &system_tenant, + entity_type, + entity_id, + action, + params, + &agent_ctx, + ) + .await +} + pub(super) async fn dispatch_system_action_required( state: &ServerState, entity_type: &str, diff --git a/crates/temper-server/src/observe/mod_test.rs b/crates/temper-server/src/observe/mod_test.rs index b5bf6abd0..f15c5270d 100644 --- a/crates/temper-server/src/observe/mod_test.rs +++ b/crates/temper-server/src/observe/mod_test.rs @@ -20,7 +20,7 @@ use tracing_subscriber::registry::LookupSpan; use crate::registry::SpecRegistry; use crate::request_context::AgentContext; use crate::secrets::vault::SecretsVault; -use crate::state::TrajectoryEntry; +use crate::state::{TrajectoryEntry, TrajectorySource}; use crate::storage::StorageStack; const CSDL_XML: &str = include_str!("../../../../test-fixtures/specs/model.csdl.xml"); @@ -67,6 +67,51 @@ async fn test_state_with_turso() -> ServerState { state } +fn feature_request_db_url() -> String { + static COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + let id = COUNTER.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + format!( + "file:/tmp/temper-feature-request-test-{}-{}.db", + std::process::id(), + id, + ) +} + +async fn test_state_with_feature_request_runtime_at(db_url: &str) -> ServerState { + let order_csdl = parse_csdl(CSDL_XML).expect("order CSDL should parse"); + let platform_csdl_source = include_str!("../../../temper-platform/src/specs/model.csdl.xml"); + let platform_csdl = parse_csdl(platform_csdl_source).expect("platform CSDL should parse"); + let feature_request_ioa = + include_str!("../../../temper-platform/src/specs/FeatureRequest.ioa.toml"); + let mut registry = SpecRegistry::new(); + registry.register_tenant( + "default", + order_csdl, + CSDL_XML.to_string(), + &[("Order", ORDER_IOA)], + ); + registry.register_tenant( + "temper-system", + platform_csdl, + platform_csdl_source.to_string(), + &[("FeatureRequest", feature_request_ioa)], + ); + + let turso = TursoEventStore::new(db_url, None) + .await + .expect("create feature-request Turso db"); + let system = ActorSystem::new("test-feature-request-runtime"); + let mut state = ServerState::from_registry(system, registry); + state.set_storage_stack(StorageStack::from_turso(turso)); + state +} + +async fn test_state_with_feature_request_runtime() -> ServerState { + let db_url = feature_request_db_url(); + let _ = std::fs::remove_file(db_url.strip_prefix("file:").unwrap_or(&db_url)); + test_state_with_feature_request_runtime_at(&db_url).await +} + fn build_test_app() -> Router { let state = test_state_with_registry(); Router::new() @@ -1638,6 +1683,591 @@ async fn test_intent_evidence_returns_richer_intent_candidates() { assert_eq!(json["workaround_patterns"][0]["occurrences"], 1); } +async fn persist_feature_request_evidence(state: &ServerState, index: i64) { + persist_feature_request_evidence_for_action(state, index, "GenerateReport").await; +} + +async fn persist_feature_request_evidence_for_action( + state: &ServerState, + index: i64, + action: &str, +) { + let timestamp = (sim_now() + chrono::Duration::seconds(index)).to_rfc3339(); + persist_feature_request_evidence_for_action_at(state, index, action, ×tamp).await; +} + +async fn persist_feature_request_evidence_for_action_at( + state: &ServerState, + index: i64, + action: &str, + timestamp: &str, +) { + persist_feature_request_evidence_for_tenant_action_at( + state, "default", index, action, timestamp, + ) + .await; +} + +async fn persist_feature_request_evidence_for_tenant_action_at( + state: &ServerState, + tenant: &str, + index: i64, + action: &str, + timestamp: &str, +) { + state + .persist_trajectory_entry(&TrajectoryEntry { + timestamp: timestamp.to_string(), + tenant: tenant.to_string(), + entity_type: "MissingCapability".to_string(), + entity_id: format!("missing-{index}"), + action: action.to_string(), + success: false, + from_status: None, + to_status: None, + error: Some("EntitySetNotFound: Report".to_string()), + agent_id: Some("agent-1".to_string()), + session_id: Some("session-1".to_string()), + authz_denied: None, + denied_resource: None, + denied_module: None, + source: Some(TrajectorySource::Platform), + spec_governed: Some(true), + agent_type: Some("swe".to_string()), + request_body: None, + intent: Some("Generate a report".to_string()), + matched_policy_ids: None, + }) + .await + .expect("persist trajectory evidence"); +} + +#[test] +fn feature_request_identity_is_stable_across_replay_and_explicitly_versioned() { + let entries = (0..3) + .map(|index| TrajectoryEntry { + timestamp: (sim_now() + chrono::Duration::seconds(index)).to_rfc3339(), + tenant: "default".to_string(), + entity_type: "MissingCapability".to_string(), + entity_id: format!("missing-{index}"), + action: "GenerateReport".to_string(), + success: false, + from_status: None, + to_status: None, + error: Some(format!("EntitySetNotFound: Report variant {index}")), + agent_id: Some("agent-1".to_string()), + session_id: Some("session-1".to_string()), + authz_denied: None, + denied_resource: None, + denied_module: None, + source: Some(TrajectorySource::Platform), + spec_governed: Some(true), + agent_type: Some("swe".to_string()), + request_body: None, + intent: Some("Generate a report".to_string()), + matched_policy_ids: None, + }) + .collect::>(); + let generated = evolution::insight_generator::generate_feature_requests(&entries); + let original = generated.first().expect("feature request at threshold"); + let mut reordered_entries = entries; + reordered_entries.reverse(); + let reordered = evolution::insight_generator::generate_feature_requests(&reordered_entries); + let replayed = reordered + .first() + .expect("reordered feature request at threshold"); + let tenant = TenantId::default(); + + let original_id = evolution::stable_feature_request_id(&tenant, "v1", original); + let replayed_id = evolution::stable_feature_request_id(&tenant, "v1", replayed); + let revised_model_id = evolution::stable_feature_request_id(&tenant, "v2", original); + + assert_eq!( + original_id, replayed_id, + "evidence order must be normalized" + ); + assert_eq!( + original.description, replayed.description, + "generated content must be independent of evidence arrival order", + ); + assert_ne!( + original_id, revised_model_id, + "generator version changes must create explicit revisions", + ); +} + +#[tokio::test] +async fn feature_request_get_is_a_pure_read() { + let state = test_state_with_turso().await; + for index in 0..3 { + persist_feature_request_evidence(&state, index).await; + } + + let store = state + .platform_metadata_store() + .expect("Turso metadata store"); + let app = build_app_with_state(state); + + let first = observe_json(app.clone(), "/observe/evolution/feature-requests").await; + let second = observe_json(app.clone(), "/observe/evolution/feature-requests").await; + let (concurrent_a, concurrent_b) = tokio::join!( + observe_json(app.clone(), "/observe/evolution/feature-requests"), + observe_json(app, "/observe/evolution/feature-requests"), + ); + + assert_eq!( + first["total"], 0, + "GET must not materialize feature requests" + ); + assert_eq!( + second["total"], 0, + "repeated GET must remain side-effect free" + ); + assert_eq!(concurrent_a["total"], 0); + assert_eq!(concurrent_b["total"], 0); + assert!( + store + .list_feature_requests(None) + .await + .expect("list persisted feature requests") + .is_empty(), + "GET must not write the feature-request projection", + ); +} + +#[tokio::test] +async fn sentinel_materializes_feature_requests_idempotently_by_evidence_revision() { + let state = test_state_with_feature_request_runtime().await; + for index in 0..3 { + persist_feature_request_evidence(&state, index).await; + } + let store = state + .platform_metadata_store() + .expect("Turso metadata store"); + let app = build_app_with_state(state.clone()); + let generated = evolution::insight_generator::generate_feature_requests( + &state.load_trajectory_entries(100).await, + ); + let legacy = generated.first().expect("feature request at threshold"); + let legacy_id = "FR-2026-0123456789ab"; + store + .upsert_feature_request( + legacy_id, + &format!("{:?}", legacy.category), + &legacy.description, + legacy.frequency as i64, + &serde_json::to_string(&legacy.trajectory_refs).expect("serialize legacy refs"), + "Open", + None, + ) + .await + .expect("seed legacy GET-created projection"); + store + .update_feature_request(legacy_id, "Planned", Some("Reviewed before upgrade")) + .await + .expect("seed human review on legacy projection"); + + let (first, concurrent_retry) = tokio::join!( + app.clone() + .oneshot(system_post("/api/evolution/sentinel/check", "")), + app.clone() + .oneshot(system_post("/api/evolution/sentinel/check", "")), + ); + let first = first.expect("first sentinel request"); + let concurrent_retry = concurrent_retry.expect("concurrent sentinel retry"); + assert_eq!(first.status(), StatusCode::OK); + assert_eq!(concurrent_retry.status(), StatusCode::OK); + let first_json: serde_json::Value = serde_json::from_slice( + &axum::body::to_bytes(first.into_body(), 1024 * 1024) + .await + .expect("first sentinel body"), + ) + .expect("first sentinel JSON"); + let stable_id = first_json["feature_request_ids"][0] + .as_str() + .expect("stable feature-request id") + .to_string(); + let concurrent_json: serde_json::Value = serde_json::from_slice( + &axum::body::to_bytes(concurrent_retry.into_body(), 1024 * 1024) + .await + .expect("concurrent sentinel body"), + ) + .expect("concurrent sentinel JSON"); + assert_eq!(concurrent_json["feature_request_ids"][0], stable_id); + let first_entity = state + .get_tenant_entity_state( + &TenantId::new("temper-system"), + "FeatureRequest", + &stable_id, + ) + .await + .expect("first FeatureRequest entity"); + let first_event_count = first_entity.state.events.len(); + let creation_event_count = first_entity + .state + .events + .iter() + .filter(|event| event.action == "CreateFeatureRequest") + .count(); + assert_eq!( + creation_event_count, 1, + "concurrent materialization must append exactly one creation event", + ); + let migrated_rows = store + .list_feature_requests(None) + .await + .expect("list reconciled legacy projection"); + assert_eq!( + migrated_rows.len(), + 1, + "upgrade reconciliation must replace legacy duplicates with one canonical row", + ); + assert_eq!(migrated_rows[0].id, stable_id); + assert_eq!(migrated_rows[0].disposition, "Planned"); + assert_eq!( + migrated_rows[0].developer_notes.as_deref(), + Some("Reviewed before upgrade"), + "upgrade reconciliation must preserve human review state", + ); + store + .update_feature_request(&stable_id, "Planned", Some("Reviewed after migration")) + .await + .expect("record human feature-request review"); + + let second = app + .clone() + .oneshot(system_post("/api/evolution/sentinel/check", "")) + .await + .expect("second sentinel request"); + assert_eq!(second.status(), StatusCode::OK); + let second_json: serde_json::Value = serde_json::from_slice( + &axum::body::to_bytes(second.into_body(), 1024 * 1024) + .await + .expect("second sentinel body"), + ) + .expect("second sentinel JSON"); + assert_eq!(second_json["feature_request_ids"][0], stable_id); + let stable_rows = store + .list_feature_requests(None) + .await + .expect("list stable projection"); + assert_eq!(stable_rows.len(), 1, "repeat must keep one projection row"); + assert_eq!(stable_rows[0].disposition, "Planned"); + assert_eq!( + stable_rows[0].developer_notes.as_deref(), + Some("Reviewed after migration"), + "materialization must preserve mutable human review state", + ); + let stable_entity = state + .get_tenant_entity_state( + &TenantId::new("temper-system"), + "FeatureRequest", + &stable_id, + ) + .await + .expect("stable FeatureRequest entity"); + assert_eq!( + stable_entity.state.events.len(), + first_event_count, + "durable idempotency must suppress duplicate CreateFeatureRequest events", + ); + + persist_feature_request_evidence(&state, 3).await; + let revised = app + .oneshot(system_post("/api/evolution/sentinel/check", "")) + .await + .expect("revised sentinel request"); + assert_eq!(revised.status(), StatusCode::OK); + let revised_json: serde_json::Value = serde_json::from_slice( + &axum::body::to_bytes(revised.into_body(), 1024 * 1024) + .await + .expect("revised sentinel body"), + ) + .expect("revised sentinel JSON"); + assert_ne!(revised_json["feature_request_ids"][0], stable_id); + assert_eq!( + store + .list_feature_requests(None) + .await + .expect("list revised projection") + .len(), + 2, + "changed evidence must create an explicit new revision", + ); +} + +#[tokio::test] +async fn sentinel_reconciles_only_matching_legacy_evidence_and_deduplicates_notes() { + let state = test_state_with_feature_request_runtime().await; + for index in 0..3 { + let timestamp = (sim_now() + chrono::Duration::seconds(index)).to_rfc3339(); + persist_feature_request_evidence_for_action_at(&state, index, "GenerateReport", ×tamp) + .await; + persist_feature_request_evidence_for_action_at( + &state, + index, + "ExportDashboard", + ×tamp, + ) + .await; + } + let generated = evolution::insight_generator::generate_feature_requests( + &state.load_trajectory_entries(100).await, + ); + assert_eq!(generated.len(), 2); + let report = generated + .iter() + .find(|record| record.description.contains("'GenerateReport'")) + .expect("report feature request"); + let dashboard = generated + .iter() + .find(|record| record.description.contains("'ExportDashboard'")) + .expect("dashboard feature request"); + assert_eq!( + report.trajectory_refs, dashboard.trajectory_refs, + "regression requires both actions to share the same evidence timestamps", + ); + + let store = state + .platform_metadata_store() + .expect("Turso metadata store"); + let report_id = evolution::stable_feature_request_id(&TenantId::default(), "v1", report); + let seeds = [ + ( + report_id.as_str(), + report, + "Planned", + Some(" Report note B \n\nReport note A"), + ), + ( + "FR-2026-222222222222", + report, + "Planned", + Some("Report note B"), + ), + ( + "FR-2026-333333333333", + dashboard, + "Acknowledged", + Some("Dashboard note"), + ), + ]; + for (id, record, disposition, notes) in seeds { + store + .upsert_feature_request( + id, + &format!("{:?}", record.category), + &record.description, + record.frequency as i64, + &serde_json::to_string(&record.trajectory_refs).expect("serialize legacy refs"), + disposition, + notes, + ) + .await + .expect("seed feature-request projection"); + } + + let response = build_app_with_state(state) + .oneshot(system_post("/api/evolution/sentinel/check", "")) + .await + .expect("sentinel reconciliation request"); + assert_eq!(response.status(), StatusCode::OK); + + let rows = store + .list_feature_requests(None) + .await + .expect("list reconciled feature requests"); + assert_eq!(rows.len(), 2, "each action must retain one canonical row"); + assert!( + rows.iter().all(|row| !row.id.starts_with("FR-2026-")), + "all legacy projections must be removed", + ); + let report_row = rows + .iter() + .find(|row| row.description.contains("'GenerateReport'")) + .expect("canonical report row"); + assert_eq!(report_row.disposition, "Planned"); + assert_eq!( + report_row.developer_notes.as_deref(), + Some(" Report note B \n\nReport note A"), + "retrying after one legacy row was deleted must preserve canonical note bytes", + ); + let dashboard_row = rows + .iter() + .find(|row| row.description.contains("'ExportDashboard'")) + .expect("canonical dashboard row"); + assert_eq!(dashboard_row.disposition, "Acknowledged"); + assert_eq!( + dashboard_row.developer_notes.as_deref(), + Some("Dashboard note"), + "same-timestamp evidence from another action must not contaminate review state", + ); +} + +#[tokio::test] +async fn sentinel_does_not_claim_tenant_ambiguous_legacy_feature_requests() { + let state = test_state_with_feature_request_runtime().await; + for index in 0..3 { + let timestamp = (sim_now() + chrono::Duration::seconds(index)).to_rfc3339(); + for tenant in ["tenant-a", "tenant-b"] { + persist_feature_request_evidence_for_tenant_action_at( + &state, + tenant, + index, + "GenerateReport", + ×tamp, + ) + .await; + } + } + let entries = state.load_trajectory_entries(100).await; + let tenant_a_entries = entries + .iter() + .filter(|entry| entry.tenant == "tenant-a") + .cloned() + .collect::>(); + let generated = evolution::insight_generator::generate_feature_requests(&tenant_a_entries); + let legacy = generated.first().expect("tenant feature request"); + let store = state + .platform_metadata_store() + .expect("Turso metadata store"); + for (legacy_id, notes) in [ + ("FR-2026-aaaaaaaaaaaa", "Tenant A review"), + ("FR-2026-bbbbbbbbbbbb", "Tenant B review"), + ] { + store + .upsert_feature_request( + legacy_id, + &format!("{:?}", legacy.category), + &legacy.description, + legacy.frequency as i64, + &serde_json::to_string(&legacy.trajectory_refs).expect("serialize legacy refs"), + "Planned", + Some(notes), + ) + .await + .expect("seed tenant-ambiguous legacy projection"); + } + + for tenant in ["tenant-a", "tenant-b"] { + let ids = evolution::materialize_feature_requests_for_test( + &state, + &TenantId::new(tenant), + &entries, + ) + .await + .expect("tenant feature-request materialization"); + assert_eq!(ids.len(), 1); + } + + let rows = store + .list_feature_requests(None) + .await + .expect("list tenant-isolated feature requests"); + assert_eq!( + rows.len(), + 4, + "each tenant gets a canonical row while both ambiguous legacy rows remain", + ); + for (legacy_id, notes) in [ + ("FR-2026-aaaaaaaaaaaa", "Tenant A review"), + ("FR-2026-bbbbbbbbbbbb", "Tenant B review"), + ] { + let row = rows + .iter() + .find(|row| row.id == legacy_id) + .expect("ambiguous legacy row must remain"); + assert_eq!(row.disposition, "Planned"); + assert_eq!(row.developer_notes.as_deref(), Some(notes)); + } + let canonical_rows = rows + .iter() + .filter(|row| !row.id.starts_with("FR-2026-")) + .collect::>(); + assert_eq!(canonical_rows.len(), 2); + assert!( + canonical_rows + .iter() + .all(|row| row.disposition == "Open" && row.developer_notes.is_none()), + "neither tenant may consume review state from an ambiguous legacy row", + ); +} + +#[tokio::test] +async fn sentinel_retry_after_restart_reuses_the_durable_feature_request() { + let db_url = feature_request_db_url(); + let _ = std::fs::remove_file(db_url.strip_prefix("file:").unwrap_or(&db_url)); + + let (stable_id, event_count) = { + let state = test_state_with_feature_request_runtime_at(&db_url).await; + for index in 0..3 { + persist_feature_request_evidence(&state, index).await; + } + let response = build_app_with_state(state.clone()) + .oneshot(system_post("/api/evolution/sentinel/check", "")) + .await + .expect("sentinel request before restart"); + assert_eq!(response.status(), StatusCode::OK); + let json: serde_json::Value = serde_json::from_slice( + &axum::body::to_bytes(response.into_body(), 1024 * 1024) + .await + .expect("sentinel body before restart"), + ) + .expect("sentinel JSON before restart"); + let stable_id = json["feature_request_ids"][0] + .as_str() + .expect("stable feature-request id") + .to_string(); + let entity = state + .get_tenant_entity_state( + &TenantId::new("temper-system"), + "FeatureRequest", + &stable_id, + ) + .await + .expect("FeatureRequest before restart"); + (stable_id, entity.state.events.len()) + }; + + let restarted_state = test_state_with_feature_request_runtime_at(&db_url).await; + let response = build_app_with_state(restarted_state.clone()) + .oneshot(system_post("/api/evolution/sentinel/check", "")) + .await + .expect("sentinel retry after restart"); + assert_eq!(response.status(), StatusCode::OK); + let json: serde_json::Value = serde_json::from_slice( + &axum::body::to_bytes(response.into_body(), 1024 * 1024) + .await + .expect("sentinel body after restart"), + ) + .expect("sentinel JSON after restart"); + assert_eq!(json["feature_request_ids"][0], stable_id); + let rehydrated = restarted_state + .get_tenant_entity_state( + &TenantId::new("temper-system"), + "FeatureRequest", + &stable_id, + ) + .await + .expect("rehydrated FeatureRequest"); + assert_eq!( + rehydrated.state.events.len(), + event_count, + "restart retry must not append a duplicate creation event", + ); + assert_eq!( + restarted_state + .platform_metadata_store() + .expect("Turso metadata store") + .list_feature_requests(None) + .await + .expect("list projections after restart") + .len(), + 1, + "restart retry must retain one projection row", + ); +} + // -- Sentinel endpoint tests -- #[tokio::test] diff --git a/crates/temper-server/src/storage/evolution.rs b/crates/temper-server/src/storage/evolution.rs new file mode 100644 index 000000000..840690582 --- /dev/null +++ b/crates/temper-server/src/storage/evolution.rs @@ -0,0 +1,224 @@ +//! Evolution-engine durable metadata boundary. + +use temper_runtime::persistence::PersistenceError; +use temper_store_postgres::PostgresEventStore; +use temper_store_turso::{EvolutionRecordRow, FeatureRequestRow, TursoEventStore}; + +use super::{pg_evolution_record_to_turso, pg_feature_request_to_turso}; + +/// Evolution engine durable metadata capability. +#[async_trait::async_trait] +pub trait EvolutionStore: Send + Sync { + #[allow(clippy::too_many_arguments)] + async fn upsert_feature_request( + &self, + id: &str, + category: &str, + description: &str, + frequency: i64, + trajectory_refs_json: &str, + disposition: &str, + developer_notes: Option<&str>, + ) -> Result<(), PersistenceError>; + + async fn list_feature_requests( + &self, + disposition: Option<&str>, + ) -> Result, PersistenceError>; + + async fn update_feature_request( + &self, + id: &str, + disposition: &str, + developer_notes: Option<&str>, + ) -> Result; + + async fn delete_feature_request(&self, id: &str) -> Result; + + async fn insert_evolution_record( + &self, + id: &str, + record_type: &str, + status: &str, + created_by: &str, + derived_from: Option<&str>, + data_json: &str, + ) -> Result<(), PersistenceError>; + + async fn get_evolution_record( + &self, + id: &str, + ) -> Result, PersistenceError>; + + async fn list_evolution_records( + &self, + record_type: Option<&str>, + status: Option<&str>, + ) -> Result, PersistenceError>; + + async fn list_ranked_insights(&self) -> Result, PersistenceError>; +} + +#[async_trait::async_trait] +impl EvolutionStore for PostgresEventStore { + async fn upsert_feature_request( + &self, + id: &str, + category: &str, + description: &str, + frequency: i64, + trajectory_refs_json: &str, + disposition: &str, + developer_notes: Option<&str>, + ) -> Result<(), PersistenceError> { + self.upsert_feature_request( + id, + category, + description, + frequency, + trajectory_refs_json, + disposition, + developer_notes, + ) + .await + } + + async fn list_feature_requests( + &self, + disposition: Option<&str>, + ) -> Result, PersistenceError> { + self.list_feature_requests(disposition) + .await + .map(|rows| rows.into_iter().map(pg_feature_request_to_turso).collect()) + } + + async fn update_feature_request( + &self, + id: &str, + disposition: &str, + developer_notes: Option<&str>, + ) -> Result { + self.update_feature_request(id, disposition, developer_notes) + .await + } + + async fn delete_feature_request(&self, id: &str) -> Result { + self.delete_feature_request(id).await + } + + async fn insert_evolution_record( + &self, + id: &str, + record_type: &str, + status: &str, + created_by: &str, + derived_from: Option<&str>, + data_json: &str, + ) -> Result<(), PersistenceError> { + self.insert_evolution_record(id, record_type, status, created_by, derived_from, data_json) + .await + } + + async fn get_evolution_record( + &self, + id: &str, + ) -> Result, PersistenceError> { + self.get_evolution_record(id) + .await + .map(|row| row.map(pg_evolution_record_to_turso)) + } + + async fn list_evolution_records( + &self, + record_type: Option<&str>, + status: Option<&str>, + ) -> Result, PersistenceError> { + self.list_evolution_records(record_type, status) + .await + .map(|rows| rows.into_iter().map(pg_evolution_record_to_turso).collect()) + } + + async fn list_ranked_insights(&self) -> Result, PersistenceError> { + self.list_ranked_insights() + .await + .map(|rows| rows.into_iter().map(pg_evolution_record_to_turso).collect()) + } +} + +#[async_trait::async_trait] +impl EvolutionStore for TursoEventStore { + async fn upsert_feature_request( + &self, + id: &str, + category: &str, + description: &str, + frequency: i64, + trajectory_refs_json: &str, + disposition: &str, + developer_notes: Option<&str>, + ) -> Result<(), PersistenceError> { + self.upsert_feature_request( + id, + category, + description, + frequency, + trajectory_refs_json, + disposition, + developer_notes, + ) + .await + } + + async fn list_feature_requests( + &self, + disposition: Option<&str>, + ) -> Result, PersistenceError> { + self.list_feature_requests(disposition).await + } + + async fn update_feature_request( + &self, + id: &str, + disposition: &str, + developer_notes: Option<&str>, + ) -> Result { + self.update_feature_request(id, disposition, developer_notes) + .await + } + + async fn delete_feature_request(&self, id: &str) -> Result { + self.delete_feature_request(id).await + } + + async fn insert_evolution_record( + &self, + id: &str, + record_type: &str, + status: &str, + created_by: &str, + derived_from: Option<&str>, + data_json: &str, + ) -> Result<(), PersistenceError> { + self.insert_evolution_record(id, record_type, status, created_by, derived_from, data_json) + .await + } + + async fn get_evolution_record( + &self, + id: &str, + ) -> Result, PersistenceError> { + self.get_evolution_record(id).await + } + + async fn list_evolution_records( + &self, + record_type: Option<&str>, + status: Option<&str>, + ) -> Result, PersistenceError> { + self.list_evolution_records(record_type, status).await + } + + async fn list_ranked_insights(&self) -> Result, PersistenceError> { + self.list_ranked_insights().await + } +} diff --git a/crates/temper-server/src/storage/mod.rs b/crates/temper-server/src/storage/mod.rs index f689b7db0..28e19d9df 100644 --- a/crates/temper-server/src/storage/mod.rs +++ b/crates/temper-server/src/storage/mod.rs @@ -34,9 +34,11 @@ use crate::platform_store::PlatformStore; use crate::platform_store::SimPlatformStore; use crate::state::trajectory::{TrajectoryEntry, TrajectorySource}; +mod evolution; mod published_artifacts; mod query_plane_impls; mod query_plane_read; +pub use evolution::EvolutionStore; pub use published_artifacts::{ PublishedArtifactStore, PublishedArtifactStoreRow, PublishedArtifactStoreUpsert, }; @@ -861,57 +863,6 @@ pub trait ObserveReadStore: Send + Sync { ) -> Result, PersistenceError>; } -/// Evolution engine durable metadata capability. -#[async_trait::async_trait] -pub trait EvolutionStore: Send + Sync { - #[allow(clippy::too_many_arguments)] - async fn upsert_feature_request( - &self, - id: &str, - category: &str, - description: &str, - frequency: i64, - trajectory_refs_json: &str, - disposition: &str, - developer_notes: Option<&str>, - ) -> Result<(), PersistenceError>; - - async fn list_feature_requests( - &self, - disposition: Option<&str>, - ) -> Result, PersistenceError>; - - async fn update_feature_request( - &self, - id: &str, - disposition: &str, - developer_notes: Option<&str>, - ) -> Result; - - async fn insert_evolution_record( - &self, - id: &str, - record_type: &str, - status: &str, - created_by: &str, - derived_from: Option<&str>, - data_json: &str, - ) -> Result<(), PersistenceError>; - - async fn get_evolution_record( - &self, - id: &str, - ) -> Result, PersistenceError>; - - async fn list_evolution_records( - &self, - record_type: Option<&str>, - status: Option<&str>, - ) -> Result, PersistenceError>; - - async fn list_ranked_insights(&self) -> Result, PersistenceError>; -} - /// Design-time verification event capability. #[async_trait::async_trait] pub trait DesignTimeEventStore: Send + Sync { @@ -1856,162 +1807,6 @@ impl ObserveReadStore for TursoEventStore { } } -#[async_trait::async_trait] -impl EvolutionStore for PostgresEventStore { - async fn upsert_feature_request( - &self, - id: &str, - category: &str, - description: &str, - frequency: i64, - trajectory_refs_json: &str, - disposition: &str, - developer_notes: Option<&str>, - ) -> Result<(), PersistenceError> { - self.upsert_feature_request( - id, - category, - description, - frequency, - trajectory_refs_json, - disposition, - developer_notes, - ) - .await - } - - async fn list_feature_requests( - &self, - disposition: Option<&str>, - ) -> Result, PersistenceError> { - self.list_feature_requests(disposition) - .await - .map(|rows| rows.into_iter().map(pg_feature_request_to_turso).collect()) - } - - async fn update_feature_request( - &self, - id: &str, - disposition: &str, - developer_notes: Option<&str>, - ) -> Result { - self.update_feature_request(id, disposition, developer_notes) - .await - } - - async fn insert_evolution_record( - &self, - id: &str, - record_type: &str, - status: &str, - created_by: &str, - derived_from: Option<&str>, - data_json: &str, - ) -> Result<(), PersistenceError> { - self.insert_evolution_record(id, record_type, status, created_by, derived_from, data_json) - .await - } - - async fn get_evolution_record( - &self, - id: &str, - ) -> Result, PersistenceError> { - self.get_evolution_record(id) - .await - .map(|row| row.map(pg_evolution_record_to_turso)) - } - - async fn list_evolution_records( - &self, - record_type: Option<&str>, - status: Option<&str>, - ) -> Result, PersistenceError> { - self.list_evolution_records(record_type, status) - .await - .map(|rows| rows.into_iter().map(pg_evolution_record_to_turso).collect()) - } - - async fn list_ranked_insights(&self) -> Result, PersistenceError> { - self.list_ranked_insights() - .await - .map(|rows| rows.into_iter().map(pg_evolution_record_to_turso).collect()) - } -} - -#[async_trait::async_trait] -impl EvolutionStore for TursoEventStore { - async fn upsert_feature_request( - &self, - id: &str, - category: &str, - description: &str, - frequency: i64, - trajectory_refs_json: &str, - disposition: &str, - developer_notes: Option<&str>, - ) -> Result<(), PersistenceError> { - self.upsert_feature_request( - id, - category, - description, - frequency, - trajectory_refs_json, - disposition, - developer_notes, - ) - .await - } - - async fn list_feature_requests( - &self, - disposition: Option<&str>, - ) -> Result, PersistenceError> { - self.list_feature_requests(disposition).await - } - - async fn update_feature_request( - &self, - id: &str, - disposition: &str, - developer_notes: Option<&str>, - ) -> Result { - self.update_feature_request(id, disposition, developer_notes) - .await - } - - async fn insert_evolution_record( - &self, - id: &str, - record_type: &str, - status: &str, - created_by: &str, - derived_from: Option<&str>, - data_json: &str, - ) -> Result<(), PersistenceError> { - self.insert_evolution_record(id, record_type, status, created_by, derived_from, data_json) - .await - } - - async fn get_evolution_record( - &self, - id: &str, - ) -> Result, PersistenceError> { - self.get_evolution_record(id).await - } - - async fn list_evolution_records( - &self, - record_type: Option<&str>, - status: Option<&str>, - ) -> Result, PersistenceError> { - self.list_evolution_records(record_type, status).await - } - - async fn list_ranked_insights(&self) -> Result, PersistenceError> { - self.list_ranked_insights().await - } -} - #[async_trait::async_trait] impl DesignTimeEventStore for PostgresEventStore { async fn insert_design_time_event( diff --git a/crates/temper-store-postgres/src/platform.rs b/crates/temper-store-postgres/src/platform.rs index 322e61fe9..f7599c3bb 100644 --- a/crates/temper-store-postgres/src/platform.rs +++ b/crates/temper-store-postgres/src/platform.rs @@ -1775,6 +1775,7 @@ impl PostgresEventStore { Ok(rows.into_iter().map(row_to_agent_summary).collect()) } + /// Upsert generated feature-request fields while preserving existing human review state. #[allow(clippy::too_many_arguments)] pub async fn upsert_feature_request( &self, @@ -1793,8 +1794,7 @@ impl PostgresEventStore { VALUES ($1, $2, $3, $4, $5, $6, $7, now()) \ ON CONFLICT (id) DO UPDATE SET \ category = EXCLUDED.category, description = EXCLUDED.description, frequency = EXCLUDED.frequency, \ - trajectory_refs = EXCLUDED.trajectory_refs, disposition = EXCLUDED.disposition, \ - developer_notes = EXCLUDED.developer_notes, updated_at = now()", + trajectory_refs = EXCLUDED.trajectory_refs, updated_at = now()", ) .bind(id) .bind(category) @@ -1845,6 +1845,16 @@ impl PostgresEventStore { Ok(result.rows_affected() > 0) } + /// Delete an obsolete feature-request projection after canonical reconciliation. + pub async fn delete_feature_request(&self, id: &str) -> Result { + let result = crate::dbm::postgres_query!("DELETE FROM feature_requests WHERE id = $1") + .bind(id) + .execute(self.pool()) + .await + .map_err(storage_error)?; + Ok(result.rows_affected() > 0) + } + pub async fn insert_evolution_record( &self, id: &str, diff --git a/crates/temper-store-turso/src/store/evolution.rs b/crates/temper-store-turso/src/store/evolution.rs index ee220eb1c..960538e48 100644 --- a/crates/temper-store-turso/src/store/evolution.rs +++ b/crates/temper-store-turso/src/store/evolution.rs @@ -12,7 +12,7 @@ use crate::metrics::TursoQueryTimer; // ----------------------------------------------------------------------- impl TursoEventStore { - /// Upsert a feature request. + /// Upsert generated feature-request fields while preserving existing human review state. #[allow(clippy::too_many_arguments)] #[instrument(skip_all, fields(id, otel.name = "turso.upsert_feature_request"))] pub async fn upsert_feature_request( @@ -32,7 +32,7 @@ impl TursoEventStore { VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, datetime('now')) \ ON CONFLICT(id) DO UPDATE SET \ category = ?2, description = ?3, frequency = ?4, trajectory_refs = ?5, \ - disposition = ?6, developer_notes = ?7, updated_at = datetime('now')", + updated_at = datetime('now')", params![id, category, description, frequency, trajectory_refs_json, disposition, developer_notes], ) .await @@ -97,6 +97,18 @@ impl TursoEventStore { Ok(affected > 0) } + /// Delete an obsolete feature-request projection after canonical reconciliation. + #[instrument(skip_all, fields(id, otel.name = "turso.delete_feature_request"))] + pub async fn delete_feature_request(&self, id: &str) -> Result { + let _query_timer = TursoQueryTimer::start("turso.delete_feature_request"); + let conn = self.configured_connection().await?; + let affected = conn + .execute("DELETE FROM feature_requests WHERE id = ?1", params![id]) + .await + .map_err(storage_error)?; + Ok(affected > 0) + } + // ----------------------------------------------------------------------- // Evolution record CRUD // ----------------------------------------------------------------------- diff --git a/docs/adrs/0175-feature-request-get-idempotency.md b/docs/adrs/0175-feature-request-get-idempotency.md new file mode 100644 index 000000000..5ada19fa8 --- /dev/null +++ b/docs/adrs/0175-feature-request-get-idempotency.md @@ -0,0 +1,114 @@ +# ADR-0175: Feature-Request GET Is a Pure Read + +- Status: Accepted +- Date: 2026-07-14 +- Deciders: Temper core maintainers +- Related: + - Linear: ARN-240 + - ADR-0025: Evolution records as entities + - ADR-0013: Evolution loop agent integration + - `crates/temper-server/src/observe/evolution/operations.rs` + - `crates/temper-server/src/observe/evolution/operations/reconcile.rs` + - `crates/temper-store-turso/src/store/evolution.rs` + - `crates/temper-store-postgres/src/platform.rs` + +## Context + +`GET /observe/evolution/feature-requests` was state-changing. On every read it: + +1. Regenerated feature requests from trajectory evidence. +2. Upserted the legacy feature-request projection. +3. Created a new IOA `FeatureRequest` entity with `next_system_entity_id("FR")`. + +Polling, browser refresh, caches, and retries therefore materialize duplicate identities and widen dual-store divergence. A nominally safe HTTP GET was neither idempotent nor side-effect-free. + +## Decision + +### Sub-Decision 1: GET is a bounded pure query + +`handle_feature_requests` only authorizes and lists durable projections. It does not load trajectories for generation, does not upsert metadata, and does not create system entities. + +**Why this approach**: HTTP GET must be safe for observers. Materialization belongs on an explicit write path. + +### Sub-Decision 2: Materialization is an authorized write command + +Feature-request generation and durable projection run from the evolution write path (`POST /api/evolution/sentinel/check` today; any future explicit materialize command follows the same contract). The GET surface remains read-only even if generation later moves to a durable worker. + +### Sub-Decision 3: Stable identity from tenant + evidence + generator version + +Entity identity is derived, not allocated: + +```text +stable_id = "FR-" || sha256(json({ + tenant, + generator_version, + category, + description, + frequency, + trajectory_refs (sorted) +})) +``` + +Materialization dispatches `CreateFeatureRequest` with an idempotency key `feature-request:{stable_id}` so concurrent and retry writes converge on one entity. + +A deliberate generator/evidence revision changes the hash and creates a new identity (explicit version semantics). Order of evidence arrival must not. + +### Sub-Decision 4: Projection upsert preserves human review state + +Metadata upserts update generated fields (`category`, `description`, `frequency`, `trajectory_refs`) only. `disposition` and `developer_notes` are human-owned and are never clobbered by regeneration. + +### Sub-Decision 5: Legacy projection reconciliation is tenant-safe + +Legacy rows using the old `FR-YYYY-<12 hex>` id shape may be reconciled into the stable id when: + +- evidence revision matches (category, frequency, description prefix, sorted trajectory refs), and +- trajectory refs resolve unambiguously to a single tenant matching the materializing tenant. + +Ambiguous multi-tenant legacy rows are left alone. Review notes merge without dropping canonical note bytes; duplicate note components are deduplicated. + +## Rollout Plan + +1. **Phase 0 (This PR)** — Pure GET; stable idempotent materialization on sentinel; legacy reconcile; store delete + review-preserving upsert; regression tests. +2. **Phase 1 (Follow-up)** — Optional dedicated authorized materialize command / worker if generation should leave the sentinel hot path. +3. **Phase 2** — Drop dual projection once IOA entity store is sole source of truth (related: ARN-218). + +## Consequences + +### Positive + +- Repeated and concurrent GETs perform no writes. +- Same evidence revision materializes exactly one durable identity. +- Generator version changes create explicit new identities. +- Human disposition/notes survive rematerialization and upgrades. + +### Negative + +- Fresh feature requests no longer appear as a side effect of opening Observe; callers must run sentinel (or a future materialize command). +- Legacy ambiguous multi-tenant rows require manual cleanup if refs cannot prove ownership. + +### Risks + +- Callers that relied on GET to populate data will see empty lists until materialization runs. Mitigation: document the write path; sentinel already runs in the evolution loop. +- Hash identity couples to description text; model changes must bump `generator_version`. + +### DST Compliance + +- Stable ids use SHA-256 over a sorted deterministic JSON payload (no wall clock / random UUID on the write path for identity). +- Trajectory ref sorting uses deterministic order. +- Idempotent dispatch keys are pure functions of the stable id. + +## Non-Goals + +- Removing the legacy metadata projection entirely (dual-store cleanup remains related work). +- Changing FeatureRequest IOA state machine transitions or Observe UI beyond the read contract. +- Cross-tenant feature-request listing redesign. + +## Alternatives Considered + +1. **Keep generation on GET but cache by evidence hash** — Still makes GET a writer; races and dual-store growth remain. Rejected. +2. **Soft-delete duplicates after GET** — Leaves write-on-read and repair complexity. Rejected. +3. **UUID entity ids with unique constraint on evidence hash** — Workable, but content-addressed stable ids make retries and restarts simpler without a separate index. Prefer stable id as primary. + +## Rollback Policy + +Revert the PR: restore generation inside `handle_feature_requests` and the previous non-preserving upsert SQL. Durable rows written with stable ids remain valid entities; legacy reconcile deletes would need manual restore from backups if rolled back after production cleanup.