diff --git a/apps/elf-api/src/routes.rs b/apps/elf-api/src/routes.rs index a22920a6..4de227b7 100644 --- a/apps/elf-api/src/routes.rs +++ b/apps/elf-api/src/routes.rs @@ -55,12 +55,13 @@ use elf_service::{ KnowledgePageLintRequest, KnowledgePageLintResponse, KnowledgePageRebuildRequest, KnowledgePageRebuildResponse, KnowledgePageResponse, KnowledgePageSearchRequest, KnowledgePageSearchResponse, KnowledgePagesListRequest, KnowledgePagesListResponse, - ListRequest, ListResponse, NoteFetchRequest, NoteFetchResponse, NoteProvenanceBundleResponse, - NoteProvenanceGetRequest, PayloadLevel, PublishNoteRequest, QueryPlan, RankingRequestOverride, - RebuildReport, SearchDetailsRequest, SearchDetailsResult, SearchExplainRequest, - SearchExplainResponse, SearchIndexItem, SearchRequest, SearchResponse, SearchSessionGetRequest, - SearchTimelineGroup, SearchTimelineRequest, SearchTrajectoryResponse, SearchTrajectorySummary, - ShareScope, SpaceGrantRevokeRequest, SpaceGrantRevokeResponse, SpaceGrantUpsertRequest, + ListRequest, ListResponse, MemoryHistoryGetRequest, MemoryHistoryResponse, NoteFetchRequest, + NoteFetchResponse, NoteProvenanceBundleResponse, NoteProvenanceGetRequest, PayloadLevel, + PublishNoteRequest, QueryPlan, RankingRequestOverride, RebuildReport, SearchDetailsRequest, + SearchDetailsResult, SearchExplainRequest, SearchExplainResponse, SearchIndexItem, + SearchRequest, SearchResponse, SearchSessionGetRequest, SearchTimelineGroup, + SearchTimelineRequest, SearchTrajectoryResponse, SearchTrajectorySummary, ShareScope, + SpaceGrantRevokeRequest, SpaceGrantRevokeResponse, SpaceGrantUpsertRequest, SpaceGrantsListRequest, TextPositionSelector, TextQuoteSelector, TraceBundleGetRequest, TraceBundleResponse, TraceGetRequest, TraceGetResponse, TraceRecentListRequest, TraceRecentListResponse, TraceTrajectoryGetRequest, UnpublishNoteRequest, UpdateRequest, @@ -154,6 +155,7 @@ const VIEWER_HTML: &str = include_str!("../static/viewer.html"); admin_graph_predicate_alias_add, admin_graph_predicate_aliases_list, admin_note_provenance_get, + admin_note_history_get, ), components(schemas( AdminIngestionProfileDefaultResponseV2, @@ -707,6 +709,7 @@ pub fn admin_router(state: AppState) -> Router { routing::post(admin_graph_predicate_alias_add).get(admin_graph_predicate_aliases_list), ) .route("/v2/admin/notes/{note_id}/provenance", routing::get(admin_note_provenance_get)) + .route("/v2/admin/notes/{note_id}/history", routing::get(admin_note_history_get)) .with_state(state) .layer(DefaultBodyLimit::max(MAX_REQUEST_BYTES)) .layer(middleware::from_fn_with_state(auth_state, admin_auth_middleware)); @@ -2481,6 +2484,38 @@ async fn admin_note_provenance_get( Ok(Json(response)) } +#[utoipa::path( + get, + path = "/v2/admin/notes/{note_id}/history", + tag = "admin", + params(("note_id" = Uuid, Path, description = "Note ID.")), + responses( + (status = 200, description = "Memory history timeline.", body = Value), + (status = 400, description = "Invalid request.", body = ErrorBody), + (status = 401, description = "Authentication required.", body = ErrorBody), + (status = 403, description = "Admin access required.", body = ErrorBody), + (status = 404, description = "Note was not found.", body = ErrorBody), + (status = 500, description = "Internal error.", body = ErrorBody), + ) +)] +async fn admin_note_history_get( + State(state): State, + headers: HeaderMap, + Path(note_id): Path, +) -> Result, ApiError> { + let ctx = RequestContext::from_headers(&headers)?; + let response = state + .service + .memory_history_get(MemoryHistoryGetRequest { + tenant_id: ctx.tenant_id, + project_id: ctx.project_id, + note_id, + }) + .await?; + + Ok(Json(response)) +} + #[utoipa::path( post, path = "/v2/admin/consolidation/runs", diff --git a/apps/elf-api/tests/http.rs b/apps/elf-api/tests/http.rs index 6d894994..fe5a4d9d 100644 --- a/apps/elf-api/tests/http.rs +++ b/apps/elf-api/tests/http.rs @@ -2373,6 +2373,58 @@ async fn admin_note_provenance_includes_request_id_on_success() { test_db.cleanup().await.expect("Failed to cleanup test database."); } +#[tokio::test] +#[ignore = "Requires external Postgres and Qdrant. Set ELF_PG_DSN and ELF_QDRANT_GRPC_URL (or ELF_QDRANT_URL) to run."] +async fn admin_note_history_includes_request_id_on_success() { + let Some((test_db, qdrant_url, collection)) = test_env().await else { + return; + }; + let mut config = test_config(test_db.dsn().to_string(), qdrant_url, collection); + + config.security.auth_mode = "off".to_string(); + + let state = AppState::new(config).await.expect("Failed to initialize app state."); + let app = routes::admin_router(state.clone()); + let note_id = Uuid::new_v4(); + let request_id = Uuid::new_v4(); + + insert_note(&state, note_id, "agent_private", TEST_AGENT_A, "History integration test note.") + .await; + + let response = app + .oneshot( + Request::builder() + .uri(format!("/v2/admin/notes/{note_id}/history")) + .header("X-ELF-Tenant-Id", TEST_TENANT_ID) + .header("X-ELF-Project-Id", TEST_PROJECT_ID) + .header("X-ELF-Agent-Id", TEST_AGENT_A) + .header("X-ELF-Request-Id", request_id.to_string()) + .body(Body::empty()) + .expect("Failed to build history request."), + ) + .await + .expect("Failed to call admin note history."); + + assert_eq!(response.status(), StatusCode::OK); + + let expected_request_id = request_id.to_string(); + + assert_eq!( + response.headers().get("X-ELF-Request-Id").and_then(|value| value.to_str().ok()), + Some(expected_request_id.as_str()) + ); + + let body = body::to_bytes(response.into_body(), usize::MAX) + .await + .expect("Failed to read history response body."); + let json: serde_json::Value = serde_json::from_slice(&body).expect("Failed to parse response."); + + assert_eq!(json["schema"], "elf.memory_history/v1"); + assert_eq!(json["request_id"], request_id.to_string()); + + test_db.cleanup().await.expect("Failed to cleanup test database."); +} + #[tokio::test] #[ignore = "Requires external Postgres and Qdrant. Set ELF_PG_DSN and ELF_QDRANT_GRPC_URL (or ELF_QDRANT_URL) to run."] async fn admin_note_provenance_rejects_invalid_request_id_header() { diff --git a/apps/elf-eval/fixtures/real_world_memory/evolution/preference_changed_current_vs_historical.json b/apps/elf-eval/fixtures/real_world_memory/evolution/preference_changed_current_vs_historical.json index bf5e93c7..3e43dd25 100644 --- a/apps/elf-eval/fixtures/real_world_memory/evolution/preference_changed_current_vs_historical.json +++ b/apps/elf-eval/fixtures/real_world_memory/evolution/preference_changed_current_vs_historical.json @@ -212,6 +212,11 @@ "required": false, "encoded": false, "follow_up": null + }, + "history_readback": { + "encoded": true, + "required_event_types": ["add", "update", "ignore"], + "requires_note_version_links": true } }, "tags": [ diff --git a/apps/elf-eval/src/bin/real_world_job_benchmark.rs b/apps/elf-eval/src/bin/real_world_job_benchmark.rs index 9c41027f..1fd02874 100644 --- a/apps/elf-eval/src/bin/real_world_job_benchmark.rs +++ b/apps/elf-eval/src/bin/real_world_job_benchmark.rs @@ -298,6 +298,7 @@ struct MemoryEvolution { conflicts: Vec, update_rationale: Option, temporal_validity: Option, + history_readback: Option, } #[derive(Debug, Deserialize)] @@ -324,6 +325,14 @@ struct TemporalValidity { follow_up: Option, } +#[derive(Debug, Deserialize)] +struct HistoryReadback { + encoded: bool, + #[serde(default)] + required_event_types: Vec, + requires_note_version_links: bool, +} + #[derive(Debug, Deserialize)] struct ScoringRubric { #[serde(default)] @@ -763,6 +772,8 @@ struct ReportSummary { update_rationale_available_count: usize, #[serde(default)] temporal_validity_not_encoded_count: usize, + #[serde(default)] + history_readback_encoded_count: usize, expected_evidence_total: usize, expected_evidence_matched: usize, expected_evidence_recall: f64, @@ -865,6 +876,8 @@ struct SuiteReport { update_rationale_available_count: usize, #[serde(default)] temporal_validity_not_encoded_count: usize, + #[serde(default)] + history_readback_encoded_count: usize, expected_evidence_recall: Option, irrelevant_context_ratio: Option, trace_explainability_count: usize, @@ -896,6 +909,8 @@ struct JobReport { update_rationale_available: bool, #[serde(default)] temporal_validity_not_encoded: bool, + #[serde(default)] + history_readback_encoded: bool, retrieval_quality: RetrievalQualityReport, latency_ms: Option, cost: Option, @@ -1036,6 +1051,7 @@ struct EvolutionSummary { conflict_detection_count: usize, update_rationale_available_count: usize, temporal_validity_not_encoded_count: usize, + history_readback_encoded_count: usize, } #[derive(Clone, Debug, Deserialize, Serialize)] @@ -1050,6 +1066,9 @@ struct EvolutionJobReport { temporal_validity_required: bool, temporal_validity_encoded: bool, temporal_validity_not_encoded: bool, + history_readback_encoded: bool, + history_event_types: Vec, + history_requires_note_version_links: bool, #[serde(skip_serializing_if = "Option::is_none")] follow_up: Option, } @@ -2265,6 +2284,16 @@ fn evolution_job_report( let temporal_validity_encoded = evolution.temporal_validity.as_ref().is_some_and(|temporal| temporal.encoded); let temporal_validity_not_encoded = temporal_validity_required && !temporal_validity_encoded; + let history_readback_encoded = + evolution.history_readback.as_ref().is_some_and(|history| history.encoded); + let history_event_types = evolution + .history_readback + .as_ref() + .map_or_else(Vec::new, |history| history.required_event_types.clone()); + let history_requires_note_version_links = evolution + .history_readback + .as_ref() + .is_some_and(|history| history.requires_note_version_links); let follow_up = evolution .temporal_validity .as_ref() @@ -2282,6 +2311,9 @@ fn evolution_job_report( temporal_validity_required, temporal_validity_encoded, temporal_validity_not_encoded, + history_readback_encoded, + history_event_types, + history_requires_note_version_links, follow_up, }) } @@ -2783,6 +2815,10 @@ fn job_report(job: &RealWorldJob, scoring: JobScoring) -> JobReport { .evolution .as_ref() .is_some_and(|report| report.temporal_validity_not_encoded), + history_readback_encoded: scoring + .evolution + .as_ref() + .is_some_and(|report| report.history_readback_encoded), retrieval_quality, latency_ms: answer.latency_ms, cost: answer.cost.clone(), @@ -3101,6 +3137,7 @@ fn suite_report(suite_id: &str, jobs: &[JobReport]) -> SuiteReport { conflict_detection_count: 0, update_rationale_available_count: 0, temporal_validity_not_encoded_count: 0, + history_readback_encoded_count: 0, expected_evidence_recall: None, irrelevant_context_ratio: None, trace_explainability_count: 0, @@ -3118,6 +3155,8 @@ fn suite_report(suite_id: &str, jobs: &[JobReport]) -> SuiteReport { suite_jobs.iter().filter(|job| job.update_rationale_available).count(); let temporal_validity_not_encoded_count = suite_jobs.iter().filter(|job| job.temporal_validity_not_encoded).count(); + let history_readback_encoded_count = + suite_jobs.iter().filter(|job| job.history_readback_encoded).count(); let trace_explainability_count = suite_jobs.iter().filter(|job| job.trace_explainability.is_some()).count(); @@ -3132,6 +3171,7 @@ fn suite_report(suite_id: &str, jobs: &[JobReport]) -> SuiteReport { conflict_detection_count, update_rationale_available_count, temporal_validity_not_encoded_count, + history_readback_encoded_count, expected_evidence_recall: Some(expected_evidence_recall_for_jobs(&suite_jobs)), irrelevant_context_ratio: Some(irrelevant_context_ratio_for_jobs(&suite_jobs)), trace_explainability_count, @@ -3206,6 +3246,10 @@ fn report_summary(jobs: &[JobReport], suites: &[SuiteReport]) -> ReportSummary { .iter() .filter(|job| job.temporal_validity_not_encoded) .count(), + history_readback_encoded_count: jobs + .iter() + .filter(|job| job.history_readback_encoded) + .count(), expected_evidence_total: jobs .iter() .map(|job| job.retrieval_quality.expected_evidence_total) @@ -3302,6 +3346,10 @@ fn evolution_summary(jobs: &[JobReport]) -> EvolutionSummary { .iter() .filter(|job| job.temporal_validity_not_encoded) .count(), + history_readback_encoded_count: jobs + .iter() + .filter(|job| job.history_readback_encoded) + .count(), } } @@ -4028,6 +4076,10 @@ fn render_markdown_header(out: &mut String, report: &RealWorldReport, report_pat "- Temporal validity not encoded: `{}`\n", report.summary.temporal_validity_not_encoded_count )); + out.push_str(&format!( + "- History readback encoded: `{}`\n", + report.summary.history_readback_encoded_count + )); render_markdown_quality_summary(out, report); @@ -4131,13 +4183,13 @@ fn render_markdown_quality_summary(out: &mut String, report: &RealWorldReport) { fn render_markdown_suites(out: &mut String, report: &RealWorldReport) { out.push_str("## Suites\n\n"); out.push_str( - "| Suite | Status | Jobs | Score | Evidence Recall | Irrelevant Context | Trace Explain | Stale Answers | Conflicts | Update Rationales | Temporal Gaps | Unsupported Claims | Wrong Results | Reason |\n", + "| Suite | Status | Jobs | Score | Evidence Recall | Irrelevant Context | Trace Explain | Stale Answers | Conflicts | Update Rationales | Temporal Gaps | History Readback | Unsupported Claims | Wrong Results | Reason |\n", ); - out.push_str("| --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | --- |\n"); + out.push_str("| --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | --- |\n"); for suite in &report.suites { out.push_str(&format!( - "| {} | `{}` | {} | `{}` | `{}` | `{}` | {} | {} | {} | {} | {} | {} | {} | {} |\n", + "| {} | `{}` | {} | `{}` | `{}` | `{}` | {} | {} | {} | {} | {} | {} | {} | {} | {} |\n", md_cell(suite.suite_id.as_str()), status_str(suite.status), suite.encoded_job_count, @@ -4149,6 +4201,7 @@ fn render_markdown_suites(out: &mut String, report: &RealWorldReport) { suite.conflict_detection_count, suite.update_rationale_available_count, suite.temporal_validity_not_encoded_count, + suite.history_readback_encoded_count, suite.unsupported_claim_count, suite.wrong_result_count, md_cell(suite.reason.as_str()) @@ -4306,8 +4359,12 @@ fn render_markdown_evolution(out: &mut String, report: &RealWorldReport) { "- Temporal validity not encoded: `{}`\n\n", report.evolution.temporal_validity_not_encoded_count )); - out.push_str("| Suite | Job | Current Evidence | Historical Evidence | Stale Traps Used | Conflict Count | Detected | Update Rationale | Temporal Validity | Follow-up |\n"); - out.push_str("| --- | --- | --- | --- | --- | ---: | ---: | --- | --- | --- |\n"); + out.push_str(&format!( + "- History readback encoded: `{}`\n\n", + report.evolution.history_readback_encoded_count + )); + out.push_str("| Suite | Job | Current Evidence | Historical Evidence | Stale Traps Used | Conflict Count | Detected | Update Rationale | Temporal Validity | History Readback | Follow-up |\n"); + out.push_str("| --- | --- | --- | --- | --- | ---: | ---: | --- | --- | --- | --- |\n"); for job in &report.jobs { let Some(evolution) = &job.evolution else { @@ -4315,7 +4372,7 @@ fn render_markdown_evolution(out: &mut String, report: &RealWorldReport) { }; out.push_str(&format!( - "| {} | {} | `{}` | `{}` | `{}` | {} | {} | `{}` | `{}` | {} |\n", + "| {} | {} | `{}` | `{}` | `{}` | {} | {} | `{}` | `{}` | `{}` | {} |\n", md_cell(job.suite_id.as_str()), md_cell(job.job_id.as_str()), md_inline(evolution.current_evidence.join(", ").as_str()), @@ -4325,6 +4382,7 @@ fn render_markdown_evolution(out: &mut String, report: &RealWorldReport) { evolution.conflict_detection_count, bool_display(evolution.update_rationale_available), temporal_display(evolution), + history_display(evolution), md_cell(evolution.follow_up.as_deref().unwrap_or("-")) )); } @@ -4695,6 +4753,20 @@ fn temporal_display(evolution: &EvolutionJobReport) -> &'static str { } } +fn history_display(evolution: &EvolutionJobReport) -> String { + if !evolution.history_readback_encoded { + return "-".to_string(); + } + + let mut parts = vec![format!("events={}", evolution.history_event_types.join(","))]; + + if evolution.history_requires_note_version_links { + parts.push("note_version_links=true".to_string()); + } + + parts.join(";") +} + fn cost_display(cost: Option<&CostReport>) -> String { let Some(cost) = cost else { return "-".to_string(); diff --git a/apps/elf-eval/tests/real_world_job_benchmark.rs b/apps/elf-eval/tests/real_world_job_benchmark.rs index 8e2a9056..0011dd6c 100644 --- a/apps/elf-eval/tests/real_world_job_benchmark.rs +++ b/apps/elf-eval/tests/real_world_job_benchmark.rs @@ -1011,10 +1011,18 @@ fn memory_evolution_fixtures_report_temporal_and_staleness_metrics() -> Result<( report.pointer("/summary/temporal_validity_not_encoded_count").and_then(Value::as_u64), Some(0) ); + assert_eq!( + report.pointer("/summary/history_readback_encoded_count").and_then(Value::as_u64), + Some(1) + ); assert_eq!( report.pointer("/evolution/temporal_validity_not_encoded_count").and_then(Value::as_u64), Some(0) ); + assert_eq!( + report.pointer("/evolution/history_readback_encoded_count").and_then(Value::as_u64), + Some(1) + ); let suites = array_at(&report, "/suites")?; let memory_evolution = find_by_field(suites, "/suite_id", "memory_evolution")?; @@ -1024,10 +1032,28 @@ fn memory_evolution_fixtures_report_temporal_and_staleness_metrics() -> Result<( memory_evolution.pointer("/temporal_validity_not_encoded_count").and_then(Value::as_u64), Some(0) ); + assert_eq!( + memory_evolution.pointer("/history_readback_encoded_count").and_then(Value::as_u64), + Some(1) + ); let jobs = array_at(&report, "/jobs")?; + let preference_job = find_by_field(jobs, "/job_id", "memory-evolution-preference-001")?; let relation_job = find_by_field(jobs, "/job_id", "memory-evolution-relation-temporal-001")?; + assert_eq!( + preference_job.pointer("/evolution/history_readback_encoded").and_then(Value::as_bool), + Some(true) + ); + assert!(array_contains_str(preference_job, "/evolution/history_event_types", "add")?); + assert!(array_contains_str(preference_job, "/evolution/history_event_types", "update")?); + assert!(array_contains_str(preference_job, "/evolution/history_event_types", "ignore")?); + assert_eq!( + preference_job + .pointer("/evolution/history_requires_note_version_links") + .and_then(Value::as_bool), + Some(true) + ); assert_eq!(relation_job.pointer("/status").and_then(Value::as_str), Some("pass")); assert_eq!( relation_job.pointer("/evolution/temporal_validity_not_encoded").and_then(Value::as_bool), diff --git a/apps/elf-mcp/src/server.rs b/apps/elf-mcp/src/server.rs index 80829255..2d67b4b8 100644 --- a/apps/elf-mcp/src/server.rs +++ b/apps/elf-mcp/src/server.rs @@ -568,6 +568,21 @@ impl ElfMcp { self.forward(HttpMethod::Get, &path, JsonObject::new(), None).await } + #[rmcp::tool( + name = "elf_admin_memory_history_get", + description = "Fetch chronological memory history for one note.", + input_schema = admin_memory_history_get_schema() + )] + async fn elf_admin_memory_history_get( + &self, + mut params: JsonObject, + ) -> Result { + let note_id = take_required_string(&mut params, "note_id")?; + let path = format!("/v2/admin/notes/{note_id}/history"); + + self.forward(HttpMethod::Get, &path, JsonObject::new(), None).await + } + #[rmcp::tool( name = "elf_admin_trace_bundle_get", description = "Fetch trace bundle for replay and diagnostics by trace_id.", @@ -1383,6 +1398,10 @@ fn admin_note_provenance_get_schema() -> Arc { })) } +fn admin_memory_history_get_schema() -> Arc { + admin_note_provenance_get_schema() +} + fn admin_trace_bundle_get_schema() -> Arc { Arc::new(rmcp::object!({ "type": "object", @@ -1532,7 +1551,7 @@ mod tests { type RequestRecorder = Arc>>>; - const ALL_TOOL_DEFINITIONS: [ToolDefinition; 28] = [ + const ALL_TOOL_DEFINITIONS: [ToolDefinition; 29] = [ ToolDefinition::new( "elf_notes_ingest", HttpMethod::Post, @@ -1659,6 +1678,12 @@ mod tests { "/v2/admin/notes/{note_id}/provenance", "Fetch provenance bundle for a note.", ), + ToolDefinition::new( + "elf_admin_memory_history_get", + HttpMethod::Get, + "/v2/admin/notes/{note_id}/history", + "Fetch chronological memory history for a note.", + ), ToolDefinition::new( "elf_admin_trace_bundle_get", HttpMethod::Get, @@ -1758,6 +1783,7 @@ mod tests { "elf_admin_trajectory_get", "elf_admin_trace_item_get", "elf_admin_note_provenance_get", + "elf_admin_memory_history_get", "elf_admin_trace_bundle_get", "elf_admin_events_ingestion_profiles_list", "elf_admin_events_ingestion_profiles_create", @@ -1869,6 +1895,7 @@ mod tests { mcp.api_base_for_path("/v2/admin/notes/abcd/provenance"), "http://127.0.0.1:9001" ); + assert_eq!(mcp.api_base_for_path("/v2/admin/notes/abcd/history"), "http://127.0.0.1:9001"); assert_eq!(mcp.api_base_for_path("/v2/searches"), "http://127.0.0.1:9000"); } diff --git a/docs/guide/observability.md b/docs/guide/observability.md index e355c6b3..d0bfccfb 100644 --- a/docs/guide/observability.md +++ b/docs/guide/observability.md @@ -32,12 +32,17 @@ For a note-level traceability trail: - Equivalent HTTP endpoint: - `GET /v2/admin/notes/{note_id}/provenance` - Schema: `elf.note_provenance_bundle/v1` +- Memory history readback: + - MCP tool: `elf_admin_memory_history_get` + - `GET /v2/admin/notes/{note_id}/history` + - Schema: `elf.memory_history/v1` Returned bundle sections: - `note` - `ingest_decisions` - `note_versions` +- `history` - `indexing_outbox` - `recent_traces` @@ -61,6 +66,7 @@ Recommended loop: 1. Start from a user-facing error `trace_id` or note `note_id`. 2. Query `elf_admin_trace_*` family to inspect trajectory and trace items. 3. Use `elf_admin_note_provenance_get` to connect trace history with ingest and indexing state. +4. Use `elf_admin_memory_history_get` when you only need chronological memory evolution events. ## 4) MCP admin/debug surface map @@ -70,3 +76,4 @@ Recommended loop: - `elf_admin_trace_item_get` -> `GET /v2/admin/trace-items/{item_id}` - `elf_admin_trace_bundle_get` -> `GET /v2/admin/traces/{trace_id}/bundle` - `elf_admin_note_provenance_get` -> `GET /v2/admin/notes/{note_id}/provenance` +- `elf_admin_memory_history_get` -> `GET /v2/admin/notes/{note_id}/history` diff --git a/docs/spec/system_elf_memory_service_v2.md b/docs/spec/system_elf_memory_service_v2.md index 7ef7218b..ad86d61b 100644 --- a/docs/spec/system_elf_memory_service_v2.md +++ b/docs/spec/system_elf_memory_service_v2.md @@ -603,6 +603,7 @@ Indexes: - note_type text not null - note_key text null - note_id uuid null +- note_version_id uuid null - base_decision text not null - policy_decision text not null - note_op text not null @@ -612,6 +613,7 @@ Indexes: Indexing: - idx_memory_ingest_decisions_tenant_scope_pipeline: (tenant_id, project_id, agent_id, scope, pipeline, ts) +- idx_memory_ingest_decisions_note_version_id: (note_version_id) details must include: - similarity_best @@ -1412,6 +1414,48 @@ Response: "recent_traces": [...] } +GET /v2/admin/notes/{note_id}/history + +Headers: +- X-ELF-Tenant-Id (required) +- X-ELF-Project-Id (required) +- X-ELF-Agent-Id (required) + +Path: +- note_id: uuid + +Response: +{ + "schema": "elf.memory_history/v1", + "note_id": "uuid", + "events": [ + { + "event_id": "string", + "event_type": "add|update|ignore|reject|expire|delete|derived|applied|invalidated|related", + "subject_type": "note", + "note_id": "uuid", + "source_table": "string", + "source_id": "uuid|null", + "related_note_version_id": "uuid|null", + "related_decision_id": "uuid|null", + "related_proposal_id": "uuid|null", + "actor": "string|null", + "op": "string|null", + "reason_code": "string|null", + "summary": "string", + "details": { ... }, + "ts": "..." + } + ] +} + +Notes: +- History events are a chronological read-only projection over durable source tables. +- Ingest decisions that produce note versions should set `note_version_id` so history + can link the decision to the resulting note version. +- Derived, applied, and invalidated events come from consolidation proposals and + review events that reference the note in `source_refs`. + ============================================================ 15. HTTP API (PUBLIC) ============================================================ @@ -2106,6 +2150,7 @@ Original query: - elf_admin_trace_item_get -> GET /v2/admin/trace-items/{item_id} - elf_admin_trace_bundle_get -> GET /v2/admin/traces/{trace_id}/bundle - elf_admin_note_provenance_get -> GET /v2/admin/notes/{note_id}/provenance + - elf_admin_memory_history_get -> GET /v2/admin/notes/{note_id}/history - The MCP server must contain zero business logic or policy. - All policy remains in elf-api and elf-service. diff --git a/docs/spec/system_provenance_mapping_v1.md b/docs/spec/system_provenance_mapping_v1.md index 9fdcb3d4..fdffaf11 100644 --- a/docs/spec/system_provenance_mapping_v1.md +++ b/docs/spec/system_provenance_mapping_v1.md @@ -8,6 +8,7 @@ Defines: `elf.note_provenance_bundle/v1`. Identifier: - `elf.note_provenance_bundle/v1` +- `elf.memory_history/v1` Status: active. @@ -16,12 +17,14 @@ Scope ================================================== - Defines the response contract for `/v2/admin/notes/{note_id}/provenance`. +- Defines the response contract for `/v2/admin/notes/{note_id}/history`. - Captures the same note-level artifacts needed for auditability and debugging: - source note state - ingest decisions - note version history - indexing outbox state - recent traces involving the note + - normalized memory history events - Does not define any mutation semantics. ================================================== @@ -39,7 +42,8 @@ This admin endpoint returns a single JSON object that **must** use: "ingest_decisions": [...], "note_versions": [...], "indexing_outbox": [...], - "recent_traces": [...] + "recent_traces": [...], + "history": [...] } ``` @@ -69,6 +73,15 @@ and ordered by `updated_at DESC`. - `search_traces` and `search_trace_items` where the trace references the note id, ordered by `created_at DESC, trace_id DESC`. +`history` is a normalized chronological projection joined from: +- `memory_note_versions` for add/update/delete/publish/unpublish and related transitions. +- `memory_ingest_decisions` for ignore/reject decisions and for decision-to-version links. +- `memory_notes.expires_at` for persisted expiry readback when the note has reached its + TTL timestamp and no explicit expiry version row exists. +- `consolidation_proposals` and `consolidation_proposal_reviews` for derived, + applied, and invalidated proposal outcomes that reference the note in + `source_refs`. + ================================================== 2) Response field shape ================================================== @@ -81,16 +94,55 @@ Core envelope: - `note_versions` (array, required): ordered historical versions. - `indexing_outbox` (array, required): active/retry indexing jobs for the note. - `recent_traces` (array, required): bounded traces involving this note. +- `history` (array, required): bounded chronological memory events. No additional top-level keys are required by this contract. ================================================== -3) MCP exposure +3) History endpoint +================================================== + +`GET /v2/admin/notes/{note_id}/history` + +This admin endpoint returns: + +```json +{ + "schema": "elf.memory_history/v1", + "note_id": "uuid", + "events": [ + { + "event_id": "string", + "event_type": "add|update|ignore|reject|expire|delete|derived|applied|invalidated|related", + "subject_type": "note", + "note_id": "uuid", + "source_table": "string", + "source_id": "uuid|null", + "related_note_version_id": "uuid|null", + "related_decision_id": "uuid|null", + "related_proposal_id": "uuid|null", + "actor": "string|null", + "op": "string|null", + "reason_code": "string|null", + "summary": "string", + "details": {}, + "ts": "RFC3339 timestamp" + } + ] +} +``` + +History ordering is chronological by `ts ASC`, then `event_id ASC`. Events are +bounded by service limits. + +================================================== +4) MCP exposure ================================================== MCP tool: - `elf_admin_note_provenance_get` -> `GET /v2/admin/notes/{note_id}/provenance` +- `elf_admin_memory_history_get` -> `GET /v2/admin/notes/{note_id}/history` Request input: @@ -101,7 +153,7 @@ Request input: ``` ================================================== -4) Operational guidance +5) Operational guidance ================================================== - Keep `recent_traces` small (bounded by service defaults) to avoid large admin payloads. diff --git a/docs/spec/system_version_registry.md b/docs/spec/system_version_registry.md index 7053678b..efe338af 100644 --- a/docs/spec/system_version_registry.md +++ b/docs/spec/system_version_registry.md @@ -50,6 +50,14 @@ This document is normative. When a new versioned identifier is introduced, it mu - Consumers: Admin tooling and MCP adapter (`elf_admin_note_provenance_get`), diagnostics runbooks. - Bump rule: Introduce a new bundle version only when existing keys/shape/required joins become incompatible with v1 clients. +### Memory history schema + +- Identifier: `elf.memory_history/v1`. +- Type: Admin memory history response envelope for chronological memory evolution readback. +- Defined in: `docs/spec/system_provenance_mapping_v1.md`. +- Consumers: Admin tooling and MCP adapter (`elf_admin_memory_history_get`), diagnostics runbooks, lifecycle benchmarks. +- Bump rule: Introduce a new history version only when event shape or ordering semantics become incompatible with v1 clients. + ### Doc Extension v1 docs filters contract - Identifier: `docs_search_filters/v1`. diff --git a/packages/elf-service/src/add_event.rs b/packages/elf-service/src/add_event.rs index 753fd5f2..a6eb0b80 100644 --- a/packages/elf-service/src/add_event.rs +++ b/packages/elf-service/src/add_event.rs @@ -25,6 +25,7 @@ use elf_domain::{ use elf_storage::models::MemoryNote; type ProcessedEventOutput = (Vec, Vec, Option>); +type AddEventPersistOutput = (AddEventResult, Option); const REJECT_STRUCTURED_INVALID: &str = "REJECT_STRUCTURED_INVALID"; const IGNORE_DUPLICATE: &str = "IGNORE_DUPLICATE"; @@ -366,6 +367,8 @@ impl ElfService { ignore_reason_code, ); + let mut note_version_id = None; + if should_apply && !dry_run { let persist_args = PersistExtractedNoteArgs { req, @@ -395,10 +398,12 @@ impl ElfService { now, embed_version, }; - - result = self + let persisted = self .persist_extracted_note_decision(tx, persist_args, decision, policy_decision) .await?; + + result = persisted.0; + note_version_id = persisted.1; } result.write_policy_audits = write_policy_audits.cloned(); @@ -410,6 +415,7 @@ impl ElfService { note, note_data.note_type.as_str(), result.note_id, + note_version_id, base_decision, policy_decision, result.op, @@ -461,6 +467,7 @@ impl ElfService { note, note_data.note_type.as_str(), None, + None, MemoryPolicyDecision::Reject, MemoryPolicyDecision::Reject, NoteOp::Rejected, @@ -497,6 +504,7 @@ impl ElfService { note, note_data.note_type.as_str(), None, + None, MemoryPolicyDecision::Reject, MemoryPolicyDecision::Reject, NoteOp::Rejected, @@ -534,6 +542,7 @@ impl ElfService { note, note_data.note_type.as_str(), None, + None, MemoryPolicyDecision::Reject, MemoryPolicyDecision::Reject, NoteOp::Rejected, @@ -594,7 +603,7 @@ impl ElfService { args: PersistExtractedNoteArgs<'_>, decision: UpdateDecision, policy_decision: MemoryPolicyDecision, - ) -> Result { + ) -> Result { match (decision, args) { (UpdateDecision::Add { note_id, .. }, args) => self.persist_extracted_note_add(tx, args, note_id, policy_decision).await, @@ -611,7 +620,7 @@ impl ElfService { args: PersistExtractedNoteArgs<'_>, note_id: Uuid, policy_decision: MemoryPolicyDecision, - ) -> Result { + ) -> Result { access::ensure_active_project_scope_grant( &mut **tx, args.req.tenant_id.as_str(), @@ -644,7 +653,7 @@ impl ElfService { insert_memory_note_tx(tx, &memory_note).await?; - crate::insert_version( + let note_version_id = crate::insert_version( &mut **tx, InsertVersionArgs { note_id: memory_note.note_id, @@ -657,6 +666,7 @@ impl ElfService { }, ) .await?; + crate::enqueue_outbox_tx( &mut **tx, memory_note.note_id, @@ -684,15 +694,18 @@ impl ElfService { .await?; } - Ok(AddEventResult { - note_id: Some(note_id), - op: NoteOp::Add, - policy_decision, - reason_code: None, - reason: args.reason.cloned(), - field_path: None, - write_policy_audits: None, - }) + Ok(( + AddEventResult { + note_id: Some(note_id), + op: NoteOp::Add, + policy_decision, + reason_code: None, + reason: args.reason.cloned(), + field_path: None, + write_policy_audits: None, + }, + Some(note_version_id), + )) } async fn persist_extracted_note_update( @@ -701,7 +714,7 @@ impl ElfService { args: PersistExtractedNoteArgs<'_>, note_id: Uuid, policy_decision: MemoryPolicyDecision, - ) -> Result { + ) -> Result { let mut existing: MemoryNote = sqlx::query_as::<_, MemoryNote>( "SELECT * FROM memory_notes WHERE note_id = $1 FOR UPDATE", ) @@ -729,7 +742,7 @@ impl ElfService { update_memory_note_tx(tx, &existing).await?; - crate::insert_version( + let note_version_id = crate::insert_version( &mut **tx, InsertVersionArgs { note_id: existing.note_id, @@ -742,6 +755,7 @@ impl ElfService { }, ) .await?; + crate::enqueue_outbox_tx( &mut **tx, existing.note_id, @@ -769,15 +783,18 @@ impl ElfService { .await?; } - Ok(AddEventResult { - note_id: Some(note_id), - op: NoteOp::Update, - policy_decision, - reason_code: None, - reason: args.reason.cloned(), - field_path: None, - write_policy_audits: None, - }) + Ok(( + AddEventResult { + note_id: Some(note_id), + op: NoteOp::Update, + policy_decision, + reason_code: None, + reason: args.reason.cloned(), + field_path: None, + write_policy_audits: None, + }, + Some(note_version_id), + )) } async fn persist_extracted_note_none( @@ -786,7 +803,7 @@ impl ElfService { args: PersistExtractedNoteArgs<'_>, note_id: Uuid, policy_decision: MemoryPolicyDecision, - ) -> Result { + ) -> Result { let mut did_update = false; if let Some(structured) = args.structured @@ -818,6 +835,26 @@ impl ElfService { } if did_update { + let note_row: MemoryNote = + sqlx::query_as("SELECT * FROM memory_notes WHERE note_id = $1") + .bind(note_id) + .fetch_one(&mut **tx) + .await?; + let snapshot = crate::note_snapshot(¬e_row); + let note_version_id = crate::insert_version( + &mut **tx, + InsertVersionArgs { + note_id, + op: "UPDATE", + prev_snapshot: Some(snapshot.clone()), + new_snapshot: Some(snapshot), + reason: "add_event_structured", + actor: args.req.agent_id.as_str(), + ts: args.now, + }, + ) + .await?; + if matches!(args.scope, "project_shared" | "org_shared") { access::ensure_active_project_scope_grant( &mut **tx, @@ -829,26 +866,32 @@ impl ElfService { .await?; } - return Ok(AddEventResult { + return Ok(( + AddEventResult { + note_id: Some(note_id), + op: NoteOp::Update, + policy_decision, + reason_code: None, + reason: args.reason.cloned(), + field_path: None, + write_policy_audits: None, + }, + Some(note_version_id), + )); + } + + Ok(( + AddEventResult { note_id: Some(note_id), - op: NoteOp::Update, + op: NoteOp::None, policy_decision, reason_code: None, reason: args.reason.cloned(), field_path: None, write_policy_audits: None, - }); - } - - Ok(AddEventResult { - note_id: Some(note_id), - op: NoteOp::None, - policy_decision, - reason_code: None, - reason: args.reason.cloned(), - field_path: None, - write_policy_audits: None, - }) + }, + None, + )) } } @@ -1207,6 +1250,7 @@ async fn record_ingest_decision( note: &ExtractedNote, note_type: &str, note_id: Option, + note_version_id: Option, base_decision: MemoryPolicyDecision, policy_decision: MemoryPolicyDecision, note_op: NoteOp, @@ -1232,6 +1276,7 @@ async fn record_ingest_decision( note_type, note_key: note.key.as_deref(), note_id, + note_version_id, base_decision, policy_decision, note_op, diff --git a/packages/elf-service/src/add_note.rs b/packages/elf-service/src/add_note.rs index 5cb433e6..4a67401c 100644 --- a/packages/elf-service/src/add_note.rs +++ b/packages/elf-service/src/add_note.rs @@ -23,6 +23,8 @@ use elf_domain::{ }; use elf_storage::models::MemoryNote; +type AddNoteApplyOutput = (AddNoteResult, NoteOp, Option); + const REJECT_STRUCTURED_INVALID: &str = "REJECT_STRUCTURED_INVALID"; const IGNORE_DUPLICATE: &str = "IGNORE_DUPLICATE"; const IGNORE_POLICY_THRESHOLD: &str = "IGNORE_POLICY_THRESHOLD"; @@ -161,7 +163,7 @@ impl ElfService { let note_id = decision.note_id(); let ignore_reason_code = Self::ignore_reason_code(policy_decision, base_decision, metadata.matched_dup); - let (result, note_op) = self + let (result, note_op, note_version_id) = self .apply_policy_result( &mut tx, &decision, @@ -181,6 +183,7 @@ impl ElfService { ctx, ¬e, result.note_id, + note_version_id, base_decision, result.policy_decision, note_op, @@ -223,6 +226,7 @@ impl ElfService { ctx, note, None, + None, MemoryPolicyDecision::Reject, MemoryPolicyDecision::Reject, NoteOp::Rejected, @@ -249,6 +253,7 @@ impl ElfService { ctx, note, None, + None, MemoryPolicyDecision::Reject, MemoryPolicyDecision::Reject, NoteOp::Rejected, @@ -350,25 +355,28 @@ impl ElfService { note_id: Uuid, policy_decision: MemoryPolicyDecision, ignore_reason_code: Option<&'static str>, - ) -> Result<(AddNoteResult, NoteOp)> { + ) -> Result { let should_apply = matches!( policy_decision, MemoryPolicyDecision::Remember | MemoryPolicyDecision::Update ); if should_apply { - let result = match decision { + let (result, note_version_id) = match decision { UpdateDecision::Add { .. } => { - self.handle_add_note_add(tx, ctx, note, note_id).await?; + let note_version_id = self.handle_add_note_add(tx, ctx, note, note_id).await?; - AddNoteResult { - note_id: Some(note_id), - op: NoteOp::Add, - policy_decision, - reason_code: None, - field_path: None, - write_policy_audit: None, - } + ( + AddNoteResult { + note_id: Some(note_id), + op: NoteOp::Add, + policy_decision, + reason_code: None, + field_path: None, + write_policy_audit: None, + }, + Some(note_version_id), + ) }, UpdateDecision::Update { .. } => self.handle_add_note_update( @@ -381,7 +389,7 @@ impl ElfService { ) .await?, UpdateDecision::None { .. } => { - let mut none_result = self + let (mut none_result, note_version_id) = self .handle_add_note_none( tx, ctx, @@ -395,12 +403,12 @@ impl ElfService { none_result.policy_decision = policy_decision; - none_result + (none_result, note_version_id) }, }; let note_op = result.op; - Ok((result, note_op)) + Ok((result, note_op, note_version_id)) } else { let mut result = AddNoteResult { note_id: Some(note_id), @@ -418,7 +426,7 @@ impl ElfService { UpdateDecision::Update { .. } | UpdateDecision::None { .. } => {}, } - Ok((result, NoteOp::None)) + Ok((result, NoteOp::None, None)) } } @@ -429,6 +437,7 @@ impl ElfService { ctx: &AddNoteContext<'_>, note: &AddNoteInput, note_id: Option, + note_version_id: Option, base_decision: MemoryPolicyDecision, policy_decision: MemoryPolicyDecision, note_op: NoteOp, @@ -450,6 +459,7 @@ impl ElfService { note_type: note.r#type.as_str(), note_key: note.key.as_deref(), note_id, + note_version_id, base_decision, policy_decision, note_op, @@ -507,7 +517,7 @@ impl ElfService { ctx: &AddNoteContext<'_>, note: &AddNoteInput, note_id: Uuid, - ) -> Result<()> { + ) -> Result { access::ensure_active_project_scope_grant( &mut **tx, ctx.tenant_id, @@ -542,7 +552,7 @@ impl ElfService { insert_memory_note_tx(tx, &memory_note).await?; - crate::insert_version( + let note_version_id = crate::insert_version( &mut **tx, InsertVersionArgs { note_id: memory_note.note_id, @@ -576,7 +586,7 @@ impl ElfService { ) .await?; - Ok(()) + Ok(note_version_id) } async fn handle_add_note_update( @@ -587,7 +597,7 @@ impl ElfService { agent_id: &str, now: OffsetDateTime, policy_decision: MemoryPolicyDecision, - ) -> Result { + ) -> Result<(AddNoteResult, Option)> { let mut existing: MemoryNote = sqlx::query_as::<_, MemoryNote>( "SELECT * FROM memory_notes WHERE note_id = $1 FOR UPDATE", ) @@ -619,14 +629,17 @@ impl ElfService { && existing.source_ref == note.source_ref; if unchanged { - return Ok(AddNoteResult { - note_id: Some(note_id), - op: NoteOp::None, - policy_decision: MemoryPolicyDecision::Ignore, - reason_code: None, - field_path: None, - write_policy_audit: None, - }); + return Ok(( + AddNoteResult { + note_id: Some(note_id), + op: NoteOp::None, + policy_decision: MemoryPolicyDecision::Ignore, + reason_code: None, + field_path: None, + write_policy_audit: None, + }, + None, + )); } access::ensure_active_project_scope_grant( @@ -647,7 +660,7 @@ impl ElfService { update_memory_note_tx(tx, &existing).await?; - crate::insert_version( + let note_version_id = crate::insert_version( &mut **tx, InsertVersionArgs { note_id: existing.note_id, @@ -681,14 +694,17 @@ impl ElfService { ) .await?; - Ok(AddNoteResult { - note_id: Some(note_id), - op: NoteOp::Update, - policy_decision, - reason_code: None, - field_path: None, - write_policy_audit: None, - }) + Ok(( + AddNoteResult { + note_id: Some(note_id), + op: NoteOp::Update, + policy_decision, + reason_code: None, + field_path: None, + write_policy_audit: None, + }, + Some(note_version_id), + )) } #[allow(clippy::too_many_arguments)] @@ -701,7 +717,7 @@ impl ElfService { now: OffsetDateTime, embed_version: &str, policy_decision: MemoryPolicyDecision, - ) -> Result { + ) -> Result<(AddNoteResult, Option)> { let mut should_update = false; if let Some(structured) = note.structured.as_ref() { @@ -730,6 +746,26 @@ impl ElfService { } if should_update { + let note_row: MemoryNote = + sqlx::query_as("SELECT * FROM memory_notes WHERE note_id = $1") + .bind(note_id) + .fetch_one(&mut **tx) + .await?; + let snapshot = crate::note_snapshot(¬e_row); + let note_version_id = crate::insert_version( + &mut **tx, + InsertVersionArgs { + note_id, + op: "UPDATE", + prev_snapshot: Some(snapshot.clone()), + new_snapshot: Some(snapshot), + reason: "add_note_structured", + actor: ctx.agent_id, + ts: now, + }, + ) + .await?; + if matches!(ctx.scope, "project_shared" | "org_shared") { access::ensure_active_project_scope_grant( &mut **tx, @@ -741,24 +777,30 @@ impl ElfService { .await?; } - return Ok(AddNoteResult { + return Ok(( + AddNoteResult { + note_id: Some(note_id), + op: NoteOp::Update, + policy_decision, + reason_code: None, + field_path: None, + write_policy_audit: None, + }, + Some(note_version_id), + )); + } + + Ok(( + AddNoteResult { note_id: Some(note_id), - op: NoteOp::Update, + op: NoteOp::None, policy_decision, reason_code: None, field_path: None, write_policy_audit: None, - }); - } - - Ok(AddNoteResult { - note_id: Some(note_id), - op: NoteOp::None, - policy_decision, - reason_code: None, - field_path: None, - write_policy_audit: None, - }) + }, + None, + )) } #[allow(clippy::too_many_arguments)] diff --git a/packages/elf-service/src/ingest_audit.rs b/packages/elf-service/src/ingest_audit.rs index 4cd3907b..77b2d5f6 100644 --- a/packages/elf-service/src/ingest_audit.rs +++ b/packages/elf-service/src/ingest_audit.rs @@ -14,6 +14,7 @@ pub(crate) struct IngestAuditArgs<'a> { pub note_type: &'a str, pub note_key: Option<&'a str>, pub note_id: Option, + pub note_version_id: Option, pub base_decision: MemoryPolicyDecision, pub policy_decision: MemoryPolicyDecision, pub note_op: NoteOp, @@ -49,6 +50,7 @@ pub(crate) async fn insert_ingest_decision( note_type, note_key, note_id, + note_version_id, base_decision, policy_decision, note_op, @@ -83,6 +85,7 @@ INSERT INTO memory_ingest_decisions ( note_type, note_key, note_id, + note_version_id, base_decision, policy_decision, note_op, @@ -90,7 +93,7 @@ INSERT INTO memory_ingest_decisions ( details, ts ) -VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15)", +VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16)", ) .bind(Uuid::new_v4()) .bind(tenant_id) @@ -101,6 +104,7 @@ VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15)", .bind(note_type) .bind(note_key) .bind(note_id) + .bind(note_version_id) .bind(memory_policy_decision_to_str(base_decision)) .bind(memory_policy_decision_to_str(policy_decision)) .bind(note_op_to_str(note_op)) diff --git a/packages/elf-service/src/lib.rs b/packages/elf-service/src/lib.rs index 47833604..e784e4b0 100644 --- a/packages/elf-service/src/lib.rs +++ b/packages/elf-service/src/lib.rs @@ -83,6 +83,7 @@ pub use self::{ SearchTimelineGroup, SearchTimelineRequest, SearchTimelineResponse, }, provenance::{ + MemoryHistoryEvent, MemoryHistoryGetRequest, MemoryHistoryResponse, NoteProvenanceBundleResponse, NoteProvenanceGetRequest, NoteProvenanceIndexingOutbox, NoteProvenanceIngestDecision, NoteProvenanceNote, NoteProvenanceNoteVersion, NoteProvenanceRecentTrace, @@ -575,11 +576,12 @@ where }) } -pub(crate) async fn insert_version<'e, E>(executor: E, args: InsertVersionArgs<'_>) -> Result<()> +pub(crate) async fn insert_version<'e, E>(executor: E, args: InsertVersionArgs<'_>) -> Result where E: PgExecutor<'e>, { let InsertVersionArgs { note_id, op, prev_snapshot, new_snapshot, reason, actor, ts } = args; + let version_id = Uuid::new_v4(); sqlx::query( "\ @@ -595,7 +597,7 @@ INSERT INTO memory_note_versions ( ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8)", ) - .bind(Uuid::new_v4()) + .bind(version_id) .bind(note_id) .bind(op) .bind(prev_snapshot) @@ -606,7 +608,7 @@ VALUES ($1,$2,$3,$4,$5,$6,$7,$8)", .execute(executor) .await?; - Ok(()) + Ok(version_id) } pub(crate) async fn enqueue_outbox_tx<'e, E>( diff --git a/packages/elf-service/src/provenance.rs b/packages/elf-service/src/provenance.rs index c873030b..c39af394 100644 --- a/packages/elf-service/src/provenance.rs +++ b/packages/elf-service/src/provenance.rs @@ -1,7 +1,9 @@ //! Provenance inspection APIs. +use std::collections::HashMap; + use serde::{Deserialize, Serialize}; -use serde_json::Value; +use serde_json::{self, Value}; use sqlx::{FromRow, PgPool}; use time::OffsetDateTime; use uuid::Uuid; @@ -14,6 +16,8 @@ const NOTE_PROVENANCE_INGEST_DECISIONS_LIMIT: i64 = 100; const NOTE_PROVENANCE_NOTE_VERSIONS_LIMIT: i64 = 100; const NOTE_PROVENANCE_OUTBOX_LIMIT: i64 = 100; const NOTE_PROVENANCE_RECENT_TRACES_LIMIT: i64 = 20; +const NOTE_PROVENANCE_HISTORY_LIMIT: i64 = 200; +const MEMORY_HISTORY_SCHEMA_V1: &str = "elf.memory_history/v1"; /// Request payload for note provenance lookup. #[derive(Clone, Debug, Deserialize, Serialize)] @@ -26,6 +30,28 @@ pub struct NoteProvenanceGetRequest { pub note_id: Uuid, } +/// Request payload for memory-history lookup. +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct MemoryHistoryGetRequest { + /// Tenant that owns the memory. + pub tenant_id: String, + /// Project that owns the memory. + pub project_id: String, + /// Identifier of the note to inspect. + pub note_id: Uuid, +} + +/// Timeline response for one memory. +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct MemoryHistoryResponse { + /// History schema identifier. + pub schema: String, + /// Inspected note identifier. + pub note_id: Uuid, + /// Chronological memory events. + pub events: Vec, +} + /// Full provenance bundle for one note. #[derive(Clone, Debug, Deserialize, Serialize)] pub struct NoteProvenanceBundleResponse { @@ -41,6 +67,8 @@ pub struct NoteProvenanceBundleResponse { pub indexing_outbox: Vec, /// Recent search traces that referenced the note. pub recent_traces: Vec, + /// Chronological memory event timeline for the note. + pub history: Vec, } /// Current note snapshot returned by provenance APIs. @@ -133,6 +161,9 @@ pub struct NoteProvenanceIngestDecision { pub note_key: Option, /// Note identifier, when a note was persisted or matched. pub note_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + /// Note version produced by this decision, when applicable. + pub note_version_id: Option, /// Pre-policy base decision. pub base_decision: String, /// Final policy decision. @@ -159,6 +190,7 @@ impl From for NoteProvenanceIngestDecision { note_type: row.note_type, note_key: row.note_key, note_id: row.note_id, + note_version_id: row.note_version_id, base_decision: row.base_decision, policy_decision: row.policy_decision, note_op: row.note_op, @@ -272,6 +304,48 @@ pub struct NoteProvenanceRecentTrace { pub created_at: OffsetDateTime, } +/// One normalized memory-history event. +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct MemoryHistoryEvent { + /// Stable event identifier within its source table. + pub event_id: String, + /// Normalized event type. + pub event_type: String, + /// Subject kind for the event. + pub subject_type: String, + /// Inspected note identifier. + pub note_id: Uuid, + /// Durable source table behind the event. + pub source_table: String, + /// Source row identifier when available. + pub source_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + /// Related note version, when an ingest decision produced a version row. + pub related_note_version_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + /// Related ingest decision, when a version or history event was caused by ingestion. + pub related_decision_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + /// Related consolidation proposal, when a derived memory proposal references the note. + pub related_proposal_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + /// Actor that caused the event, when available. + pub actor: Option, + #[serde(skip_serializing_if = "Option::is_none")] + /// Source operation string. + pub op: Option, + #[serde(skip_serializing_if = "Option::is_none")] + /// Machine-readable reason code, when available. + pub reason_code: Option, + /// Human-readable one-line event summary. + pub summary: String, + /// Source-specific event details. + pub details: Value, + #[serde(with = "crate::time_serde")] + /// Event timestamp. + pub ts: OffsetDateTime, +} + #[derive(Clone, Debug)] struct ValidatedNoteProvenanceRequest { tenant_id: String, @@ -290,6 +364,7 @@ struct NoteIngestDecisionRow { note_type: String, note_key: Option, note_id: Option, + note_version_id: Option, base_decision: String, policy_decision: String, note_op: String, @@ -335,6 +410,40 @@ struct NoteRecentTraceRow { created_at: OffsetDateTime, } +#[derive(FromRow)] +struct NoteDerivedProposalRow { + proposal_id: Uuid, + run_id: Uuid, + agent_id: String, + proposal_kind: String, + apply_intent: String, + review_state: String, + source_refs: Value, + source_snapshot: Value, + lineage: Value, + diff: Value, + confidence: f32, + target_ref: Value, + proposed_payload: Value, + created_at: OffsetDateTime, +} + +#[derive(FromRow)] +struct NoteProposalReviewRow { + review_id: Uuid, + proposal_id: Uuid, + run_id: Uuid, + reviewer_agent_id: String, + action: String, + from_review_state: String, + to_review_state: String, + review_comment: Option, + created_at: OffsetDateTime, + proposal_kind: String, + apply_intent: String, + diff: Value, +} + impl ElfService { /// Loads the current note plus recent provenance tables as one bundle. pub async fn note_provenance_get( @@ -371,6 +480,7 @@ WHERE note_id = $1 req.note_id, ) .await?; + let history = load_memory_history_events(&self.db.pool, &req, ¬e_row).await?; Ok(NoteProvenanceBundleResponse { schema: NOTE_PROVENANCE_BUNDLE_SCHEMA_V1.to_string(), @@ -379,6 +489,42 @@ WHERE note_id = $1 note_versions, indexing_outbox, recent_traces, + history, + }) + } + + /// Loads the normalized memory-history timeline for one note. + pub async fn memory_history_get( + &self, + req: MemoryHistoryGetRequest, + ) -> Result { + let req = validate_note_provenance_request(NoteProvenanceGetRequest { + tenant_id: req.tenant_id, + project_id: req.project_id, + note_id: req.note_id, + })?; + let note_row = sqlx::query_as::<_, MemoryNote>( + "\ +SELECT * +FROM memory_notes +WHERE note_id = $1 + AND tenant_id = $2 + AND project_id = $3", + ) + .bind(req.note_id) + .bind(&req.tenant_id) + .bind(&req.project_id) + .fetch_optional(&self.db.pool) + .await?; + let Some(note_row) = note_row else { + return Err(Error::InvalidRequest { message: "Note not found.".to_string() }); + }; + let events = load_memory_history_events(&self.db.pool, &req, ¬e_row).await?; + + Ok(MemoryHistoryResponse { + schema: MEMORY_HISTORY_SCHEMA_V1.to_string(), + note_id: req.note_id, + events, }) } } @@ -414,6 +560,248 @@ fn to_recent_trace(item: NoteRecentTraceRow) -> NoteProvenanceRecentTrace { } } +fn version_history_event( + version: &NoteProvenanceNoteVersion, + decision: Option<&&NoteProvenanceIngestDecision>, +) -> MemoryHistoryEvent { + let event_type = version_event_type(version.op.as_str(), version.reason.as_str()); + let related_decision_id = decision.map(|decision| decision.decision_id); + let details = serde_json::json!({ + "reason": version.reason, + "prev_snapshot": version.prev_snapshot, + "new_snapshot": version.new_snapshot, + "ingest_decision": decision.map(|decision| serde_json::json!({ + "decision_id": decision.decision_id, + "pipeline": decision.pipeline, + "base_decision": decision.base_decision, + "policy_decision": decision.policy_decision, + "note_op": decision.note_op, + "reason_code": decision.reason_code, + })), + }); + + MemoryHistoryEvent { + event_id: format!("memory_note_versions:{}", version.version_id), + event_type: event_type.to_string(), + subject_type: "note".to_string(), + note_id: version.note_id, + source_table: "memory_note_versions".to_string(), + source_id: Some(version.version_id), + related_note_version_id: Some(version.version_id), + related_decision_id, + related_proposal_id: None, + actor: Some(version.actor.clone()), + op: Some(version.op.clone()), + reason_code: None, + summary: version_summary(event_type, version.reason.as_str()), + details, + ts: version.ts, + } +} + +fn decision_history_event( + note_id: Uuid, + decision: &NoteProvenanceIngestDecision, +) -> MemoryHistoryEvent { + let event_type = decision_event_type(decision); + let details = serde_json::json!({ + "pipeline": decision.pipeline, + "note_type": decision.note_type, + "note_key": decision.note_key, + "base_decision": decision.base_decision, + "policy_decision": decision.policy_decision, + "note_op": decision.note_op, + "details": decision.details, + }); + + MemoryHistoryEvent { + event_id: format!("memory_ingest_decisions:{}", decision.decision_id), + event_type: event_type.to_string(), + subject_type: "note".to_string(), + note_id, + source_table: "memory_ingest_decisions".to_string(), + source_id: Some(decision.decision_id), + related_note_version_id: decision.note_version_id, + related_decision_id: Some(decision.decision_id), + related_proposal_id: None, + actor: Some(decision.agent_id.clone()), + op: Some(decision.note_op.clone()), + reason_code: decision.reason_code.clone(), + summary: decision_summary(event_type, decision), + details, + ts: decision.ts, + } +} + +fn expire_history_event(note: &MemoryNote, expires_at: OffsetDateTime) -> MemoryHistoryEvent { + MemoryHistoryEvent { + event_id: format!("memory_notes:{}:expire:{expires_at}", note.note_id), + event_type: "expire".to_string(), + subject_type: "note".to_string(), + note_id: note.note_id, + source_table: "memory_notes".to_string(), + source_id: Some(note.note_id), + related_note_version_id: None, + related_decision_id: None, + related_proposal_id: None, + actor: Some(note.agent_id.clone()), + op: Some("EXPIRE".to_string()), + reason_code: None, + summary: "Note reached its persisted expires_at timestamp.".to_string(), + details: serde_json::json!({ + "status": note.status, + "expires_at": expires_at, + }), + ts: expires_at, + } +} + +fn derived_proposal_history_event( + note_id: Uuid, + proposal: NoteDerivedProposalRow, +) -> MemoryHistoryEvent { + MemoryHistoryEvent { + event_id: format!("consolidation_proposals:{}", proposal.proposal_id), + event_type: "derived".to_string(), + subject_type: "note".to_string(), + note_id, + source_table: "consolidation_proposals".to_string(), + source_id: Some(proposal.proposal_id), + related_note_version_id: None, + related_decision_id: None, + related_proposal_id: Some(proposal.proposal_id), + actor: Some(proposal.agent_id), + op: Some(proposal.apply_intent.clone()), + reason_code: None, + summary: format!( + "Derived proposal '{}' was created with review_state '{}'.", + proposal.proposal_kind, proposal.review_state + ), + details: serde_json::json!({ + "run_id": proposal.run_id, + "proposal_kind": proposal.proposal_kind, + "apply_intent": proposal.apply_intent, + "review_state": proposal.review_state, + "source_refs": proposal.source_refs, + "source_snapshot": proposal.source_snapshot, + "lineage": proposal.lineage, + "diff": proposal.diff, + "confidence": proposal.confidence, + "target_ref": proposal.target_ref, + "proposed_payload": proposal.proposed_payload, + }), + ts: proposal.created_at, + } +} + +fn proposal_review_history_event( + note_id: Uuid, + review: NoteProposalReviewRow, +) -> MemoryHistoryEvent { + let event_type = proposal_review_event_type(review.action.as_str()); + + MemoryHistoryEvent { + event_id: format!("consolidation_proposal_reviews:{}", review.review_id), + event_type: event_type.to_string(), + subject_type: "note".to_string(), + note_id, + source_table: "consolidation_proposal_reviews".to_string(), + source_id: Some(review.review_id), + related_note_version_id: None, + related_decision_id: None, + related_proposal_id: Some(review.proposal_id), + actor: Some(review.reviewer_agent_id), + op: Some(review.action.clone()), + reason_code: None, + summary: format!( + "Proposal review action '{}' moved '{}' from '{}' to '{}'.", + review.action, review.proposal_kind, review.from_review_state, review.to_review_state + ), + details: serde_json::json!({ + "proposal_id": review.proposal_id, + "run_id": review.run_id, + "proposal_kind": review.proposal_kind, + "apply_intent": review.apply_intent, + "from_review_state": review.from_review_state, + "to_review_state": review.to_review_state, + "review_comment": review.review_comment, + "diff": review.diff, + }), + ts: review.created_at, + } +} + +fn should_emit_decision_event(decision: &NoteProvenanceIngestDecision) -> bool { + if matches!(decision.note_op.as_str(), "NONE" | "REJECTED") { + return true; + } + + decision.note_version_id.is_none() +} + +fn version_event_type(op: &str, reason: &str) -> &'static str { + let reason = reason.to_ascii_lowercase(); + + match op { + "ADD" => "add", + "UPDATE" => "update", + "DELETE" if reason.contains("expire") => "expire", + "DELETE" => "delete", + "PUBLISH" | "UNPUBLISH" => "related", + "DEPRECATE" | "INVALIDATE" => "invalidated", + _ => "related", + } +} + +fn decision_event_type(decision: &NoteProvenanceIngestDecision) -> &'static str { + if decision.policy_decision == "reject" || decision.note_op == "REJECTED" { + return "reject"; + } + if decision.policy_decision == "ignore" || decision.note_op == "NONE" { + return "ignore"; + } + + match decision.note_op.as_str() { + "ADD" => "add", + "UPDATE" => "update", + "DELETE" => "delete", + _ => "related", + } +} + +fn proposal_review_event_type(action: &str) -> &'static str { + match action { + "apply" => "applied", + "discard" | "defer" => "invalidated", + "approve" => "related", + _ => "related", + } +} + +fn version_summary(event_type: &str, reason: &str) -> String { + match event_type { + "add" => format!("Note was added by {reason}."), + "update" => format!("Note was updated by {reason}."), + "delete" => format!("Note was deleted by {reason}."), + "expire" => format!("Note expired through {reason}."), + "invalidated" => format!("Note was invalidated by {reason}."), + _ => format!("Note recorded related transition {reason}."), + } +} + +fn decision_summary(event_type: &str, decision: &NoteProvenanceIngestDecision) -> String { + let reason = decision.reason_code.as_deref().unwrap_or("no_reason_code"); + + match event_type { + "ignore" => format!("Ingestion ignored candidate memory with {reason}."), + "reject" => format!("Ingestion rejected candidate memory with {reason}."), + _ => format!( + "Ingestion recorded {} decision for operation {}.", + decision.policy_decision, decision.note_op + ), + } +} + async fn load_ingest_decisions( pool: &PgPool, req: &ValidatedNoteProvenanceRequest, @@ -430,6 +818,7 @@ SELECT note_type, note_key, note_id, + note_version_id, base_decision, policy_decision, note_op, @@ -556,6 +945,142 @@ LIMIT $4", Ok(rows.into_iter().map(to_recent_trace).collect()) } +async fn load_memory_history_events( + pool: &PgPool, + req: &ValidatedNoteProvenanceRequest, + note: &MemoryNote, +) -> Result> { + let decisions = load_ingest_decisions(pool, req).await?; + let versions = load_note_versions(pool, &req.tenant_id, &req.project_id, req.note_id).await?; + let proposal_ref = serde_json::json!([{ "kind": "note", "id": req.note_id }]); + let proposals = load_derived_proposals_for_note(pool, req, &proposal_ref).await?; + let reviews = load_proposal_reviews_for_note(pool, req, &proposal_ref).await?; + let mut decision_by_version = HashMap::new(); + + for decision in &decisions { + if let Some(version_id) = decision.note_version_id { + decision_by_version.insert(version_id, decision); + } + } + + let mut events = Vec::new(); + + for version in &versions { + events.push(version_history_event(version, decision_by_version.get(&version.version_id))); + } + for decision in &decisions { + if should_emit_decision_event(decision) { + events.push(decision_history_event(req.note_id, decision)); + } + } + + if let Some(expires_at) = note.expires_at + && expires_at <= OffsetDateTime::now_utc() + && !events.iter().any(|event| event.event_type == "expire") + { + events.push(expire_history_event(note, expires_at)); + } + + for proposal in proposals { + events.push(derived_proposal_history_event(req.note_id, proposal)); + } + for review in reviews { + events.push(proposal_review_history_event(req.note_id, review)); + } + + events.sort_by(|left, right| { + left.ts.cmp(&right.ts).then_with(|| left.event_id.cmp(&right.event_id)) + }); + + let history_limit = NOTE_PROVENANCE_HISTORY_LIMIT as usize; + + if events.len() > history_limit { + let drop_count = events.len() - history_limit; + + events.drain(0..drop_count); + } + + Ok(events) +} + +async fn load_derived_proposals_for_note( + pool: &PgPool, + req: &ValidatedNoteProvenanceRequest, + proposal_ref: &Value, +) -> Result> { + let rows = sqlx::query_as::<_, NoteDerivedProposalRow>( + "\ +SELECT + proposal_id, + run_id, + agent_id, + proposal_kind, + apply_intent, + review_state, + source_refs, + source_snapshot, + lineage, + diff, + confidence, + COALESCE(target_ref, '{}'::jsonb) AS target_ref, + COALESCE(proposed_payload, '{}'::jsonb) AS proposed_payload, + created_at +FROM consolidation_proposals +WHERE tenant_id = $1 + AND project_id = $2 + AND source_refs @> $3 +ORDER BY created_at DESC, proposal_id DESC +LIMIT $4", + ) + .bind(&req.tenant_id) + .bind(&req.project_id) + .bind(proposal_ref) + .bind(NOTE_PROVENANCE_HISTORY_LIMIT) + .fetch_all(pool) + .await?; + + Ok(rows) +} + +async fn load_proposal_reviews_for_note( + pool: &PgPool, + req: &ValidatedNoteProvenanceRequest, + proposal_ref: &Value, +) -> Result> { + let rows = sqlx::query_as::<_, NoteProposalReviewRow>( + "\ +SELECT + reviews.review_id, + reviews.proposal_id, + reviews.run_id, + reviews.reviewer_agent_id, + reviews.action, + reviews.from_review_state, + reviews.to_review_state, + reviews.review_comment, + reviews.created_at, + proposals.proposal_kind, + proposals.apply_intent, + proposals.diff +FROM consolidation_proposal_reviews reviews +JOIN consolidation_proposals proposals + ON proposals.proposal_id = reviews.proposal_id +WHERE reviews.tenant_id = $1 + AND reviews.project_id = $2 + AND proposals.source_refs @> $3 +ORDER BY reviews.created_at DESC, reviews.review_id DESC +LIMIT $4", + ) + .bind(&req.tenant_id) + .bind(&req.project_id) + .bind(proposal_ref) + .bind(NOTE_PROVENANCE_HISTORY_LIMIT) + .fetch_all(pool) + .await?; + + Ok(rows) +} + #[cfg(test)] mod tests { use uuid::Uuid; diff --git a/packages/elf-service/tests/acceptance/memory_history.rs b/packages/elf-service/tests/acceptance/memory_history.rs new file mode 100644 index 00000000..f803067d --- /dev/null +++ b/packages/elf-service/tests/acceptance/memory_history.rs @@ -0,0 +1,138 @@ +use std::{ + collections::HashSet, + sync::{Arc, atomic::AtomicUsize}, +}; + +use crate::acceptance::{self, SpyExtractor, StubEmbedding, StubRerank}; +use elf_service::{ + AddNoteInput, AddNoteRequest, MemoryHistoryGetRequest, NoteOp, NoteProvenanceGetRequest, + Providers, +}; + +fn history_request(text: &str, importance: f32) -> AddNoteRequest { + AddNoteRequest { + tenant_id: "tenant-history".to_string(), + project_id: "project-history".to_string(), + agent_id: "agent-history".to_string(), + scope: "agent_private".to_string(), + notes: vec![AddNoteInput { + r#type: "fact".to_string(), + key: Some("memory_history_target".to_string()), + text: text.to_string(), + structured: None, + importance, + confidence: 0.9, + ttl_days: None, + source_ref: serde_json::json!({ "schema": "acceptance/history" }), + write_policy: None, + }], + } +} + +#[tokio::test] +#[ignore = "Requires external Postgres and Qdrant. Set ELF_PG_DSN and ELF_QDRANT_URL to run."] +async fn memory_history_links_versions_and_ignored_decisions() { + let Some(test_db) = acceptance::test_db().await else { + eprintln!("Skipping memory_history_links_versions_and_ignored_decisions; set ELF_PG_DSN."); + + return; + }; + let Some(qdrant_url) = acceptance::test_qdrant_url() else { + eprintln!( + "Skipping memory_history_links_versions_and_ignored_decisions; set ELF_QDRANT_URL." + ); + + return; + }; + let providers = Providers::new( + Arc::new(StubEmbedding { vector_dim: 4_096 }), + Arc::new(StubRerank), + Arc::new(SpyExtractor { + calls: Arc::new(AtomicUsize::new(0)), + payload: serde_json::json!({ "notes": [] }), + }), + ); + let collection = test_db.collection_name("elf_history"); + let docs_collection = test_db.collection_name("elf_history_docs"); + let cfg = acceptance::test_config( + test_db.dsn().to_string(), + qdrant_url, + 4_096, + collection, + docs_collection, + ); + let service = + acceptance::build_service(cfg, providers).await.expect("Failed to build service."); + + acceptance::reset_db(&service.db.pool).await.expect("Failed to reset test database."); + + let first = service + .add_note(history_request( + "Fact: Memory history readback starts with original evidence.", + 0.7, + )) + .await + .expect("initial note should be added"); + let note_id = first.results[0].note_id.expect("add should return note id"); + + assert_eq!(first.results[0].op, NoteOp::Add); + + let updated = service + .add_note(history_request("Fact: Memory history readback records updated evidence.", 0.8)) + .await + .expect("second note should update by key"); + let ignored = service + .add_note(history_request("Fact: Memory history readback records updated evidence.", 0.8)) + .await + .expect("third note should be ignored as unchanged"); + + assert_eq!(updated.results[0].op, NoteOp::Update); + assert_eq!(ignored.results[0].op, NoteOp::None); + + let history = service + .memory_history_get(MemoryHistoryGetRequest { + tenant_id: "tenant-history".to_string(), + project_id: "project-history".to_string(), + note_id, + }) + .await + .expect("history should be readable"); + let event_types: HashSet<&str> = + history.events.iter().map(|event| event.event_type.as_str()).collect(); + + assert_eq!(history.schema, "elf.memory_history/v1"); + assert!(event_types.contains("add")); + assert!(event_types.contains("update")); + assert!(event_types.contains("ignore")); + assert!( + history + .events + .iter() + .filter(|event| matches!(event.event_type.as_str(), "add" | "update")) + .all(|event| event.related_decision_id.is_some() + && event.related_note_version_id.is_some()) + ); + + let linked_decision_count: i64 = sqlx::query_scalar( + "SELECT count(*) FROM memory_ingest_decisions WHERE note_id = $1 AND note_version_id IS NOT NULL", + ) + .bind(note_id) + .fetch_one(&service.db.pool) + .await + .expect("linked decision count should be queryable"); + + assert_eq!(linked_decision_count, 2); + + let provenance = service + .note_provenance_get(NoteProvenanceGetRequest { + tenant_id: "tenant-history".to_string(), + project_id: "project-history".to_string(), + note_id, + }) + .await + .expect("provenance should include history"); + + assert_eq!(provenance.history.len(), history.events.len()); + + test_db.cleanup().await.expect("Failed to cleanup test database."); +} diff --git a/packages/elf-service/tests/acceptance/suite.rs b/packages/elf-service/tests/acceptance/suite.rs index e7d102ef..7db8daac 100644 --- a/packages/elf-service/tests/acceptance/suite.rs +++ b/packages/elf-service/tests/acceptance/suite.rs @@ -8,6 +8,7 @@ mod evidence_binding; mod graph_ingestion; mod idempotency; mod knowledge_pages; +mod memory_history; mod outbox_eventual_consistency; mod rebuild_qdrant; mod sot_vectors; diff --git a/sql/tables/023_memory_ingest_decisions.sql b/sql/tables/023_memory_ingest_decisions.sql index e90aa54a..b08843c6 100644 --- a/sql/tables/023_memory_ingest_decisions.sql +++ b/sql/tables/023_memory_ingest_decisions.sql @@ -8,6 +8,7 @@ CREATE TABLE IF NOT EXISTS memory_ingest_decisions ( note_type text NOT NULL, note_key text NULL, note_id uuid NULL, + note_version_id uuid NULL, base_decision text NOT NULL, policy_decision text NOT NULL, note_op text NOT NULL, @@ -21,13 +22,18 @@ CREATE TABLE IF NOT EXISTS memory_ingest_decisions ( CONSTRAINT ck_memory_ingest_decisions_policy_decision CHECK (policy_decision IN ('remember', 'update', 'ignore', 'reject')), CONSTRAINT ck_memory_ingest_decisions_note_op - CHECK (note_op IN ('ADD', 'UPDATE', 'NONE', 'DELETE', 'REJECTED')) + CHECK (note_op IN ('ADD', 'UPDATE', 'NONE', 'DELETE', 'REJECTED')) ); +ALTER TABLE memory_ingest_decisions + ADD COLUMN IF NOT EXISTS note_version_id uuid NULL; + CREATE INDEX IF NOT EXISTS idx_memory_ingest_decisions_context ON memory_ingest_decisions (tenant_id, project_id, agent_id, ts desc); CREATE INDEX IF NOT EXISTS idx_memory_ingest_decisions_note_id ON memory_ingest_decisions (note_id); +CREATE INDEX IF NOT EXISTS idx_memory_ingest_decisions_note_version_id + ON memory_ingest_decisions (note_version_id); CREATE INDEX IF NOT EXISTS idx_memory_ingest_decisions_policy_decision ON memory_ingest_decisions (policy_decision); CREATE INDEX IF NOT EXISTS idx_memory_ingest_decisions_pipeline