diff --git a/apps/elf-eval/src/bin/real_world_live_adapter/elf_domain_materializers.rs b/apps/elf-eval/src/bin/real_world_live_adapter/elf_domain_materializers.rs index 256d1d6f..a89719c3 100644 --- a/apps/elf-eval/src/bin/real_world_live_adapter/elf_domain_materializers.rs +++ b/apps/elf-eval/src/bin/real_world_live_adapter/elf_domain_materializers.rs @@ -1,257 +1,6 @@ -use crate::{ - AGENT_ID, AddNoteInput, AddNoteRequest, BaselineRuntime, ConsolidationLineage, - ConsolidationMaterializationEvidence, ConsolidationProposalResponse, - ConsolidationProposalReviewRequest, ConsolidationProposalsListRequest, - ConsolidationRunCreateRequest, ElfService, IngestedCorpus, KnowledgeMaterializationEvidence, - KnowledgePageKind, KnowledgePageLintRequest, KnowledgePageRebuildRequest, - KnowledgePageSearchRequest, LiveConsolidationFixture, LoadedJob, Result, SCOPE, TENANT_ID, - Uuid, Value, eyre, serde_json, -}; - -pub(super) async fn materialize_elf_consolidation( - runtime: &BaselineRuntime, - service: &ElfService, - loaded: &LoadedJob, - ingested: &IngestedCorpus, - adapter_id: &str, -) -> Result<(Option, Option, Option)> { - if loaded.job.suite != "consolidation" { - return Ok((None, None, None)); - } - - let project_id = crate::project_id_for_job(&loaded.job.job_id); - let fixture = crate::live_consolidation_fixture(loaded)?; - let corpus = crate::corpus_texts(loaded)?; - let prepared = - crate::prepare_consolidation_run(loaded, adapter_id, ingested, &fixture, &corpus)?; - let run = service - .consolidation_run_create(ConsolidationRunCreateRequest { - tenant_id: TENANT_ID.to_string(), - project_id: project_id.clone(), - agent_id: AGENT_ID.to_string(), - job_kind: "fixture".to_string(), - input_refs: prepared.input_refs.clone(), - source_snapshot: serde_json::json!({ - "schema": "real_world_live_consolidation_run_snapshot/v1", - "adapter_id": adapter_id, - "job_id": loaded.job.job_id, - "source_ref_count": prepared.input_refs.len() - }), - lineage: ConsolidationLineage { - source_refs: prepared.input_refs.clone(), - parent_run_id: None, - parent_proposal_ids: Vec::new(), - }, - proposals: prepared.proposals, - }) - .await - .map_err(|err| { - eyre::eyre!("ELF consolidation_run_create failed for {}: {err}", loaded.job.job_id) - })?; - - crate::run_worker(runtime).await?; - - let reviewed = review_live_consolidation_proposals( - service, - loaded, - project_id.as_str(), - run.run.run_id, - &fixture, - ) - .await?; - let consolidation_response = crate::live_consolidation_response(&fixture, &reviewed)?; - let evidence = crate::consolidation_materialization_evidence( - run.run.run_id, - &fixture, - &prepared.input_refs, - &reviewed, - ); - - Ok((Some(consolidation_response), Some(evidence), None)) -} - -pub(super) async fn materialize_elf_knowledge( - service: &ElfService, - loaded: &LoadedJob, - ingested: &IngestedCorpus, - adapter_id: &str, -) -> Result<(Vec, Option, Option)> { - if loaded.job.suite != "knowledge_compilation" { - return Ok((Vec::new(), None, None)); - } - - let project_id = crate::project_id_for_job(&loaded.job.job_id); - let note_ids = crate::live_note_ids(ingested); - - if note_ids.is_empty() { - return Err(eyre::eyre!( - "{} has no live note sources for knowledge rebuild.", - loaded.job.job_id - )); - } - - let page_key = crate::slug(&loaded.job.job_id); - let request = KnowledgePageRebuildRequest { - tenant_id: TENANT_ID.to_string(), - project_id: project_id.clone(), - agent_id: AGENT_ID.to_string(), - page_kind: KnowledgePageKind::Project, - page_key, - title: Some(loaded.job.title.clone()), - doc_ids: Vec::new(), - doc_chunk_ids: Vec::new(), - note_ids: note_ids.clone(), - event_ids: Vec::new(), - relation_ids: Vec::new(), - proposal_ids: Vec::new(), - provider_metadata: serde_json::json!({ - "adapter_id": adapter_id, - "job_id": loaded.job.job_id, - "llm_derived": false, - "runtime_path": "ElfService::knowledge_page_rebuild" - }), - }; - let first = service.knowledge_page_rebuild(request.clone()).await.map_err(|err| { - eyre::eyre!("ELF knowledge_page_rebuild failed for {}: {err}", loaded.job.job_id) - })?; - let second = service.knowledge_page_rebuild(request).await.map_err(|err| { - eyre::eyre!("ELF second knowledge_page_rebuild failed for {}: {err}", loaded.job.job_id) - })?; +mod consolidation; +mod knowledge; - update_stale_trap_sources(service, loaded, adapter_id, project_id.as_str()).await?; - - let lint = service - .knowledge_page_lint(KnowledgePageLintRequest { - tenant_id: TENANT_ID.to_string(), - project_id: project_id.clone(), - page_id: second.page.page.page_id, - }) - .await - .map_err(|err| { - eyre::eyre!("ELF knowledge_page_lint failed for {}: {err}", loaded.job.job_id) - })?; - let search = service - .knowledge_pages_search(KnowledgePageSearchRequest { - tenant_id: TENANT_ID.to_string(), - project_id, - agent_id: AGENT_ID.to_string(), - read_profile: "private_only".to_string(), - query: "source notes".to_string(), - page_kind: Some(KnowledgePageKind::Project), - limit: Some(10), - }) - .await - .map_err(|err| { - eyre::eyre!("ELF knowledge_pages_search failed for {}: {err}", loaded.job.job_id) - })?; - let page = crate::knowledge_page_artifact(loaded, ingested, &first.page, &second.page, &lint)?; - let evidence = - crate::knowledge_materialization_evidence(&second.page, &lint, search.items.len()); - - Ok((vec![page], Some(evidence), None)) -} - -async fn review_live_consolidation_proposals( - service: &ElfService, - loaded: &LoadedJob, - project_id: &str, - run_id: Uuid, - fixture: &LiveConsolidationFixture, -) -> Result> { - let listed = service - .consolidation_proposals_list(ConsolidationProposalsListRequest { - tenant_id: TENANT_ID.to_string(), - project_id: project_id.to_string(), - run_id: Some(run_id), - review_state: None, - limit: Some(100), - }) - .await - .map_err(|err| { - eyre::eyre!("ELF consolidation proposal list failed for {}: {err}", loaded.job.job_id) - })?; - let mut reviewed = Vec::new(); - - for (index, proposal) in listed.proposals.into_iter().enumerate() { - let fixture_proposal = fixture.proposals.get(index).ok_or_else(|| { - eyre::eyre!( - "ELF consolidation materialized extra proposal {} for {}.", - proposal.proposal_id, - loaded.job.job_id - ) - })?; - let review_action = - crate::consolidation_review_action(fixture_proposal.actual_review_action.as_str())?; - - reviewed.push( - service - .consolidation_proposal_review(ConsolidationProposalReviewRequest { - tenant_id: TENANT_ID.to_string(), - project_id: project_id.to_string(), - reviewer_agent_id: AGENT_ID.to_string(), - proposal_id: proposal.proposal_id, - review_action, - review_comment: Some( - "Live adapter review transition for real-world benchmark evidence." - .to_string(), - ), - }) - .await - .map_err(|err| { - eyre::eyre!( - "ELF consolidation proposal review failed for {}: {err}", - loaded.job.job_id - ) - })?, - ); - } - - crate::validate_reviewed_consolidation_count(loaded, fixture, &reviewed)?; - - Ok(reviewed) -} - -async fn update_stale_trap_sources( - service: &ElfService, - loaded: &LoadedJob, - adapter_id: &str, - project_id: &str, -) -> Result<()> { - for evidence_id in crate::stale_trap_evidence_ids(loaded) { - service - .add_note(AddNoteRequest { - tenant_id: TENANT_ID.to_string(), - project_id: project_id.to_string(), - agent_id: AGENT_ID.to_string(), - scope: SCOPE.to_string(), - notes: vec![AddNoteInput { - r#type: "fact".to_string(), - key: Some(evidence_id.clone()), - text: format!( - "Current lint probe: evidence {evidence_id} changed after the knowledge page rebuild and should mark the derived page source snapshot stale." - ), - structured: None, - importance: 0.9, - confidence: 0.95, - ttl_days: None, - source_ref: serde_json::json!({ - "schema": "real_world_live_adapter/v1", - "adapter": adapter_id, - "job_id": loaded.job.job_id, - "evidence_id": evidence_id, - "lint_probe": "stale_source_ref" - }), - write_policy: None, - }], - }) - .await - .map_err(|err| { - eyre::eyre!( - "ELF add_note stale-source update failed for {}: {err}", - loaded.job.job_id - ) - })?; - } - - Ok(()) -} +pub(super) use self::{ + consolidation::materialize_elf_consolidation, knowledge::materialize_elf_knowledge, +}; diff --git a/apps/elf-eval/src/bin/real_world_live_adapter/elf_domain_materializers/consolidation.rs b/apps/elf-eval/src/bin/real_world_live_adapter/elf_domain_materializers/consolidation.rs new file mode 100644 index 00000000..18bd5c67 --- /dev/null +++ b/apps/elf-eval/src/bin/real_world_live_adapter/elf_domain_materializers/consolidation.rs @@ -0,0 +1,128 @@ +use crate::{ + AGENT_ID, BaselineRuntime, ConsolidationLineage, ConsolidationMaterializationEvidence, + ConsolidationProposalResponse, ConsolidationProposalReviewRequest, + ConsolidationProposalsListRequest, ConsolidationRunCreateRequest, ElfService, IngestedCorpus, + LiveConsolidationFixture, LoadedJob, Result, TENANT_ID, Uuid, Value, eyre, serde_json, +}; + +pub(crate) async fn materialize_elf_consolidation( + runtime: &BaselineRuntime, + service: &ElfService, + loaded: &LoadedJob, + ingested: &IngestedCorpus, + adapter_id: &str, +) -> Result<(Option, Option, Option)> { + if loaded.job.suite != "consolidation" { + return Ok((None, None, None)); + } + + let project_id = crate::project_id_for_job(&loaded.job.job_id); + let fixture = crate::live_consolidation_fixture(loaded)?; + let corpus = crate::corpus_texts(loaded)?; + let prepared = + crate::prepare_consolidation_run(loaded, adapter_id, ingested, &fixture, &corpus)?; + let run = service + .consolidation_run_create(ConsolidationRunCreateRequest { + tenant_id: TENANT_ID.to_string(), + project_id: project_id.clone(), + agent_id: AGENT_ID.to_string(), + job_kind: "fixture".to_string(), + input_refs: prepared.input_refs.clone(), + source_snapshot: serde_json::json!({ + "schema": "real_world_live_consolidation_run_snapshot/v1", + "adapter_id": adapter_id, + "job_id": loaded.job.job_id, + "source_ref_count": prepared.input_refs.len() + }), + lineage: ConsolidationLineage { + source_refs: prepared.input_refs.clone(), + parent_run_id: None, + parent_proposal_ids: Vec::new(), + }, + proposals: prepared.proposals, + }) + .await + .map_err(|err| { + eyre::eyre!("ELF consolidation_run_create failed for {}: {err}", loaded.job.job_id) + })?; + + crate::run_worker(runtime).await?; + + let reviewed = review_live_consolidation_proposals( + service, + loaded, + project_id.as_str(), + run.run.run_id, + &fixture, + ) + .await?; + let consolidation_response = crate::live_consolidation_response(&fixture, &reviewed)?; + let evidence = crate::consolidation_materialization_evidence( + run.run.run_id, + &fixture, + &prepared.input_refs, + &reviewed, + ); + + Ok((Some(consolidation_response), Some(evidence), None)) +} + +async fn review_live_consolidation_proposals( + service: &ElfService, + loaded: &LoadedJob, + project_id: &str, + run_id: Uuid, + fixture: &LiveConsolidationFixture, +) -> Result> { + let listed = service + .consolidation_proposals_list(ConsolidationProposalsListRequest { + tenant_id: TENANT_ID.to_string(), + project_id: project_id.to_string(), + run_id: Some(run_id), + review_state: None, + limit: Some(100), + }) + .await + .map_err(|err| { + eyre::eyre!("ELF consolidation proposal list failed for {}: {err}", loaded.job.job_id) + })?; + let mut reviewed = Vec::new(); + + for (index, proposal) in listed.proposals.into_iter().enumerate() { + let fixture_proposal = fixture.proposals.get(index).ok_or_else(|| { + eyre::eyre!( + "ELF consolidation materialized extra proposal {} for {}.", + proposal.proposal_id, + loaded.job.job_id + ) + })?; + let review_action = + crate::consolidation_review_action(fixture_proposal.actual_review_action.as_str())?; + + reviewed.push( + service + .consolidation_proposal_review(ConsolidationProposalReviewRequest { + tenant_id: TENANT_ID.to_string(), + project_id: project_id.to_string(), + reviewer_agent_id: AGENT_ID.to_string(), + proposal_id: proposal.proposal_id, + review_action, + review_comment: Some( + "Live adapter review transition for real-world benchmark evidence." + .to_string(), + ), + }) + .await + .map_err(|err| { + eyre::eyre!( + "ELF consolidation proposal review failed for {}: {err}", + loaded.job.job_id + ) + })?, + ); + } + + crate::validate_reviewed_consolidation_count(loaded, fixture, &reviewed)?; + + Ok(reviewed) +} diff --git a/apps/elf-eval/src/bin/real_world_live_adapter/elf_domain_materializers/knowledge.rs b/apps/elf-eval/src/bin/real_world_live_adapter/elf_domain_materializers/knowledge.rs new file mode 100644 index 00000000..33977e9c --- /dev/null +++ b/apps/elf-eval/src/bin/real_world_live_adapter/elf_domain_materializers/knowledge.rs @@ -0,0 +1,132 @@ +use crate::{ + AGENT_ID, AddNoteInput, AddNoteRequest, ElfService, IngestedCorpus, + KnowledgeMaterializationEvidence, KnowledgePageKind, KnowledgePageLintRequest, + KnowledgePageRebuildRequest, KnowledgePageSearchRequest, LoadedJob, Result, SCOPE, TENANT_ID, + Value, eyre, serde_json, +}; + +pub(crate) async fn materialize_elf_knowledge( + service: &ElfService, + loaded: &LoadedJob, + ingested: &IngestedCorpus, + adapter_id: &str, +) -> Result<(Vec, Option, Option)> { + if loaded.job.suite != "knowledge_compilation" { + return Ok((Vec::new(), None, None)); + } + + let project_id = crate::project_id_for_job(&loaded.job.job_id); + let note_ids = crate::live_note_ids(ingested); + + if note_ids.is_empty() { + return Err(eyre::eyre!( + "{} has no live note sources for knowledge rebuild.", + loaded.job.job_id + )); + } + + let page_key = crate::slug(&loaded.job.job_id); + let request = KnowledgePageRebuildRequest { + tenant_id: TENANT_ID.to_string(), + project_id: project_id.clone(), + agent_id: AGENT_ID.to_string(), + page_kind: KnowledgePageKind::Project, + page_key, + title: Some(loaded.job.title.clone()), + doc_ids: Vec::new(), + doc_chunk_ids: Vec::new(), + note_ids: note_ids.clone(), + event_ids: Vec::new(), + relation_ids: Vec::new(), + proposal_ids: Vec::new(), + provider_metadata: serde_json::json!({ + "adapter_id": adapter_id, + "job_id": loaded.job.job_id, + "llm_derived": false, + "runtime_path": "ElfService::knowledge_page_rebuild" + }), + }; + let first = service.knowledge_page_rebuild(request.clone()).await.map_err(|err| { + eyre::eyre!("ELF knowledge_page_rebuild failed for {}: {err}", loaded.job.job_id) + })?; + let second = service.knowledge_page_rebuild(request).await.map_err(|err| { + eyre::eyre!("ELF second knowledge_page_rebuild failed for {}: {err}", loaded.job.job_id) + })?; + + update_stale_trap_sources(service, loaded, adapter_id, project_id.as_str()).await?; + + let lint = service + .knowledge_page_lint(KnowledgePageLintRequest { + tenant_id: TENANT_ID.to_string(), + project_id: project_id.clone(), + page_id: second.page.page.page_id, + }) + .await + .map_err(|err| { + eyre::eyre!("ELF knowledge_page_lint failed for {}: {err}", loaded.job.job_id) + })?; + let search = service + .knowledge_pages_search(KnowledgePageSearchRequest { + tenant_id: TENANT_ID.to_string(), + project_id, + agent_id: AGENT_ID.to_string(), + read_profile: "private_only".to_string(), + query: "source notes".to_string(), + page_kind: Some(KnowledgePageKind::Project), + limit: Some(10), + }) + .await + .map_err(|err| { + eyre::eyre!("ELF knowledge_pages_search failed for {}: {err}", loaded.job.job_id) + })?; + let page = crate::knowledge_page_artifact(loaded, ingested, &first.page, &second.page, &lint)?; + let evidence = + crate::knowledge_materialization_evidence(&second.page, &lint, search.items.len()); + + Ok((vec![page], Some(evidence), None)) +} + +async fn update_stale_trap_sources( + service: &ElfService, + loaded: &LoadedJob, + adapter_id: &str, + project_id: &str, +) -> Result<()> { + for evidence_id in crate::stale_trap_evidence_ids(loaded) { + service + .add_note(AddNoteRequest { + tenant_id: TENANT_ID.to_string(), + project_id: project_id.to_string(), + agent_id: AGENT_ID.to_string(), + scope: SCOPE.to_string(), + notes: vec![AddNoteInput { + r#type: "fact".to_string(), + key: Some(evidence_id.clone()), + text: format!( + "Current lint probe: evidence {evidence_id} changed after the knowledge page rebuild and should mark the derived page source snapshot stale." + ), + structured: None, + importance: 0.9, + confidence: 0.95, + ttl_days: None, + source_ref: serde_json::json!({ + "schema": "real_world_live_adapter/v1", + "adapter": adapter_id, + "job_id": loaded.job.job_id, + "evidence_id": evidence_id, + "lint_probe": "stale_source_ref" + }), + write_policy: None, + }], + }) + .await + .map_err(|err| { + eyre::eyre!( + "ELF add_note stale-source update failed for {}: {err}", + loaded.job.job_id + ) + })?; + } + + Ok(()) +} diff --git a/apps/elf-eval/src/bin/real_world_live_adapter/elf_runtime.rs b/apps/elf-eval/src/bin/real_world_live_adapter/elf_runtime.rs index 1a9388d9..00f432a6 100644 --- a/apps/elf-eval/src/bin/real_world_live_adapter/elf_runtime.rs +++ b/apps/elf-eval/src/bin/real_world_live_adapter/elf_runtime.rs @@ -1,253 +1,5 @@ -use crate::{ - AGENT_ID, AdapterKind, BaselineRuntime, CommandEvidence, ConsolidationMaterializationEvidence, - DreamingReadbackOutput, ElfArgs, ElfService, IngestedCorpus, Instant, - KnowledgeMaterializationEvidence, LoadedJob, MaterializedJob, MaterializedJobInput, - MaterializedOutput, PayloadLevel, Result, SearchRequest, SearchResponse, - SuiteMaterializationSelectionInput, TENANT_ID, TestDatabase, Uuid, Value, aggregate_status, - env, eyre, -}; +mod jobs; +mod search; +mod surfaces; -struct OptionalElfMaterializations { - pages: Vec, - knowledge: Option, - consolidation_response: Option, - consolidation: Option, - dreaming_readback: Option, - failure: Option, -} - -pub(super) async fn run_elf(args: ElfArgs) -> Result<()> { - let jobs = crate::load_jobs(&args.fixtures)?; - let result = materialize_elf_jobs(&args, &jobs).await; - let materialized = match result { - Ok(jobs) => jobs, - Err(err) => - crate::failure_jobs(&args.adapter_id, &jobs, "elf_service_runtime", err.to_string()), - }; - - crate::write_materialized_output(MaterializedOutput { - adapter_id: &args.adapter_id, - adapter_kind: AdapterKind::ElfServiceRuntime, - fixtures: &args.fixtures, - out_fixtures: &args.out_fixtures, - evidence_out: &args.evidence_out, - jobs: &jobs, - materialized: &materialized, - command_evidence: vec![CommandEvidence { - label: "elf_service_runtime".to_string(), - status: aggregate_status(&materialized), - command: "cargo run -p elf-eval --bin real_world_live_adapter -- elf".to_string(), - artifact: Some(args.evidence_out.display().to_string()), - reason: "ELF live adapter used ElfService, worker indexing, and search_raw." - .to_string(), - }], - metadata: None, - }) -} - -async fn materialize_elf_jobs(args: &ElfArgs, jobs: &[LoadedJob]) -> Result> { - let base_dsn = env::var("ELF_PG_DSN") - .map_err(|_| eyre::eyre!("ELF_PG_DSN must be set for ELF live real-world adapter."))?; - let qdrant_url = env::var("ELF_QDRANT_GRPC_URL") - .or_else(|_| env::var("ELF_QDRANT_URL")) - .map_err(|_| eyre::eyre!("ELF_QDRANT_GRPC_URL or ELF_QDRANT_URL must be set."))?; - let test_db = TestDatabase::new(&base_dsn).await?; - let run_suffix = crate::short_hash(format!("{}:{}", args.adapter_id, Uuid::new_v4()).as_str()); - let runtime = BaselineRuntime { - config_path: args.config.clone(), - dsn: test_db.dsn().to_string(), - qdrant_url, - collection: format!("elf_live_real_world_{run_suffix}"), - docs_collection: format!("elf_live_real_world_docs_{run_suffix}"), - }; - let service = crate::build_service(&runtime).await?; - let mut out = Vec::with_capacity(jobs.len()); - - for loaded in jobs { - out.push(materialize_elf_job(&runtime, &service, loaded, &args.adapter_id).await?); - } - - drop(service); - - test_db.cleanup().await?; - - Ok(out) -} - -async fn materialize_elf_job( - runtime: &BaselineRuntime, - service: &ElfService, - loaded: &LoadedJob, - adapter_id: &str, -) -> Result { - if let Some(job) = crate::declared_encoding_job(adapter_id, loaded) { - return Ok(job); - } - if let Some(job) = crate::not_encoded_job(adapter_id, loaded) { - return Ok(job); - } - - let corpus = crate::corpus_texts(loaded)?; - let stored_corpus = crate::elf_stored_corpus_texts(&corpus)?; - let project_id = crate::project_id_for_job(&loaded.job.job_id); - let ingested = - crate::ingest_elf_corpus(service, loaded, adapter_id, project_id.as_str(), &corpus).await?; - - crate::run_worker(runtime).await?; - - let (response, latency_ms) = search_elf_job(service, loaded, &project_id).await?; - let evidence_ids = crate::search_response_evidence_ids(&response); - let runtime_capture = crate::capture_runtime_evidence_from_search_items(&response.items); - let capture = - crate::capture_with_runtime_source_refs(ingested.capture.clone(), &runtime_capture); - let capture_failure = crate::validate_capture_runtime_evidence( - loaded.job.suite.as_str(), - &corpus, - &capture, - &runtime_capture, - ); - let (selected, temporal_reconciliation, trace_stages) = crate::elf_selected_evidence_text( - loaded, - &stored_corpus, - &evidence_ids, - &ingested, - &capture_failure, - ); - let replay_command = crate::elf_replay_command(response.trace_id, project_id.as_str()); - let (operator_debug, operator_debug_evidence) = crate::operator_debug_output( - AdapterKind::ElfServiceRuntime, - loaded, - Some(response.trace_id), - replay_command, - format!( - "/v2/admin/traces/{}/bundle?mode=full&stage_items_limit=128&candidates_limit=200", - response.trace_id - ), - ); - let optional = materialize_optional_elf_surfaces( - runtime, - service, - loaded, - &ingested, - project_id.as_str(), - response.trace_id, - adapter_id, - ) - .await?; - let suite_selection = - crate::suite_materialization_selection(SuiteMaterializationSelectionInput { - loaded, - ingested: &ingested, - capture_failure: &capture_failure, - selected, - trace_stages, - knowledge: &optional.knowledge, - consolidation: &optional.consolidation, - dreaming_readback: optional.dreaming_readback, - }); - - Ok(crate::materialized_job( - loaded, - adapter_id, - MaterializedJobInput { - content: suite_selection.selected.content, - evidence_ids: suite_selection.selected.evidence_ids, - pages: optional.pages, - latency_ms, - indexing_latency_ms: None, - returned_count: response.items.len(), - trace_id: Some(response.trace_id), - failure: optional.failure, - source_mappings: Vec::new(), - operator_debug, - operator_debug_evidence, - capture: crate::capture_for_job(loaded, capture), - capture_failure, - consolidation_response: optional.consolidation_response, - consolidation: optional.consolidation, - knowledge: optional.knowledge, - temporal_reconciliation, - dreaming_readback: suite_selection.dreaming_readback, - memory_summaries: suite_selection.memory_summaries, - proactive_briefs: suite_selection.proactive_briefs, - scheduled_tasks: suite_selection.scheduled_tasks, - trace_stages: suite_selection.trace_stages, - }, - )) -} - -async fn materialize_optional_elf_surfaces( - runtime: &BaselineRuntime, - service: &ElfService, - loaded: &LoadedJob, - ingested: &IngestedCorpus, - project_id: &str, - trace_id: Uuid, - adapter_id: &str, -) -> Result { - let (pages, knowledge, knowledge_failure) = - match crate::materialize_elf_knowledge(service, loaded, ingested, adapter_id).await { - Ok(output) => output, - Err(err) if loaded.job.suite == "knowledge_compilation" => - (Vec::new(), None, Some(format!("live_adapter.knowledge: {err}"))), - Err(_) => (Vec::new(), None, None), - }; - let (consolidation_response, consolidation, consolidation_failure) = - match crate::materialize_elf_consolidation(runtime, service, loaded, ingested, adapter_id) - .await - { - Ok(output) => output, - Err(err) if loaded.job.suite == "consolidation" => - (None, None, Some(format!("live_adapter.consolidation: {err}"))), - Err(_) => (None, None, None), - }; - let dreaming_readback = - crate::materialize_elf_dreaming_readback(service, loaded, project_id, trace_id, adapter_id) - .await?; - let dreaming_failure = dreaming_readback.as_ref().and_then(|output| { - if output.materialization.missing_source_refs.is_empty() { - None - } else { - Some(format!( - "live_adapter.dreaming_readback missing source refs: {}", - output.materialization.missing_source_refs.join(", ") - )) - } - }); - - Ok(OptionalElfMaterializations { - pages, - knowledge, - consolidation_response, - consolidation, - dreaming_readback, - failure: knowledge_failure.or(consolidation_failure).or(dreaming_failure), - }) -} - -async fn search_elf_job( - service: &ElfService, - loaded: &LoadedJob, - project_id: &str, -) -> Result<(SearchResponse, f64)> { - let started_at = Instant::now(); - let response = service - .search_raw(SearchRequest { - tenant_id: TENANT_ID.to_string(), - project_id: project_id.to_string(), - agent_id: AGENT_ID.to_string(), - token_id: None, - payload_level: PayloadLevel::L2, - read_profile: "private_only".to_string(), - query: loaded.job.prompt.content.clone(), - top_k: Some(5), - candidate_k: Some(20), - filter: None, - record_hits: Some(false), - ranking: None, - }) - .await - .map_err(|err| eyre::eyre!("ELF search_raw failed for {}: {err}", loaded.job.job_id))?; - - Ok((response, started_at.elapsed().as_secs_f64() * 1_000.0)) -} +pub(super) use self::jobs::run_elf; diff --git a/apps/elf-eval/src/bin/real_world_live_adapter/elf_runtime/jobs.rs b/apps/elf-eval/src/bin/real_world_live_adapter/elf_runtime/jobs.rs new file mode 100644 index 00000000..27872281 --- /dev/null +++ b/apps/elf-eval/src/bin/real_world_live_adapter/elf_runtime/jobs.rs @@ -0,0 +1,167 @@ +use crate::{ + AdapterKind, BaselineRuntime, CommandEvidence, ElfArgs, ElfService, LoadedJob, MaterializedJob, + MaterializedJobInput, MaterializedOutput, Result, SuiteMaterializationSelectionInput, + TestDatabase, Uuid, aggregate_status, + elf_runtime::{search, surfaces}, + env, eyre, +}; + +pub(crate) async fn run_elf(args: ElfArgs) -> Result<()> { + let jobs = crate::load_jobs(&args.fixtures)?; + let result = materialize_elf_jobs(&args, &jobs).await; + let materialized = match result { + Ok(jobs) => jobs, + Err(err) => + crate::failure_jobs(&args.adapter_id, &jobs, "elf_service_runtime", err.to_string()), + }; + + crate::write_materialized_output(MaterializedOutput { + adapter_id: &args.adapter_id, + adapter_kind: AdapterKind::ElfServiceRuntime, + fixtures: &args.fixtures, + out_fixtures: &args.out_fixtures, + evidence_out: &args.evidence_out, + jobs: &jobs, + materialized: &materialized, + command_evidence: vec![CommandEvidence { + label: "elf_service_runtime".to_string(), + status: aggregate_status(&materialized), + command: "cargo run -p elf-eval --bin real_world_live_adapter -- elf".to_string(), + artifact: Some(args.evidence_out.display().to_string()), + reason: "ELF live adapter used ElfService, worker indexing, and search_raw." + .to_string(), + }], + metadata: None, + }) +} + +async fn materialize_elf_jobs(args: &ElfArgs, jobs: &[LoadedJob]) -> Result> { + let base_dsn = env::var("ELF_PG_DSN") + .map_err(|_| eyre::eyre!("ELF_PG_DSN must be set for ELF live real-world adapter."))?; + let qdrant_url = env::var("ELF_QDRANT_GRPC_URL") + .or_else(|_| env::var("ELF_QDRANT_URL")) + .map_err(|_| eyre::eyre!("ELF_QDRANT_GRPC_URL or ELF_QDRANT_URL must be set."))?; + let test_db = TestDatabase::new(&base_dsn).await?; + let run_suffix = crate::short_hash(format!("{}:{}", args.adapter_id, Uuid::new_v4()).as_str()); + let runtime = BaselineRuntime { + config_path: args.config.clone(), + dsn: test_db.dsn().to_string(), + qdrant_url, + collection: format!("elf_live_real_world_{run_suffix}"), + docs_collection: format!("elf_live_real_world_docs_{run_suffix}"), + }; + let service = crate::build_service(&runtime).await?; + let mut out = Vec::with_capacity(jobs.len()); + + for loaded in jobs { + out.push(materialize_elf_job(&runtime, &service, loaded, &args.adapter_id).await?); + } + + drop(service); + + test_db.cleanup().await?; + + Ok(out) +} + +async fn materialize_elf_job( + runtime: &BaselineRuntime, + service: &ElfService, + loaded: &LoadedJob, + adapter_id: &str, +) -> Result { + if let Some(job) = crate::declared_encoding_job(adapter_id, loaded) { + return Ok(job); + } + if let Some(job) = crate::not_encoded_job(adapter_id, loaded) { + return Ok(job); + } + + let corpus = crate::corpus_texts(loaded)?; + let stored_corpus = crate::elf_stored_corpus_texts(&corpus)?; + let project_id = crate::project_id_for_job(&loaded.job.job_id); + let ingested = + crate::ingest_elf_corpus(service, loaded, adapter_id, project_id.as_str(), &corpus).await?; + + crate::run_worker(runtime).await?; + + let (response, latency_ms) = search::search_elf_job(service, loaded, &project_id).await?; + let evidence_ids = crate::search_response_evidence_ids(&response); + let runtime_capture = crate::capture_runtime_evidence_from_search_items(&response.items); + let capture = + crate::capture_with_runtime_source_refs(ingested.capture.clone(), &runtime_capture); + let capture_failure = crate::validate_capture_runtime_evidence( + loaded.job.suite.as_str(), + &corpus, + &capture, + &runtime_capture, + ); + let (selected, temporal_reconciliation, trace_stages) = crate::elf_selected_evidence_text( + loaded, + &stored_corpus, + &evidence_ids, + &ingested, + &capture_failure, + ); + let replay_command = crate::elf_replay_command(response.trace_id, project_id.as_str()); + let (operator_debug, operator_debug_evidence) = crate::operator_debug_output( + AdapterKind::ElfServiceRuntime, + loaded, + Some(response.trace_id), + replay_command, + format!( + "/v2/admin/traces/{}/bundle?mode=full&stage_items_limit=128&candidates_limit=200", + response.trace_id + ), + ); + let optional = surfaces::materialize_optional_elf_surfaces( + runtime, + service, + loaded, + &ingested, + project_id.as_str(), + response.trace_id, + adapter_id, + ) + .await?; + let suite_selection = + crate::suite_materialization_selection(SuiteMaterializationSelectionInput { + loaded, + ingested: &ingested, + capture_failure: &capture_failure, + selected, + trace_stages, + knowledge: &optional.knowledge, + consolidation: &optional.consolidation, + dreaming_readback: optional.dreaming_readback, + }); + + Ok(crate::materialized_job( + loaded, + adapter_id, + MaterializedJobInput { + content: suite_selection.selected.content, + evidence_ids: suite_selection.selected.evidence_ids, + pages: optional.pages, + latency_ms, + indexing_latency_ms: None, + returned_count: response.items.len(), + trace_id: Some(response.trace_id), + failure: optional.failure, + source_mappings: Vec::new(), + operator_debug, + operator_debug_evidence, + capture: crate::capture_for_job(loaded, capture), + capture_failure, + consolidation_response: optional.consolidation_response, + consolidation: optional.consolidation, + knowledge: optional.knowledge, + temporal_reconciliation, + dreaming_readback: suite_selection.dreaming_readback, + memory_summaries: suite_selection.memory_summaries, + proactive_briefs: suite_selection.proactive_briefs, + scheduled_tasks: suite_selection.scheduled_tasks, + trace_stages: suite_selection.trace_stages, + }, + )) +} diff --git a/apps/elf-eval/src/bin/real_world_live_adapter/elf_runtime/search.rs b/apps/elf-eval/src/bin/real_world_live_adapter/elf_runtime/search.rs new file mode 100644 index 00000000..ac62a2cb --- /dev/null +++ b/apps/elf-eval/src/bin/real_world_live_adapter/elf_runtime/search.rs @@ -0,0 +1,31 @@ +use crate::{ + AGENT_ID, ElfService, Instant, LoadedJob, PayloadLevel, Result, SearchRequest, SearchResponse, + TENANT_ID, eyre, +}; + +pub(super) async fn search_elf_job( + service: &ElfService, + loaded: &LoadedJob, + project_id: &str, +) -> Result<(SearchResponse, f64)> { + let started_at = Instant::now(); + let response = service + .search_raw(SearchRequest { + tenant_id: TENANT_ID.to_string(), + project_id: project_id.to_string(), + agent_id: AGENT_ID.to_string(), + token_id: None, + payload_level: PayloadLevel::L2, + read_profile: "private_only".to_string(), + query: loaded.job.prompt.content.clone(), + top_k: Some(5), + candidate_k: Some(20), + filter: None, + record_hits: Some(false), + ranking: None, + }) + .await + .map_err(|err| eyre::eyre!("ELF search_raw failed for {}: {err}", loaded.job.job_id))?; + + Ok((response, started_at.elapsed().as_secs_f64() * 1_000.0)) +} diff --git a/apps/elf-eval/src/bin/real_world_live_adapter/elf_runtime/surfaces.rs b/apps/elf-eval/src/bin/real_world_live_adapter/elf_runtime/surfaces.rs new file mode 100644 index 00000000..c07e5475 --- /dev/null +++ b/apps/elf-eval/src/bin/real_world_live_adapter/elf_runtime/surfaces.rs @@ -0,0 +1,62 @@ +use crate::{ + BaselineRuntime, ConsolidationMaterializationEvidence, DreamingReadbackOutput, ElfService, + IngestedCorpus, KnowledgeMaterializationEvidence, LoadedJob, Result, Uuid, Value, +}; + +pub(super) struct OptionalElfMaterializations { + pub(super) pages: Vec, + pub(super) knowledge: Option, + pub(super) consolidation_response: Option, + pub(super) consolidation: Option, + pub(super) dreaming_readback: Option, + pub(super) failure: Option, +} + +pub(super) async fn materialize_optional_elf_surfaces( + runtime: &BaselineRuntime, + service: &ElfService, + loaded: &LoadedJob, + ingested: &IngestedCorpus, + project_id: &str, + trace_id: Uuid, + adapter_id: &str, +) -> Result { + let (pages, knowledge, knowledge_failure) = + match crate::materialize_elf_knowledge(service, loaded, ingested, adapter_id).await { + Ok(output) => output, + Err(err) if loaded.job.suite == "knowledge_compilation" => + (Vec::new(), None, Some(format!("live_adapter.knowledge: {err}"))), + Err(_) => (Vec::new(), None, None), + }; + let (consolidation_response, consolidation, consolidation_failure) = + match crate::materialize_elf_consolidation(runtime, service, loaded, ingested, adapter_id) + .await + { + Ok(output) => output, + Err(err) if loaded.job.suite == "consolidation" => + (None, None, Some(format!("live_adapter.consolidation: {err}"))), + Err(_) => (None, None, None), + }; + let dreaming_readback = + crate::materialize_elf_dreaming_readback(service, loaded, project_id, trace_id, adapter_id) + .await?; + let dreaming_failure = dreaming_readback.as_ref().and_then(|output| { + if output.materialization.missing_source_refs.is_empty() { + None + } else { + Some(format!( + "live_adapter.dreaming_readback missing source refs: {}", + output.materialization.missing_source_refs.join(", ") + )) + } + }); + + Ok(OptionalElfMaterializations { + pages, + knowledge, + consolidation_response, + consolidation, + dreaming_readback, + failure: knowledge_failure.or(consolidation_failure).or(dreaming_failure), + }) +} diff --git a/apps/elf-eval/src/bin/real_world_live_adapter/knowledge_adapter.rs b/apps/elf-eval/src/bin/real_world_live_adapter/knowledge_adapter.rs index 296ae99b..02084446 100644 --- a/apps/elf-eval/src/bin/real_world_live_adapter/knowledge_adapter.rs +++ b/apps/elf-eval/src/bin/real_world_live_adapter/knowledge_adapter.rs @@ -1,219 +1,8 @@ -use crate::{ - HashMap, IngestedCorpus, KnowledgeMaterializationEvidence, KnowledgePageLintResponse, - KnowledgePageResponse, LoadedJob, Result, Uuid, serde_json, -}; - -pub(super) fn knowledge_page_artifact( - loaded: &LoadedJob, - ingested: &IngestedCorpus, - first: &KnowledgePageResponse, - second: &KnowledgePageResponse, - lint: &KnowledgePageLintResponse, -) -> Result { - let reverse = note_id_to_evidence_id(ingested); - let mut sections = second - .sections - .iter() - .map(|section| { - let evidence_ids = section - .source_backlinks - .iter() - .filter_map(|source| reverse.get(&source.source_id).cloned()) - .collect::>(); - - serde_json::json!({ - "section_id": section.section_key.clone(), - "heading": section.heading.clone(), - "role": section.role.clone(), - "content": section.content.clone(), - "evidence_ids": evidence_ids, - "timeline_event_ids": [] - }) - }) - .collect::>(); - - sections.extend(unsupported_sections_from_fixture(loaded)); - - Ok(serde_json::json!({ - "page_id": second.page.page_id.to_string(), - "page_type": second.page.page_kind.clone(), - "title": second.page.title.clone(), - "sections": sections, - "backlinks": source_backlinks(ingested), - "lint_findings": lint_findings_for_page(loaded, ingested, lint), - "page_version_diff": second.page.previous_version_diff.clone(), - "rebuild": { - "first_hash": first.page.content_hash.clone(), - "second_hash": second.page.content_hash.clone(), - "deterministic": first.page.content_hash == second.page.content_hash, - "allowed_variance": [] - } - })) -} - -pub(super) fn knowledge_materialization_evidence( - page: &KnowledgePageResponse, - lint: &KnowledgePageLintResponse, - search_result_count: usize, -) -> KnowledgeMaterializationEvidence { - let unsupported_claim_count = - lint.findings.iter().filter(|finding| finding.finding_type == "unsupported_claim").count() - + page.sections.iter().filter(|section| section.unsupported_reason.is_some()).count(); - - KnowledgeMaterializationEvidence { - page_ids: vec![page.page.page_id], - search_result_count, - lint_finding_count: lint.findings.len(), - stale_source_finding_count: lint - .findings - .iter() - .filter(|finding| finding.finding_type == "stale_source_ref") - .count(), - unsupported_claim_count, - citation_count: page.sections.iter().map(|section| section.citation_count).sum(), - source_ref_count: page.source_refs.len(), - version_diff_available: page - .page - .previous_version_diff - .as_ref() - .and_then(|diff| diff.get("available")) - .and_then(serde_json::Value::as_bool) - .unwrap_or(false), - } -} - -pub(super) fn stale_trap_evidence_ids(loaded: &LoadedJob) -> Vec { - loaded - .value - .get("negative_traps") - .and_then(serde_json::Value::as_array) - .into_iter() - .flatten() - .filter(|trap| { - trap.get("type").and_then(serde_json::Value::as_str) == Some("stale_fact") - && trap.get("failure_if_used").and_then(serde_json::Value::as_bool).unwrap_or(false) - }) - .flat_map(|trap| { - trap.get("evidence_ids") - .and_then(serde_json::Value::as_array) - .into_iter() - .flatten() - .filter_map(serde_json::Value::as_str) - .map(ToString::to_string) - .collect::>() - }) - .collect() -} - -fn note_id_to_evidence_id(ingested: &IngestedCorpus) -> HashMap { - let mut out = HashMap::new(); - - for (evidence_id, note_ids) in &ingested.note_ids_by_evidence { - for note_id in note_ids { - out.insert(*note_id, evidence_id.clone()); - } - } - - out -} - -fn source_backlinks(ingested: &IngestedCorpus) -> Vec { - let mut backlinks = ingested - .note_ids_by_evidence - .keys() - .map(|evidence_id| format!("source:{evidence_id}")) - .collect::>(); +mod artifact; +mod evidence; +mod traps; - backlinks.sort(); - - backlinks -} - -fn lint_findings_for_page( - loaded: &LoadedJob, - ingested: &IngestedCorpus, - lint: &KnowledgePageLintResponse, -) -> Vec { - let reverse = note_id_to_evidence_id(ingested); - - lint.findings - .iter() - .map(|finding| { - let evidence_ids = finding - .source_id - .and_then(|source_id| reverse.get(&source_id).cloned()) - .into_iter() - .collect::>(); - let trap_id = evidence_ids - .first() - .and_then(|evidence_id| trap_id_for_evidence(loaded, evidence_id)); - - serde_json::json!({ - "finding_id": finding.finding_id.to_string(), - "finding_type": finding.finding_type.clone(), - "severity": finding.severity.clone(), - "text": finding.message.clone(), - "evidence_ids": evidence_ids, - "trap_id": trap_id - }) - }) - .collect() -} - -fn unsupported_sections_from_fixture(loaded: &LoadedJob) -> Vec { - let Some(pages) = loaded - .value - .pointer("/corpus/adapter_response/answer/pages") - .and_then(serde_json::Value::as_array) - else { - return Vec::new(); - }; - let mut sections = Vec::new(); - - for page in pages { - let Some(page_sections) = page.get("sections").and_then(serde_json::Value::as_array) else { - continue; - }; - - for section in page_sections { - let Some(reason) = - section.get("unsupported_reason").and_then(serde_json::Value::as_str) - else { - continue; - }; - - sections.push(serde_json::json!({ - "section_id": section - .get("section_id") - .and_then(serde_json::Value::as_str) - .unwrap_or("unsupported-summary"), - "heading": section - .get("heading") - .and_then(serde_json::Value::as_str) - .unwrap_or("Unsupported Summary"), - "role": section.get("role").and_then(serde_json::Value::as_str).unwrap_or("summary"), - "content": section.get("content").and_then(serde_json::Value::as_str).unwrap_or(reason), - "evidence_ids": [], - "timeline_event_ids": [], - "unsupported_reason": reason - })); - } - } - - sections -} - -fn trap_id_for_evidence(loaded: &LoadedJob, evidence_id: &str) -> Option { - loaded - .value - .get("negative_traps") - .and_then(serde_json::Value::as_array)? - .iter() - .find(|trap| { - trap.get("evidence_ids") - .and_then(serde_json::Value::as_array) - .is_some_and(|ids| ids.iter().any(|id| id.as_str() == Some(evidence_id))) - }) - .and_then(|trap| trap.get("trap_id").and_then(serde_json::Value::as_str)) - .map(ToString::to_string) -} +pub(super) use self::{ + artifact::knowledge_page_artifact, evidence::knowledge_materialization_evidence, + traps::stale_trap_evidence_ids, +}; diff --git a/apps/elf-eval/src/bin/real_world_live_adapter/knowledge_adapter/artifact.rs b/apps/elf-eval/src/bin/real_world_live_adapter/knowledge_adapter/artifact.rs new file mode 100644 index 00000000..97b42539 --- /dev/null +++ b/apps/elf-eval/src/bin/real_world_live_adapter/knowledge_adapter/artifact.rs @@ -0,0 +1,150 @@ +use crate::{ + HashMap, IngestedCorpus, KnowledgePageLintResponse, KnowledgePageResponse, LoadedJob, Result, + Uuid, knowledge_adapter::traps, serde_json, +}; + +pub(crate) fn knowledge_page_artifact( + loaded: &LoadedJob, + ingested: &IngestedCorpus, + first: &KnowledgePageResponse, + second: &KnowledgePageResponse, + lint: &KnowledgePageLintResponse, +) -> Result { + let reverse = note_id_to_evidence_id(ingested); + let mut sections = second + .sections + .iter() + .map(|section| { + let evidence_ids = section + .source_backlinks + .iter() + .filter_map(|source| reverse.get(&source.source_id).cloned()) + .collect::>(); + + serde_json::json!({ + "section_id": section.section_key.clone(), + "heading": section.heading.clone(), + "role": section.role.clone(), + "content": section.content.clone(), + "evidence_ids": evidence_ids, + "timeline_event_ids": [] + }) + }) + .collect::>(); + + sections.extend(unsupported_sections_from_fixture(loaded)); + + Ok(serde_json::json!({ + "page_id": second.page.page_id.to_string(), + "page_type": second.page.page_kind.clone(), + "title": second.page.title.clone(), + "sections": sections, + "backlinks": source_backlinks(ingested), + "lint_findings": lint_findings_for_page(loaded, ingested, lint), + "page_version_diff": second.page.previous_version_diff.clone(), + "rebuild": { + "first_hash": first.page.content_hash.clone(), + "second_hash": second.page.content_hash.clone(), + "deterministic": first.page.content_hash == second.page.content_hash, + "allowed_variance": [] + } + })) +} + +fn note_id_to_evidence_id(ingested: &IngestedCorpus) -> HashMap { + let mut out = HashMap::new(); + + for (evidence_id, note_ids) in &ingested.note_ids_by_evidence { + for note_id in note_ids { + out.insert(*note_id, evidence_id.clone()); + } + } + + out +} + +fn source_backlinks(ingested: &IngestedCorpus) -> Vec { + let mut backlinks = ingested + .note_ids_by_evidence + .keys() + .map(|evidence_id| format!("source:{evidence_id}")) + .collect::>(); + + backlinks.sort(); + + backlinks +} + +fn lint_findings_for_page( + loaded: &LoadedJob, + ingested: &IngestedCorpus, + lint: &KnowledgePageLintResponse, +) -> Vec { + let reverse = note_id_to_evidence_id(ingested); + + lint.findings + .iter() + .map(|finding| { + let evidence_ids = finding + .source_id + .and_then(|source_id| reverse.get(&source_id).cloned()) + .into_iter() + .collect::>(); + let trap_id = evidence_ids + .first() + .and_then(|evidence_id| traps::trap_id_for_evidence(loaded, evidence_id)); + + serde_json::json!({ + "finding_id": finding.finding_id.to_string(), + "finding_type": finding.finding_type.clone(), + "severity": finding.severity.clone(), + "text": finding.message.clone(), + "evidence_ids": evidence_ids, + "trap_id": trap_id + }) + }) + .collect() +} + +fn unsupported_sections_from_fixture(loaded: &LoadedJob) -> Vec { + let Some(pages) = loaded + .value + .pointer("/corpus/adapter_response/answer/pages") + .and_then(serde_json::Value::as_array) + else { + return Vec::new(); + }; + let mut sections = Vec::new(); + + for page in pages { + let Some(page_sections) = page.get("sections").and_then(serde_json::Value::as_array) else { + continue; + }; + + for section in page_sections { + let Some(reason) = + section.get("unsupported_reason").and_then(serde_json::Value::as_str) + else { + continue; + }; + + sections.push(serde_json::json!({ + "section_id": section + .get("section_id") + .and_then(serde_json::Value::as_str) + .unwrap_or("unsupported-summary"), + "heading": section + .get("heading") + .and_then(serde_json::Value::as_str) + .unwrap_or("Unsupported Summary"), + "role": section.get("role").and_then(serde_json::Value::as_str).unwrap_or("summary"), + "content": section.get("content").and_then(serde_json::Value::as_str).unwrap_or(reason), + "evidence_ids": [], + "timeline_event_ids": [], + "unsupported_reason": reason + })); + } + } + + sections +} diff --git a/apps/elf-eval/src/bin/real_world_live_adapter/knowledge_adapter/evidence.rs b/apps/elf-eval/src/bin/real_world_live_adapter/knowledge_adapter/evidence.rs new file mode 100644 index 00000000..c00efd6a --- /dev/null +++ b/apps/elf-eval/src/bin/real_world_live_adapter/knowledge_adapter/evidence.rs @@ -0,0 +1,34 @@ +use crate::{ + KnowledgeMaterializationEvidence, KnowledgePageLintResponse, KnowledgePageResponse, Value, +}; + +pub(crate) fn knowledge_materialization_evidence( + page: &KnowledgePageResponse, + lint: &KnowledgePageLintResponse, + search_result_count: usize, +) -> KnowledgeMaterializationEvidence { + let unsupported_claim_count = + lint.findings.iter().filter(|finding| finding.finding_type == "unsupported_claim").count() + + page.sections.iter().filter(|section| section.unsupported_reason.is_some()).count(); + + KnowledgeMaterializationEvidence { + page_ids: vec![page.page.page_id], + search_result_count, + lint_finding_count: lint.findings.len(), + stale_source_finding_count: lint + .findings + .iter() + .filter(|finding| finding.finding_type == "stale_source_ref") + .count(), + unsupported_claim_count, + citation_count: page.sections.iter().map(|section| section.citation_count).sum(), + source_ref_count: page.source_refs.len(), + version_diff_available: page + .page + .previous_version_diff + .as_ref() + .and_then(|diff| diff.get("available")) + .and_then(Value::as_bool) + .unwrap_or(false), + } +} diff --git a/apps/elf-eval/src/bin/real_world_live_adapter/knowledge_adapter/traps.rs b/apps/elf-eval/src/bin/real_world_live_adapter/knowledge_adapter/traps.rs new file mode 100644 index 00000000..cdf98abc --- /dev/null +++ b/apps/elf-eval/src/bin/real_world_live_adapter/knowledge_adapter/traps.rs @@ -0,0 +1,39 @@ +use crate::{LoadedJob, Value}; + +pub(crate) fn stale_trap_evidence_ids(loaded: &LoadedJob) -> Vec { + loaded + .value + .get("negative_traps") + .and_then(Value::as_array) + .into_iter() + .flatten() + .filter(|trap| { + trap.get("type").and_then(Value::as_str) == Some("stale_fact") + && trap.get("failure_if_used").and_then(Value::as_bool).unwrap_or(false) + }) + .flat_map(|trap| { + trap.get("evidence_ids") + .and_then(Value::as_array) + .into_iter() + .flatten() + .filter_map(Value::as_str) + .map(ToString::to_string) + .collect::>() + }) + .collect() +} + +pub(super) fn trap_id_for_evidence(loaded: &LoadedJob, evidence_id: &str) -> Option { + loaded + .value + .get("negative_traps") + .and_then(Value::as_array)? + .iter() + .find(|trap| { + trap.get("evidence_ids") + .and_then(Value::as_array) + .is_some_and(|ids| ids.iter().any(|id| id.as_str() == Some(evidence_id))) + }) + .and_then(|trap| trap.get("trap_id").and_then(Value::as_str)) + .map(ToString::to_string) +} diff --git a/apps/elf-eval/src/bin/real_world_live_adapter/qmd.rs b/apps/elf-eval/src/bin/real_world_live_adapter/qmd.rs index 5bb690db..41125724 100644 --- a/apps/elf-eval/src/bin/real_world_live_adapter/qmd.rs +++ b/apps/elf-eval/src/bin/real_world_live_adapter/qmd.rs @@ -1,231 +1,6 @@ -use crate::{ - AdapterKind, Command, CommandEvidence, Instant, LoadedJob, MaterializedJob, - MaterializedJobInput, MaterializedOutput, OperatorDebugMaterializationEvidence, Path, QmdArgs, - Result, SelectedEvidenceText, Value, aggregate_status, eyre, fs, serde_json, -}; +mod checkout; +mod job; +mod response; +mod runtime; -pub(super) fn run_qmd(args: QmdArgs) -> Result<()> { - let jobs = crate::load_jobs(&args.fixtures)?; - let result = materialize_qmd_jobs(&args, &jobs); - let materialized = match result { - Ok(jobs) => jobs, - Err(err) => - crate::failure_jobs(&args.adapter_id, &jobs, "qmd_cli_runtime", err.to_string()), - }; - - crate::write_materialized_output(MaterializedOutput { - adapter_id: &args.adapter_id, - adapter_kind: AdapterKind::QmdCliRuntime, - fixtures: &args.fixtures, - out_fixtures: &args.out_fixtures, - evidence_out: &args.evidence_out, - jobs: &jobs, - materialized: &materialized, - command_evidence: vec![CommandEvidence { - label: "qmd_cli_runtime".to_string(), - status: aggregate_status(&materialized), - command: "cargo run -p elf-eval --bin real_world_live_adapter -- qmd".to_string(), - artifact: Some(args.evidence_out.display().to_string()), - reason: "qmd live adapter used collection add, update, embed, and query --json." - .to_string(), - }], - metadata: None, - }) -} - -fn materialize_qmd_jobs(args: &QmdArgs, jobs: &[LoadedJob]) -> Result> { - fs::create_dir_all(&args.work_dir)?; - - let log_path = args.work_dir.join("qmd-live-real-world.log"); - - ensure_qmd_checkout(args, &log_path)?; - - let mut out = Vec::with_capacity(jobs.len()); - - for loaded in jobs { - out.push(materialize_qmd_job(args, loaded, &log_path)?); - } - - Ok(out) -} - -fn ensure_qmd_checkout(args: &QmdArgs, log_path: &Path) -> Result<()> { - if !args.qmd_dir.exists() { - if let Some(parent) = args.qmd_dir.parent() { - fs::create_dir_all(parent)?; - } - - crate::run_logged_command( - "qmd clone", - Command::new("git") - .arg("clone") - .arg("--depth") - .arg("1") - .arg(&args.qmd_repo_url) - .arg(&args.qmd_dir), - log_path, - )?; - } - - crate::run_logged_shell( - "qmd install", - &args.qmd_dir, - "(npm ci || npm install --no-audit --no-fund) && npm run build --if-present", - log_path, - ) -} - -fn materialize_qmd_job( - args: &QmdArgs, - loaded: &LoadedJob, - log_path: &Path, -) -> Result { - if let Some(job) = crate::declared_encoding_job(&args.adapter_id, loaded) { - return Ok(job); - } - if let Some(job) = crate::not_encoded_job(&args.adapter_id, loaded) { - return Ok(job); - } - - let corpus = crate::corpus_texts(loaded)?; - let job_slug = crate::slug(&loaded.job.job_id); - let corpus_dir = args.work_dir.join("corpus").join(&job_slug); - let home_dir = args.work_dir.join("home").join(&job_slug); - let collection = format!("elfrw-{job_slug}"); - - fs::create_dir_all(&corpus_dir)?; - fs::create_dir_all(&home_dir)?; - - for existing in crate::read_dir_paths(&corpus_dir)? { - if existing.is_file() { - fs::remove_file(existing)?; - } - } - for item in &corpus { - let path = corpus_dir.join(format!("{}.md", crate::slug(&item.evidence_id))); - - fs::write(path, format!("# {}\n\n{}\n", item.evidence_id, item.text))?; - } - - crate::run_qmd_command( - "qmd collection add", - args, - &home_dir, - &[ - "collection", - "add", - corpus_dir - .to_str() - .ok_or_else(|| eyre::eyre!("qmd corpus path is not valid UTF-8."))?, - "--name", - collection.as_str(), - ], - log_path, - )?; - crate::run_qmd_command("qmd update", args, &home_dir, &["update"], log_path)?; - crate::run_qmd_command( - "qmd embed", - args, - &home_dir, - &["embed", "-f", "-c", collection.as_str()], - log_path, - )?; - - let started_at = Instant::now(); - let query = format!("lex: {}\nvec: {}", loaded.job.prompt.content, loaded.job.prompt.content); - let stdout = crate::run_qmd_command( - "qmd query", - args, - &home_dir, - &[ - "query", - query.as_str(), - "-c", - collection.as_str(), - "--json", - "--no-rerank", - "--min-score", - "0", - "-n", - "5", - ], - log_path, - )?; - let latency_ms = started_at.elapsed().as_secs_f64() * 1_000.0; - let results = serde_json::from_str::(&stdout).map_err(|err| { - eyre::eyre!("qmd query did not return JSON for {}: {err}", loaded.job.job_id) - })?; - let entries = results.as_array().cloned().unwrap_or_default(); - let mut evidence_ids = Vec::new(); - - for entry in &entries { - let entry_text = serde_json::to_string(entry)?; - - for item in &corpus { - if entry_text.contains(format!("{}.md", crate::slug(&item.evidence_id)).as_str()) - || entry_text.contains(item.evidence_id.as_str()) - { - crate::push_unique(&mut evidence_ids, item.evidence_id.clone()); - } - } - } - - let selected = crate::selected_required_corpus_texts(loaded, &corpus, &evidence_ids); - let replay_command = crate::qmd_replay_command(&loaded.job.prompt.content, collection.as_str()); - let (operator_debug, operator_debug_evidence) = crate::operator_debug_output( - AdapterKind::QmdCliRuntime, - loaded, - None, - replay_command, - log_path.display().to_string(), - ); - - Ok(qmd_materialized_job( - loaded, - &args.adapter_id, - selected, - latency_ms, - entries.len(), - operator_debug, - operator_debug_evidence, - )) -} - -fn qmd_materialized_job( - loaded: &LoadedJob, - adapter_id: &str, - selected: SelectedEvidenceText, - latency_ms: f64, - returned_count: usize, - operator_debug: Option, - operator_debug_evidence: Option, -) -> MaterializedJob { - crate::materialized_job( - loaded, - adapter_id, - MaterializedJobInput { - content: selected.content, - evidence_ids: selected.evidence_ids, - pages: Vec::new(), - latency_ms, - indexing_latency_ms: None, - returned_count, - trace_id: None, - failure: None, - source_mappings: Vec::new(), - operator_debug, - operator_debug_evidence, - capture: None, - capture_failure: None, - consolidation_response: None, - consolidation: None, - knowledge: None, - temporal_reconciliation: None, - dreaming_readback: None, - memory_summaries: Vec::new(), - proactive_briefs: Vec::new(), - scheduled_tasks: Vec::new(), - trace_stages: None, - }, - ) -} +pub(super) use self::runtime::run_qmd; diff --git a/apps/elf-eval/src/bin/real_world_live_adapter/qmd/checkout.rs b/apps/elf-eval/src/bin/real_world_live_adapter/qmd/checkout.rs new file mode 100644 index 00000000..e22f04f4 --- /dev/null +++ b/apps/elf-eval/src/bin/real_world_live_adapter/qmd/checkout.rs @@ -0,0 +1,27 @@ +use crate::{Command, Path, QmdArgs, Result, fs}; + +pub(super) fn ensure_qmd_checkout(args: &QmdArgs, log_path: &Path) -> Result<()> { + if !args.qmd_dir.exists() { + if let Some(parent) = args.qmd_dir.parent() { + fs::create_dir_all(parent)?; + } + + crate::run_logged_command( + "qmd clone", + Command::new("git") + .arg("clone") + .arg("--depth") + .arg("1") + .arg(&args.qmd_repo_url) + .arg(&args.qmd_dir), + log_path, + )?; + } + + crate::run_logged_shell( + "qmd install", + &args.qmd_dir, + "(npm ci || npm install --no-audit --no-fund) && npm run build --if-present", + log_path, + ) +} diff --git a/apps/elf-eval/src/bin/real_world_live_adapter/qmd/job.rs b/apps/elf-eval/src/bin/real_world_live_adapter/qmd/job.rs new file mode 100644 index 00000000..783f95bc --- /dev/null +++ b/apps/elf-eval/src/bin/real_world_live_adapter/qmd/job.rs @@ -0,0 +1,103 @@ +use crate::{ + AdapterKind, Instant, LoadedJob, MaterializedJob, Path, QmdArgs, Result, eyre, fs, + qmd::response, +}; + +pub(super) fn materialize_qmd_job( + args: &QmdArgs, + loaded: &LoadedJob, + log_path: &Path, +) -> Result { + if let Some(job) = crate::declared_encoding_job(&args.adapter_id, loaded) { + return Ok(job); + } + if let Some(job) = crate::not_encoded_job(&args.adapter_id, loaded) { + return Ok(job); + } + + let corpus = crate::corpus_texts(loaded)?; + let job_slug = crate::slug(&loaded.job.job_id); + let corpus_dir = args.work_dir.join("corpus").join(&job_slug); + let home_dir = args.work_dir.join("home").join(&job_slug); + let collection = format!("elfrw-{job_slug}"); + + fs::create_dir_all(&corpus_dir)?; + fs::create_dir_all(&home_dir)?; + + for existing in crate::read_dir_paths(&corpus_dir)? { + if existing.is_file() { + fs::remove_file(existing)?; + } + } + for item in &corpus { + let path = corpus_dir.join(format!("{}.md", crate::slug(&item.evidence_id))); + + fs::write(path, format!("# {}\n\n{}\n", item.evidence_id, item.text))?; + } + + crate::run_qmd_command( + "qmd collection add", + args, + &home_dir, + &[ + "collection", + "add", + corpus_dir + .to_str() + .ok_or_else(|| eyre::eyre!("qmd corpus path is not valid UTF-8."))?, + "--name", + collection.as_str(), + ], + log_path, + )?; + crate::run_qmd_command("qmd update", args, &home_dir, &["update"], log_path)?; + crate::run_qmd_command( + "qmd embed", + args, + &home_dir, + &["embed", "-f", "-c", collection.as_str()], + log_path, + )?; + + let started_at = Instant::now(); + let query = format!("lex: {}\nvec: {}", loaded.job.prompt.content, loaded.job.prompt.content); + let stdout = crate::run_qmd_command( + "qmd query", + args, + &home_dir, + &[ + "query", + query.as_str(), + "-c", + collection.as_str(), + "--json", + "--no-rerank", + "--min-score", + "0", + "-n", + "5", + ], + log_path, + )?; + let latency_ms = started_at.elapsed().as_secs_f64() * 1_000.0; + let (entries, evidence_ids) = response::qmd_query_entries(loaded, &corpus, &stdout)?; + let selected = crate::selected_required_corpus_texts(loaded, &corpus, &evidence_ids); + let replay_command = crate::qmd_replay_command(&loaded.job.prompt.content, collection.as_str()); + let (operator_debug, operator_debug_evidence) = crate::operator_debug_output( + AdapterKind::QmdCliRuntime, + loaded, + None, + replay_command, + log_path.display().to_string(), + ); + + Ok(response::qmd_materialized_job( + loaded, + &args.adapter_id, + selected, + latency_ms, + entries.len(), + operator_debug, + operator_debug_evidence, + )) +} diff --git a/apps/elf-eval/src/bin/real_world_live_adapter/qmd/response.rs b/apps/elf-eval/src/bin/real_world_live_adapter/qmd/response.rs new file mode 100644 index 00000000..c40659f0 --- /dev/null +++ b/apps/elf-eval/src/bin/real_world_live_adapter/qmd/response.rs @@ -0,0 +1,69 @@ +use crate::{ + CorpusText, LoadedJob, MaterializedJob, MaterializedJobInput, + OperatorDebugMaterializationEvidence, Result, SelectedEvidenceText, Value, eyre, serde_json, +}; + +pub(super) fn qmd_query_entries( + loaded: &LoadedJob, + corpus: &[CorpusText], + stdout: &str, +) -> Result<(Vec, Vec)> { + let results = serde_json::from_str::(stdout).map_err(|err| { + eyre::eyre!("qmd query did not return JSON for {}: {err}", loaded.job.job_id) + })?; + let entries = results.as_array().cloned().unwrap_or_default(); + let mut evidence_ids = Vec::new(); + + for entry in &entries { + let entry_text = serde_json::to_string(entry)?; + + for item in corpus { + if entry_text.contains(format!("{}.md", crate::slug(&item.evidence_id)).as_str()) + || entry_text.contains(item.evidence_id.as_str()) + { + crate::push_unique(&mut evidence_ids, item.evidence_id.clone()); + } + } + } + + Ok((entries, evidence_ids)) +} + +pub(super) fn qmd_materialized_job( + loaded: &LoadedJob, + adapter_id: &str, + selected: SelectedEvidenceText, + latency_ms: f64, + returned_count: usize, + operator_debug: Option, + operator_debug_evidence: Option, +) -> MaterializedJob { + crate::materialized_job( + loaded, + adapter_id, + MaterializedJobInput { + content: selected.content, + evidence_ids: selected.evidence_ids, + pages: Vec::new(), + latency_ms, + indexing_latency_ms: None, + returned_count, + trace_id: None, + failure: None, + source_mappings: Vec::new(), + operator_debug, + operator_debug_evidence, + capture: None, + capture_failure: None, + consolidation_response: None, + consolidation: None, + knowledge: None, + temporal_reconciliation: None, + dreaming_readback: None, + memory_summaries: Vec::new(), + proactive_briefs: Vec::new(), + scheduled_tasks: Vec::new(), + trace_stages: None, + }, + ) +} diff --git a/apps/elf-eval/src/bin/real_world_live_adapter/qmd/runtime.rs b/apps/elf-eval/src/bin/real_world_live_adapter/qmd/runtime.rs new file mode 100644 index 00000000..10dbe7e6 --- /dev/null +++ b/apps/elf-eval/src/bin/real_world_live_adapter/qmd/runtime.rs @@ -0,0 +1,50 @@ +use crate::{ + AdapterKind, CommandEvidence, LoadedJob, MaterializedJob, MaterializedOutput, QmdArgs, Result, + aggregate_status, fs, + qmd::{checkout, job}, +}; + +pub(crate) fn run_qmd(args: QmdArgs) -> Result<()> { + let jobs = crate::load_jobs(&args.fixtures)?; + let result = materialize_qmd_jobs(&args, &jobs); + let materialized = match result { + Ok(jobs) => jobs, + Err(err) => + crate::failure_jobs(&args.adapter_id, &jobs, "qmd_cli_runtime", err.to_string()), + }; + + crate::write_materialized_output(MaterializedOutput { + adapter_id: &args.adapter_id, + adapter_kind: AdapterKind::QmdCliRuntime, + fixtures: &args.fixtures, + out_fixtures: &args.out_fixtures, + evidence_out: &args.evidence_out, + jobs: &jobs, + materialized: &materialized, + command_evidence: vec![CommandEvidence { + label: "qmd_cli_runtime".to_string(), + status: aggregate_status(&materialized), + command: "cargo run -p elf-eval --bin real_world_live_adapter -- qmd".to_string(), + artifact: Some(args.evidence_out.display().to_string()), + reason: "qmd live adapter used collection add, update, embed, and query --json." + .to_string(), + }], + metadata: None, + }) +} + +fn materialize_qmd_jobs(args: &QmdArgs, jobs: &[LoadedJob]) -> Result> { + fs::create_dir_all(&args.work_dir)?; + + let log_path = args.work_dir.join("qmd-live-real-world.log"); + + checkout::ensure_qmd_checkout(args, &log_path)?; + + let mut out = Vec::with_capacity(jobs.len()); + + for loaded in jobs { + out.push(job::materialize_qmd_job(args, loaded, &log_path)?); + } + + Ok(out) +} diff --git a/packages/elf-providers/src/rerank.rs b/packages/elf-providers/src/rerank.rs index 652abe09..2b644a08 100644 --- a/packages/elf-providers/src/rerank.rs +++ b/packages/elf-providers/src/rerank.rs @@ -1,52 +1,21 @@ //! Rerank-provider client helpers. -use std::{ - collections::HashSet, - sync::atomic::{AtomicU64, Ordering}, - time::Duration, -}; +mod local; +mod noise; +mod response; + +use std::time::Duration; use reqwest::Client; use serde_json::Value; -use crate::{Error, Result}; +use crate::Result; use elf_config::ProviderConfig; -static LOCAL_NOISE_CALL_COUNTER: AtomicU64 = AtomicU64::new(0); - -struct XorShift64 { - state: u64, -} -impl XorShift64 { - fn new(seed: u64) -> Self { - let state = if seed == 0 { 0x4D59_5DF4_D0F3_3173 } else { seed }; - - Self { state } - } - - fn next_u64(&mut self) -> u64 { - let mut x = self.state; - - x ^= x << 13; - x ^= x >> 7; - x ^= x << 17; - self.state = x; - - x - } - - fn next_f32(&mut self) -> f32 { - // Map to [0, 1). Keep 24 bits of precision for a stable f32. - let bits = (self.next_u64() >> 40) as u32; - - (bits as f32) / ((1_u32 << 24) as f32) - } -} - /// Reranks documents with the configured provider or local fallback implementation. pub async fn rerank(cfg: &ProviderConfig, query: &str, docs: &[String]) -> Result> { if cfg.provider_id == "local" { - return Ok(local_rerank_dispatch(cfg.model.as_str(), query, docs)); + return Ok(local::local_rerank_dispatch(cfg.model.as_str(), query, docs)); } let client = Client::builder().timeout(Duration::from_millis(cfg.timeout_ms)).build()?; @@ -60,190 +29,7 @@ pub async fn rerank(cfg: &ProviderConfig, query: &str, docs: &[String]) -> Resul .await?; let json: Value = res.error_for_status()?.json().await?; - parse_rerank_response(json, docs.len()) -} - -fn local_rerank_dispatch(model: &str, query: &str, docs: &[String]) -> Vec { - if let Some(noise_std) = parse_local_noisy_model(model) { - return local_rerank_noisy(query, docs, noise_std); - } - - local_rerank(query, docs) -} - -fn parse_local_noisy_model(model: &str) -> Option { - let prefix = "local-token-overlap-noisy@"; - let rest = model.strip_prefix(prefix)?; - let std: f32 = rest.parse().ok()?; - - Some(std.max(0.0)) -} - -fn local_rerank(query: &str, docs: &[String]) -> Vec { - let query_tokens = tokenize_ascii_alnum(query); - - if query_tokens.is_empty() { - return vec![0.0; docs.len()]; - } - - let denom = query_tokens.len() as f32; - let mut scores = Vec::with_capacity(docs.len()); - - for doc in docs { - let doc_tokens = tokenize_ascii_alnum(doc); - let matched = query_tokens.intersection(&doc_tokens).count() as f32; - - scores.push(matched / denom); - } - - scores -} - -fn local_rerank_noisy(query: &str, docs: &[String], noise_std: f32) -> Vec { - let base = local_rerank(query, docs); - - if noise_std <= 0.0 { - return base; - } - - let query_hash = blake3::hash(query.as_bytes()); - let mut seed_bytes = [0_u8; 8]; - - seed_bytes.copy_from_slice(&query_hash.as_bytes()[..8]); - // Vary the noise across calls to simulate reranker instability. - - let call_idx = LOCAL_NOISE_CALL_COUNTER.fetch_add(1, Ordering::Relaxed); - let mut seed = u64::from_le_bytes(seed_bytes); - - seed ^= call_idx.wrapping_mul(0x9E37_79B9_7F4A_7C15); - - let mut out = Vec::with_capacity(base.len()); - - for (i, score) in base.into_iter().enumerate() { - let mut rng = XorShift64::new(seed ^ (i as u64).wrapping_mul(0x9E37_79B9_7F4A_7C15)); - let u = rng.next_f32(); - let signed = (u * 2.0) - 1.0; - let noisy = score + signed * noise_std; - - out.push(noisy.clamp(0.0, 1.0)); - } - - out -} - -fn tokenize_ascii_alnum(text: &str) -> HashSet { - let mut normalized = String::with_capacity(text.len()); - - for ch in text.chars() { - if ch.is_ascii_alphanumeric() { - normalized.push(ch.to_ascii_lowercase()); - } else { - normalized.push(' '); - } - } - - let mut out = HashSet::new(); - - for token in normalized.split_whitespace() { - if token.len() < 2 { - continue; - } - - out.insert(token.to_string()); - } - - out -} - -fn parse_rerank_response(json: Value, doc_count: usize) -> Result> { - let results = - json.get("results").or_else(|| json.get("data")).and_then(|v| v.as_array()).ok_or_else( - || Error::InvalidResponse { - message: "Rerank response is missing results array.".to_string(), - }, - )?; - let mut scores = vec![0.0_f32; doc_count]; - - for item in results { - let index = item.get("index").and_then(|v| v.as_u64()).ok_or_else(|| { - Error::InvalidResponse { message: "Rerank result missing index.".to_string() } - })? as usize; - let score = item - .get("relevance_score") - .or_else(|| item.get("score")) - .and_then(|v| v.as_f64()) - .ok_or_else(|| Error::InvalidResponse { - message: "Rerank result missing score.".to_string(), - })? as f32; - - if index < scores.len() { - scores[index] = score; - } - } - - Ok(scores) + response::parse_rerank_response(json, docs.len()) } -#[cfg(test)] -mod tests { - use crate::rerank::{self}; - - #[test] - fn aligns_scores_by_index() { - let json = serde_json::json!({ - "results": [ - { "index": 1, "relevance_score": 0.2 }, - { "index": 0, "relevance_score": 0.9 } - ] - }); - let scores = rerank::parse_rerank_response(json, 2) - .expect("Rerank response parsing must succeed for the valid JSON fixture."); - - assert_eq!(scores, vec![0.9, 0.2]); - } - - #[test] - fn local_rerank_scores_match_token_overlap_fraction() { - let scores = - rerank::local_rerank("alpha beta", &[String::from("alpha"), String::from("gamma")]); - - assert_eq!(scores.len(), 2); - assert!((scores[0] - 0.5).abs() < 1e-6, "Unexpected score: {}", scores[0]); - assert_eq!(scores[1], 0.0); - } - - #[test] - fn local_noisy_model_is_detected_and_nonnegative() { - assert_eq!(rerank::parse_local_noisy_model("local-token-overlap"), None); - assert_eq!(rerank::parse_local_noisy_model("local-token-overlap-noisy@0.02"), Some(0.02)); - assert_eq!(rerank::parse_local_noisy_model("local-token-overlap-noisy@-1"), Some(0.0)); - } - - #[test] - fn local_rerank_noisy_varies_across_calls() { - // Use a base score away from 0 and 1 so clamping does not mask noise. - let docs = [String::from("alpha"), String::from("alpha")]; - let first = - rerank::local_rerank_dispatch("local-token-overlap-noisy@0.1", "alpha beta", &docs); - - assert!(first.iter().all(|v| (0.0..=1.0).contains(v))); - - let mut varied = false; - - for _ in 0..32 { - let next = - rerank::local_rerank_dispatch("local-token-overlap-noisy@0.1", "alpha beta", &docs); - - assert_eq!(first.len(), next.len()); - assert!(next.iter().all(|v| (0.0..=1.0).contains(v))); - - if next != first { - varied = true; - - break; - } - } - - assert!(varied, "Expected noisy rerank to vary across calls."); - } -} +#[cfg(test)] mod tests; diff --git a/packages/elf-providers/src/rerank/local.rs b/packages/elf-providers/src/rerank/local.rs new file mode 100644 index 00000000..6f52b779 --- /dev/null +++ b/packages/elf-providers/src/rerank/local.rs @@ -0,0 +1,83 @@ +use std::collections::HashSet; + +use crate::rerank::noise; + +pub(super) fn local_rerank_dispatch(model: &str, query: &str, docs: &[String]) -> Vec { + if let Some(noise_std) = parse_local_noisy_model(model) { + return local_rerank_noisy(query, docs, noise_std); + } + + local_rerank(query, docs) +} + +pub(super) fn parse_local_noisy_model(model: &str) -> Option { + let prefix = "local-token-overlap-noisy@"; + let rest = model.strip_prefix(prefix)?; + let std: f32 = rest.parse().ok()?; + + Some(std.max(0.0)) +} + +pub(super) fn local_rerank(query: &str, docs: &[String]) -> Vec { + let query_tokens = tokenize_ascii_alnum(query); + + if query_tokens.is_empty() { + return vec![0.0; docs.len()]; + } + + let denom = query_tokens.len() as f32; + let mut scores = Vec::with_capacity(docs.len()); + + for doc in docs { + let doc_tokens = tokenize_ascii_alnum(doc); + let matched = query_tokens.intersection(&doc_tokens).count() as f32; + + scores.push(matched / denom); + } + + scores +} + +fn local_rerank_noisy(query: &str, docs: &[String], noise_std: f32) -> Vec { + let base = local_rerank(query, docs); + + if noise_std <= 0.0 { + return base; + } + + let seed = noise::seed_for_query_call(query); + let mut out = Vec::with_capacity(base.len()); + + for (i, score) in base.into_iter().enumerate() { + let signed = noise::signed_unit_noise(seed, i); + let noisy = score + signed * noise_std; + + out.push(noisy.clamp(0.0, 1.0)); + } + + out +} + +fn tokenize_ascii_alnum(text: &str) -> HashSet { + let mut normalized = String::with_capacity(text.len()); + + for ch in text.chars() { + if ch.is_ascii_alphanumeric() { + normalized.push(ch.to_ascii_lowercase()); + } else { + normalized.push(' '); + } + } + + let mut out = HashSet::new(); + + for token in normalized.split_whitespace() { + if token.len() < 2 { + continue; + } + + out.insert(token.to_string()); + } + + out +} diff --git a/packages/elf-providers/src/rerank/noise.rs b/packages/elf-providers/src/rerank/noise.rs new file mode 100644 index 00000000..42ce9123 --- /dev/null +++ b/packages/elf-providers/src/rerank/noise.rs @@ -0,0 +1,53 @@ +use std::sync::atomic::{AtomicU64, Ordering}; + +static LOCAL_NOISE_CALL_COUNTER: AtomicU64 = AtomicU64::new(0); + +struct XorShift64 { + state: u64, +} +impl XorShift64 { + fn new(seed: u64) -> Self { + let state = if seed == 0 { 0x4D59_5DF4_D0F3_3173 } else { seed }; + + Self { state } + } + + fn next_u64(&mut self) -> u64 { + let mut x = self.state; + + x ^= x << 13; + x ^= x >> 7; + x ^= x << 17; + self.state = x; + + x + } + + fn next_f32(&mut self) -> f32 { + // Map to [0, 1). Keep 24 bits of precision for a stable f32. + let bits = (self.next_u64() >> 40) as u32; + + (bits as f32) / ((1_u32 << 24) as f32) + } +} + +pub(super) fn seed_for_query_call(query: &str) -> u64 { + let query_hash = blake3::hash(query.as_bytes()); + let mut seed_bytes = [0_u8; 8]; + + seed_bytes.copy_from_slice(&query_hash.as_bytes()[..8]); + + let call_idx = LOCAL_NOISE_CALL_COUNTER.fetch_add(1, Ordering::Relaxed); + let mut seed = u64::from_le_bytes(seed_bytes); + + seed ^= call_idx.wrapping_mul(0x9E37_79B9_7F4A_7C15); + + seed +} + +pub(super) fn signed_unit_noise(seed: u64, index: usize) -> f32 { + let mut rng = XorShift64::new(seed ^ (index as u64).wrapping_mul(0x9E37_79B9_7F4A_7C15)); + let u = rng.next_f32(); + + (u * 2.0) - 1.0 +} diff --git a/packages/elf-providers/src/rerank/response.rs b/packages/elf-providers/src/rerank/response.rs new file mode 100644 index 00000000..9926d94c --- /dev/null +++ b/packages/elf-providers/src/rerank/response.rs @@ -0,0 +1,32 @@ +use serde_json::Value; + +use crate::{Error, Result}; + +pub(super) fn parse_rerank_response(json: Value, doc_count: usize) -> Result> { + let results = + json.get("results").or_else(|| json.get("data")).and_then(|v| v.as_array()).ok_or_else( + || Error::InvalidResponse { + message: "Rerank response is missing results array.".to_string(), + }, + )?; + let mut scores = vec![0.0_f32; doc_count]; + + for item in results { + let index = item.get("index").and_then(|v| v.as_u64()).ok_or_else(|| { + Error::InvalidResponse { message: "Rerank result missing index.".to_string() } + })? as usize; + let score = item + .get("relevance_score") + .or_else(|| item.get("score")) + .and_then(|v| v.as_f64()) + .ok_or_else(|| Error::InvalidResponse { + message: "Rerank result missing score.".to_string(), + })? as f32; + + if index < scores.len() { + scores[index] = score; + } + } + + Ok(scores) +} diff --git a/packages/elf-providers/src/rerank/tests.rs b/packages/elf-providers/src/rerank/tests.rs new file mode 100644 index 00000000..bafe0433 --- /dev/null +++ b/packages/elf-providers/src/rerank/tests.rs @@ -0,0 +1,58 @@ +use crate::rerank::{local, response}; + +#[test] +fn aligns_scores_by_index() { + let json = serde_json::json!({ + "results": [ + { "index": 1, "relevance_score": 0.2 }, + { "index": 0, "relevance_score": 0.9 } + ] + }); + let scores = response::parse_rerank_response(json, 2) + .expect("Rerank response parsing must succeed for the valid JSON fixture."); + + assert_eq!(scores, vec![0.9, 0.2]); +} + +#[test] +fn local_rerank_scores_match_token_overlap_fraction() { + let scores = local::local_rerank("alpha beta", &[String::from("alpha"), String::from("gamma")]); + + assert_eq!(scores.len(), 2); + assert!((scores[0] - 0.5).abs() < 1e-6, "Unexpected score: {}", scores[0]); + assert_eq!(scores[1], 0.0); +} + +#[test] +fn local_noisy_model_is_detected_and_nonnegative() { + assert_eq!(local::parse_local_noisy_model("local-token-overlap"), None); + assert_eq!(local::parse_local_noisy_model("local-token-overlap-noisy@0.02"), Some(0.02)); + assert_eq!(local::parse_local_noisy_model("local-token-overlap-noisy@-1"), Some(0.0)); +} + +#[test] +fn local_rerank_noisy_varies_across_calls() { + // Use a base score away from 0 and 1 so clamping does not mask noise. + let docs = [String::from("alpha"), String::from("alpha")]; + let first = local::local_rerank_dispatch("local-token-overlap-noisy@0.1", "alpha beta", &docs); + + assert!(first.iter().all(|v| (0.0..=1.0).contains(v))); + + let mut varied = false; + + for _ in 0..32 { + let next = + local::local_rerank_dispatch("local-token-overlap-noisy@0.1", "alpha beta", &docs); + + assert_eq!(first.len(), next.len()); + assert!(next.iter().all(|v| (0.0..=1.0).contains(v))); + + if next != first { + varied = true; + + break; + } + } + + assert!(varied, "Expected noisy rerank to vary across calls."); +} diff --git a/packages/elf-service/tests/acceptance/docs_extension_v1.rs b/packages/elf-service/tests/acceptance/docs_extension_v1.rs index b6ac229c..b41f83d7 100644 --- a/packages/elf-service/tests/acceptance/docs_extension_v1.rs +++ b/packages/elf-service/tests/acceptance/docs_extension_v1.rs @@ -1,3 +1,5 @@ +mod lifecycle; + use std::{collections::HashSet, future::IntoFuture, string::ToString, sync::Arc, time::Instant}; use ahash::AHashMap; @@ -20,10 +22,10 @@ use uuid::Uuid; use crate::acceptance::{self, SpyExtractor, StubEmbedding, StubRerank, chunking::ChunkingConfig}; use elf_config::EmbeddingProviderConfig; use elf_service::{ - AddNoteInput, AddNoteRequest, BoxFuture, DocsDeleteRequest, DocsExcerptsGetRequest, - DocsGetRequest, DocsPutRequest, DocsPutResponse, DocsSearchL0Request, ElfService, - EmbeddingProvider, Error, NoteOp, PayloadLevel, Providers, Result, SearchRequest, - TextQuoteSelector, docs::DocRetrievalTrajectory, + AddNoteInput, AddNoteRequest, BoxFuture, DocsExcerptsGetRequest, DocsGetRequest, + DocsPutRequest, DocsPutResponse, DocsSearchL0Request, ElfService, EmbeddingProvider, Error, + PayloadLevel, Providers, Result, SearchRequest, TextQuoteSelector, + docs::DocRetrievalTrajectory, }; use elf_storage::{db::Db, qdrant::QdrantStore}; use elf_testkit::TestDatabase; @@ -239,470 +241,6 @@ async fn embed_handler( (StatusCode::OK, Json(serde_json::json!({ "data": data }))).into_response() } - -#[tokio::test] -#[ignore = "Requires external Postgres and Qdrant. Set ELF_PG_DSN and ELF_QDRANT_URL (or ELF_QDRANT_GRPC_URL) to run."] -async fn docs_put_get_excerpts_and_search_l0_work_end_to_end() { - let Some(ctx) = setup_docs_context().await else { return }; - let DocsContext { test_db, service } = ctx; - let put = put_test_doc(&service).await; - - assert_doc_get(&service, put.doc_id).await; - assert_doc_excerpt(&service, put.doc_id, put.content_hash.as_str()).await; - - let (handle, shutdown) = spawn_doc_worker(&service).await; - - assert!( - wait_for_doc_outbox_done(&service.db.pool, put.doc_id, std::time::Duration::from_secs(15)) - .await, - "Expected doc outbox to reach DONE." - ); - - assert_docs_search_l0(&service, put.doc_id).await; - - let _ = shutdown.send(()); - - handle.abort(); - - let _ = handle.await; - - drop(service); - - 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_URL (or ELF_QDRANT_GRPC_URL) to run."] -async fn docs_delete_marks_doc_deleted_and_removes_doc_vectors() { - let Some(ctx) = setup_docs_context().await else { return }; - let DocsContext { test_db, service } = ctx; - let put = put_test_doc(&service).await; - let (handle, shutdown) = spawn_doc_worker(&service).await; - - assert!( - wait_for_doc_outbox_done(&service.db.pool, put.doc_id, std::time::Duration::from_secs(15)) - .await, - "Expected doc UPSERT outbox to reach DONE." - ); - assert!( - fetch_first_doc_chunk_point(&service, put.doc_id).await.is_some(), - "Expected indexed doc chunk before delete." - ); - - let deleted = service - .docs_delete(DocsDeleteRequest { - tenant_id: "t".to_string(), - project_id: "p".to_string(), - agent_id: "owner".to_string(), - doc_id: put.doc_id, - }) - .await - .expect("Failed to delete Source Library doc."); - - assert_eq!(deleted.doc_id, put.doc_id); - assert_eq!(deleted.op, NoteOp::Delete); - assert!(deleted.chunk_delete_count > 0); - assert!( - wait_for_doc_outbox_done(&service.db.pool, put.doc_id, std::time::Duration::from_secs(15)) - .await, - "Expected doc DELETE outbox to reach DONE." - ); - - let get_after_delete = service - .docs_get(DocsGetRequest { - tenant_id: "t".to_string(), - project_id: "p".to_string(), - agent_id: "owner".to_string(), - read_profile: "private_plus_project".to_string(), - doc_id: put.doc_id, - }) - .await; - let search_after_delete = service - .docs_search_l0(DocsSearchL0Request { - tenant_id: "t".to_string(), - project_id: "p".to_string(), - caller_agent_id: "reader".to_string(), - scope: None, - status: None, - doc_type: None, - sparse_mode: None, - domain: None, - repo: None, - agent_id: None, - thread_id: None, - updated_after: None, - updated_before: None, - ts_gte: None, - ts_lte: None, - read_profile: "private_plus_project".to_string(), - query: "peregrine".to_string(), - top_k: Some(5), - candidate_k: Some(20), - explain: None, - }) - .await - .expect("Failed to search docs after delete."); - let second_delete = service - .docs_delete(DocsDeleteRequest { - tenant_id: "t".to_string(), - project_id: "p".to_string(), - agent_id: "owner".to_string(), - doc_id: put.doc_id, - }) - .await - .expect("Second Source Library delete should be idempotent."); - - assert!(matches!(get_after_delete, Err(Error::NotFound { .. }))); - assert!(search_after_delete.items.iter().all(|item| item.doc_id != put.doc_id)); - assert!( - fetch_first_doc_chunk_point(&service, put.doc_id).await.is_none(), - "Deleted Source Library doc chunk must be removed from Qdrant docs index." - ); - assert_eq!(second_delete.op, NoteOp::None); - assert_eq!(second_delete.chunk_delete_count, 0); - - let _ = shutdown.send(()); - - handle.abort(); - - let _ = handle.await; - - drop(service); - - 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_URL (or ELF_QDRANT_GRPC_URL) to run."] -async fn docs_put_source_library_records_do_not_create_memory_notes() { - let Some(ctx) = setup_docs_context().await else { return }; - let DocsContext { test_db, service } = ctx; - let before: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM memory_notes") - .fetch_one(&service.db.pool) - .await - .expect("Failed to count notes before docs_put."); - let put = put_test_doc_with( - &service, - "owner", - "project_shared", - Some("chat"), - "Captured thread", - serde_json::json!({ - "schema": "doc_source_ref/v1", - "doc_type": "chat", - "ts": "2026-02-25T12:00:00Z", - "thread_id": "thread-source-library-1", - "role": "user", - "source_kind": "social_thread", - "canonical_uri": "https://example.com/thread/source-library-1", - "captured_at": "2026-02-25T12:10:00Z", - "source_created_at": "2026-02-25T11:55:00Z", - "trust_label": "public_web", - "author": "Example Researcher", - "handle": "example-researcher", - "excerpt_locator": { - "quote": { - "exact": "Source libraries should preserve thread context." - } - } - }), - "Source libraries should preserve thread context. Agents can later promote only selected facts.", - ) - .await; - let after: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM memory_notes") - .fetch_one(&service.db.pool) - .await - .expect("Failed to count notes after docs_put."); - let doc_exists: bool = - sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM doc_documents WHERE doc_id = $1)") - .bind(put.doc_id) - .fetch_one(&service.db.pool) - .await - .expect("Failed to verify doc row."); - let stored_source_ref: serde_json::Value = - sqlx::query_scalar("SELECT source_ref FROM doc_documents WHERE doc_id = $1") - .bind(put.doc_id) - .fetch_one(&service.db.pool) - .await - .expect("Failed to fetch normalized source_ref."); - - assert!(doc_exists); - assert_eq!(after, before, "docs_put must not create durable Memory Notes."); - assert_eq!(put.source_capture.schema, "doc_source_capture/v1"); - assert_eq!(put.source_capture.source_record_id, put.doc_id); - assert_eq!(put.source_capture.origin, "https://example.com/thread/source-library-1"); - assert_eq!(put.source_capture.source_type, "social_thread"); - assert_eq!(put.source_capture.visibility_scope, "project_shared"); - assert!(!put.source_capture.source_spans.is_empty()); - assert_eq!(put.source_capture.source_spans[0].schema, "doc_source_span/v1"); - assert_eq!(put.source_capture.source_spans[0].status, "captured"); - assert_eq!(put.source_capture.source_spans[0].reason_code, None); - assert_eq!(stored_source_ref["source_record_id"], put.doc_id.to_string()); - assert_eq!(stored_source_ref["origin"], "https://example.com/thread/source-library-1"); - assert_eq!(stored_source_ref["source_type"], "social_thread"); - assert_eq!(stored_source_ref["content_hash"], put.content_hash); - assert!(stored_source_ref["source_spans"].as_array().is_some_and(|spans| !spans.is_empty())); - - drop(service); - - 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_URL (or ELF_QDRANT_GRPC_URL) to run."] -async fn docs_search_l0_respects_scope_doc_type_agent_id_and_updated_after_filters() { - let Some(ctx) = setup_docs_context().await else { return }; - let ( - test_db, - service, - shared_knowledge_doc, - _older_shared_knowledge_doc, - private_chat_doc, - handle, - shutdown, - ) = create_docs_search_filter_fixture(ctx).await; - let shared_scope_results = search_doc_ids_with_filters( - &service, - Some("project_shared"), - None, - None, - None, - None, - "reader", - ) - .await; - - assert!(shared_scope_results.contains(&shared_knowledge_doc)); - assert!(!shared_scope_results.contains(&private_chat_doc)); - - let chat_results = - search_doc_ids_with_filters(&service, None, Some("chat"), None, None, None, "reader").await; - - assert!(!chat_results.contains(&private_chat_doc)); - assert!(!chat_results.contains(&shared_knowledge_doc)); - - let assistant_chat_results = - search_doc_ids_with_filters(&service, None, Some("chat"), None, None, None, "assistant") - .await; - - assert!(assistant_chat_results.contains(&private_chat_doc)); - assert!(!assistant_chat_results.contains(&shared_knowledge_doc)); - - let assistant_results = - search_doc_ids_with_filters(&service, None, None, Some("assistant"), None, None, "reader") - .await; - - assert!(!assistant_results.contains(&private_chat_doc)); - assert!(!assistant_results.contains(&shared_knowledge_doc)); - - let past = (OffsetDateTime::now_utc() - time::Duration::seconds(60)) - .format(&Rfc3339) - .expect("Failed to format past RFC3339 timestamp."); - let future = (OffsetDateTime::now_utc() + time::Duration::seconds(60)) - .format(&Rfc3339) - .expect("Failed to format future RFC3339 timestamp."); - let updated_after_past_results = - search_doc_ids_with_filters(&service, None, None, None, Some(&past), None, "reader").await; - - assert!(updated_after_past_results.contains(&shared_knowledge_doc)); - assert!(!updated_after_past_results.contains(&private_chat_doc)); - - let updated_after_future_results = - search_doc_ids_with_filters(&service, None, None, None, Some(&future), None, "reader") - .await; - - assert!(updated_after_future_results.is_empty()); - - cleanup_docs_filter_fixture(test_db, handle, shutdown).await; -} - -#[tokio::test] -#[ignore = "Requires external Postgres and Qdrant. Set ELF_PG_DSN and ELF_QDRANT_URL (or ELF_QDRANT_GRPC_URL) to run."] -async fn docs_search_l0_respects_thread_id_filter_for_chat_docs() { - let Some(ctx) = setup_docs_context().await else { return }; - let ( - test_db, - service, - shared_knowledge_doc, - older_shared_knowledge_doc, - private_chat_doc, - handle, - shutdown, - ) = create_docs_search_filter_fixture(ctx).await; - let thread_filter_results = service - .docs_search_l0(DocsSearchL0Request { - tenant_id: "t".to_string(), - project_id: "p".to_string(), - caller_agent_id: "assistant".to_string(), - scope: None, - status: None, - doc_type: Some("chat".to_string()), - sparse_mode: None, - domain: None, - repo: None, - agent_id: None, - thread_id: Some("shared-chat-thread".to_string()), - updated_after: None, - updated_before: None, - ts_gte: None, - ts_lte: None, - read_profile: "private_plus_project".to_string(), - query: "peregrine".to_string(), - top_k: Some(20), - candidate_k: Some(50), - explain: None, - }) - .await - .expect("Failed to search docs with thread_id filter."); - let thread_filtered_docs = - thread_filter_results.items.into_iter().map(|item| item.doc_id).collect::>(); - - assert!(thread_filtered_docs.contains(&private_chat_doc)); - assert!(!thread_filtered_docs.contains(&shared_knowledge_doc)); - assert!(!thread_filtered_docs.contains(&older_shared_knowledge_doc)); - - cleanup_docs_filter_fixture(test_db, handle, shutdown).await; -} - -#[tokio::test] -#[ignore = "Requires external Postgres and Qdrant. Set ELF_PG_DSN and ELF_QDRANT_URL (or ELF_QDRANT_GRPC_URL) to run this test."] -async fn docs_search_l0_requires_chat_doc_type_for_thread_id() { - let Some(ctx) = setup_docs_context().await else { return }; - let ( - test_db, - service, - _shared_knowledge_doc, - _older_shared_knowledge_doc, - _private_chat_doc, - handle, - shutdown, - ) = create_docs_search_filter_fixture(ctx).await; - let result = service - .docs_search_l0(DocsSearchL0Request { - tenant_id: "t".to_string(), - project_id: "p".to_string(), - caller_agent_id: "assistant".to_string(), - scope: None, - status: None, - doc_type: None, - sparse_mode: None, - domain: None, - repo: None, - agent_id: None, - thread_id: Some("shared-chat-thread".to_string()), - updated_after: None, - updated_before: None, - ts_gte: None, - ts_lte: None, - read_profile: "private_plus_project".to_string(), - query: "peregrine".to_string(), - top_k: Some(20), - candidate_k: Some(50), - explain: None, - }) - .await; - - match result { - Err(Error::InvalidRequest { message }) => { - assert!(message.contains("thread_id requires")); - }, - other => { - panic!("Expected InvalidRequest for thread_id without chat doc_type, got {other:?}") - }, - } - - cleanup_docs_filter_fixture(test_db, handle, shutdown).await; -} - -#[tokio::test] -#[ignore = "Requires external Postgres and Qdrant. Set ELF_PG_DSN and ELF_QDRANT_URL (or ELF_QDRANT_GRPC_URL) to run this test."] -async fn docs_put_applies_write_policy_and_excerpt_by_chunk_id_is_verified() { - let Some(ctx) = setup_docs_context().await else { return }; - let DocsContext { test_db, service } = ctx; - let content = "Alpha normal text then secret sk-abcdef and trailing content."; - let secret = "sk-abcdef"; - let start = content.find(secret).expect("Expected secret in content."); - let end = start + secret.len(); - let write_policy = serde_json::from_value(serde_json::json!({ - "exclusions": [{"start": start, "end": end}], - })) - .expect("Failed to build write_policy."); - let put = service - .docs_put(DocsPutRequest { - tenant_id: "t".to_string(), - project_id: "p".to_string(), - agent_id: "owner".to_string(), - scope: "project_shared".to_string(), - doc_type: None, - title: Some("Docs write_policy sample".to_string()), - write_policy: Some(write_policy), - source_ref: serde_json::json!({ - "schema": "doc_source_ref/v1", - "doc_type": "knowledge", - "ts": "2026-02-25T12:00:00Z", - }), - content: content.to_string(), - }) - .await - .expect("Failed to put doc with write policy."); - let (handle, shutdown) = spawn_doc_worker(&service).await; - - assert!( - wait_for_doc_outbox_done(&service.db.pool, put.doc_id, std::time::Duration::from_secs(15)) - .await, - "Expected doc outbox to reach DONE." - ); - - let chunk_id = fetch_first_doc_chunk_id(&service, put.doc_id) - .await - .expect("Expected chunk id from transformed doc."); - let excerpt = service - .docs_excerpts_get(DocsExcerptsGetRequest { - tenant_id: "t".to_string(), - project_id: "p".to_string(), - agent_id: "reader".to_string(), - read_profile: "private_plus_project".to_string(), - doc_id: put.doc_id, - level: "L1".to_string(), - chunk_id: Some(chunk_id), - quote: None, - position: None, - explain: None, - }) - .await - .expect("Failed to hydrate excerpt by chunk_id."); - - assert!(excerpt.verification.verified); - assert!(!excerpt.excerpt.is_empty()); - assert!(!excerpt.excerpt.contains(secret)); - assert!(!excerpt.locator.span_id.is_nil()); - - let captured_chunk_span = put - .source_capture - .source_spans - .iter() - .find(|span| span.chunk_id == Some(chunk_id)) - .expect("Expected captured source span for hydrated chunk."); - - assert_eq!(excerpt.locator.span_id, captured_chunk_span.span_id); - assert_eq!(excerpt.verification.content_hash, put.content_hash); - assert!(put.write_policy_audit.is_some()); - assert_eq!(put.source_capture.policy_spans.len(), 1); - assert_eq!(put.source_capture.policy_spans[0].status, "excluded"); - assert_eq!( - put.source_capture.policy_spans[0].reason_code.as_deref(), - Some("WRITE_POLICY_EXCLUSION") - ); - - let _ = shutdown.send(()); - - handle.abort(); - - let _ = handle.await; - - 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_URL (or ELF_QDRANT_GRPC_URL) to run."] async fn docs_search_l0_respects_doc_ts_filter() { diff --git a/packages/elf-service/tests/acceptance/docs_extension_v1/lifecycle.rs b/packages/elf-service/tests/acceptance/docs_extension_v1/lifecycle.rs new file mode 100644 index 00000000..70f48b28 --- /dev/null +++ b/packages/elf-service/tests/acceptance/docs_extension_v1/lifecycle.rs @@ -0,0 +1,526 @@ +use std::collections::HashSet; + +use serde_json::Value; +use time::{Duration, OffsetDateTime, format_description::well_known::Rfc3339}; + +use crate::acceptance::docs_extension_v1::{self, DocsContext}; +use elf_service::{ + DocsDeleteRequest, DocsExcerptsGetRequest, DocsGetRequest, DocsPutRequest, DocsSearchL0Request, + Error, NoteOp, +}; + +#[tokio::test] +#[ignore = "Requires external Postgres and Qdrant. Set ELF_PG_DSN and ELF_QDRANT_URL (or ELF_QDRANT_GRPC_URL) to run."] +async fn docs_put_get_excerpts_and_search_l0_work_end_to_end() { + let Some(ctx) = docs_extension_v1::setup_docs_context().await else { return }; + let DocsContext { test_db, service } = ctx; + let put = docs_extension_v1::put_test_doc(&service).await; + + docs_extension_v1::assert_doc_get(&service, put.doc_id).await; + docs_extension_v1::assert_doc_excerpt(&service, put.doc_id, put.content_hash.as_str()).await; + + let (handle, shutdown) = docs_extension_v1::spawn_doc_worker(&service).await; + + assert!( + docs_extension_v1::wait_for_doc_outbox_done( + &service.db.pool, + put.doc_id, + std::time::Duration::from_secs(15) + ) + .await, + "Expected doc outbox to reach DONE." + ); + + docs_extension_v1::assert_docs_search_l0(&service, put.doc_id).await; + + let _ = shutdown.send(()); + + handle.abort(); + + let _ = handle.await; + + drop(service); + + 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_URL (or ELF_QDRANT_GRPC_URL) to run."] +async fn docs_delete_marks_doc_deleted_and_removes_doc_vectors() { + let Some(ctx) = docs_extension_v1::setup_docs_context().await else { return }; + let DocsContext { test_db, service } = ctx; + let put = docs_extension_v1::put_test_doc(&service).await; + let (handle, shutdown) = docs_extension_v1::spawn_doc_worker(&service).await; + + assert!( + docs_extension_v1::wait_for_doc_outbox_done( + &service.db.pool, + put.doc_id, + std::time::Duration::from_secs(15) + ) + .await, + "Expected doc UPSERT outbox to reach DONE." + ); + assert!( + docs_extension_v1::fetch_first_doc_chunk_point(&service, put.doc_id).await.is_some(), + "Expected indexed doc chunk before delete." + ); + + let deleted = service + .docs_delete(DocsDeleteRequest { + tenant_id: "t".to_string(), + project_id: "p".to_string(), + agent_id: "owner".to_string(), + doc_id: put.doc_id, + }) + .await + .expect("Failed to delete Source Library doc."); + + assert_eq!(deleted.doc_id, put.doc_id); + assert_eq!(deleted.op, NoteOp::Delete); + assert!(deleted.chunk_delete_count > 0); + assert!( + docs_extension_v1::wait_for_doc_outbox_done( + &service.db.pool, + put.doc_id, + std::time::Duration::from_secs(15) + ) + .await, + "Expected doc DELETE outbox to reach DONE." + ); + + let get_after_delete = service + .docs_get(DocsGetRequest { + tenant_id: "t".to_string(), + project_id: "p".to_string(), + agent_id: "owner".to_string(), + read_profile: "private_plus_project".to_string(), + doc_id: put.doc_id, + }) + .await; + let search_after_delete = service + .docs_search_l0(DocsSearchL0Request { + tenant_id: "t".to_string(), + project_id: "p".to_string(), + caller_agent_id: "reader".to_string(), + scope: None, + status: None, + doc_type: None, + sparse_mode: None, + domain: None, + repo: None, + agent_id: None, + thread_id: None, + updated_after: None, + updated_before: None, + ts_gte: None, + ts_lte: None, + read_profile: "private_plus_project".to_string(), + query: "peregrine".to_string(), + top_k: Some(5), + candidate_k: Some(20), + explain: None, + }) + .await + .expect("Failed to search docs after delete."); + let second_delete = service + .docs_delete(DocsDeleteRequest { + tenant_id: "t".to_string(), + project_id: "p".to_string(), + agent_id: "owner".to_string(), + doc_id: put.doc_id, + }) + .await + .expect("Second Source Library delete should be idempotent."); + + assert!(matches!(get_after_delete, Err(Error::NotFound { .. }))); + assert!(search_after_delete.items.iter().all(|item| item.doc_id != put.doc_id)); + assert!( + docs_extension_v1::fetch_first_doc_chunk_point(&service, put.doc_id).await.is_none(), + "Deleted Source Library doc chunk must be removed from Qdrant docs index." + ); + assert_eq!(second_delete.op, NoteOp::None); + assert_eq!(second_delete.chunk_delete_count, 0); + + let _ = shutdown.send(()); + + handle.abort(); + + let _ = handle.await; + + drop(service); + + 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_URL (or ELF_QDRANT_GRPC_URL) to run."] +async fn docs_put_source_library_records_do_not_create_memory_notes() { + let Some(ctx) = docs_extension_v1::setup_docs_context().await else { return }; + let DocsContext { test_db, service } = ctx; + let before: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM memory_notes") + .fetch_one(&service.db.pool) + .await + .expect("Failed to count notes before docs_put."); + let put = docs_extension_v1::put_test_doc_with( + &service, + "owner", + "project_shared", + Some("chat"), + "Captured thread", + serde_json::json!({ + "schema": "doc_source_ref/v1", + "doc_type": "chat", + "ts": "2026-02-25T12:00:00Z", + "thread_id": "thread-source-library-1", + "role": "user", + "source_kind": "social_thread", + "canonical_uri": "https://example.com/thread/source-library-1", + "captured_at": "2026-02-25T12:10:00Z", + "source_created_at": "2026-02-25T11:55:00Z", + "trust_label": "public_web", + "author": "Example Researcher", + "handle": "example-researcher", + "excerpt_locator": { + "quote": { + "exact": "Source libraries should preserve thread context." + } + } + }), + "Source libraries should preserve thread context. Agents can later promote only selected facts.", + ) + .await; + let after: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM memory_notes") + .fetch_one(&service.db.pool) + .await + .expect("Failed to count notes after docs_put."); + let doc_exists: bool = + sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM doc_documents WHERE doc_id = $1)") + .bind(put.doc_id) + .fetch_one(&service.db.pool) + .await + .expect("Failed to verify doc row."); + let stored_source_ref: Value = + sqlx::query_scalar("SELECT source_ref FROM doc_documents WHERE doc_id = $1") + .bind(put.doc_id) + .fetch_one(&service.db.pool) + .await + .expect("Failed to fetch normalized source_ref."); + + assert!(doc_exists); + assert_eq!(after, before, "docs_put must not create durable Memory Notes."); + assert_eq!(put.source_capture.schema, "doc_source_capture/v1"); + assert_eq!(put.source_capture.source_record_id, put.doc_id); + assert_eq!(put.source_capture.origin, "https://example.com/thread/source-library-1"); + assert_eq!(put.source_capture.source_type, "social_thread"); + assert_eq!(put.source_capture.visibility_scope, "project_shared"); + assert!(!put.source_capture.source_spans.is_empty()); + assert_eq!(put.source_capture.source_spans[0].schema, "doc_source_span/v1"); + assert_eq!(put.source_capture.source_spans[0].status, "captured"); + assert_eq!(put.source_capture.source_spans[0].reason_code, None); + assert_eq!(stored_source_ref["source_record_id"], put.doc_id.to_string()); + assert_eq!(stored_source_ref["origin"], "https://example.com/thread/source-library-1"); + assert_eq!(stored_source_ref["source_type"], "social_thread"); + assert_eq!(stored_source_ref["content_hash"], put.content_hash); + assert!(stored_source_ref["source_spans"].as_array().is_some_and(|spans| !spans.is_empty())); + + drop(service); + + 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_URL (or ELF_QDRANT_GRPC_URL) to run."] +async fn docs_search_l0_respects_scope_doc_type_agent_id_and_updated_after_filters() { + let Some(ctx) = docs_extension_v1::setup_docs_context().await else { return }; + let ( + test_db, + service, + shared_knowledge_doc, + _older_shared_knowledge_doc, + private_chat_doc, + handle, + shutdown, + ) = docs_extension_v1::create_docs_search_filter_fixture(ctx).await; + let shared_scope_results = docs_extension_v1::search_doc_ids_with_filters( + &service, + Some("project_shared"), + None, + None, + None, + None, + "reader", + ) + .await; + + assert!(shared_scope_results.contains(&shared_knowledge_doc)); + assert!(!shared_scope_results.contains(&private_chat_doc)); + + let chat_results = docs_extension_v1::search_doc_ids_with_filters( + &service, + None, + Some("chat"), + None, + None, + None, + "reader", + ) + .await; + + assert!(!chat_results.contains(&private_chat_doc)); + assert!(!chat_results.contains(&shared_knowledge_doc)); + + let assistant_chat_results = docs_extension_v1::search_doc_ids_with_filters( + &service, + None, + Some("chat"), + None, + None, + None, + "assistant", + ) + .await; + + assert!(assistant_chat_results.contains(&private_chat_doc)); + assert!(!assistant_chat_results.contains(&shared_knowledge_doc)); + + let assistant_results = docs_extension_v1::search_doc_ids_with_filters( + &service, + None, + None, + Some("assistant"), + None, + None, + "reader", + ) + .await; + + assert!(!assistant_results.contains(&private_chat_doc)); + assert!(!assistant_results.contains(&shared_knowledge_doc)); + + let past = (OffsetDateTime::now_utc() - Duration::seconds(60)) + .format(&Rfc3339) + .expect("Failed to format past RFC3339 timestamp."); + let future = (OffsetDateTime::now_utc() + Duration::seconds(60)) + .format(&Rfc3339) + .expect("Failed to format future RFC3339 timestamp."); + let updated_after_past_results = docs_extension_v1::search_doc_ids_with_filters( + &service, + None, + None, + None, + Some(&past), + None, + "reader", + ) + .await; + + assert!(updated_after_past_results.contains(&shared_knowledge_doc)); + assert!(!updated_after_past_results.contains(&private_chat_doc)); + + let updated_after_future_results = docs_extension_v1::search_doc_ids_with_filters( + &service, + None, + None, + None, + Some(&future), + None, + "reader", + ) + .await; + + assert!(updated_after_future_results.is_empty()); + + docs_extension_v1::cleanup_docs_filter_fixture(test_db, handle, shutdown).await; +} + +#[tokio::test] +#[ignore = "Requires external Postgres and Qdrant. Set ELF_PG_DSN and ELF_QDRANT_URL (or ELF_QDRANT_GRPC_URL) to run."] +async fn docs_search_l0_respects_thread_id_filter_for_chat_docs() { + let Some(ctx) = docs_extension_v1::setup_docs_context().await else { return }; + let ( + test_db, + service, + shared_knowledge_doc, + older_shared_knowledge_doc, + private_chat_doc, + handle, + shutdown, + ) = docs_extension_v1::create_docs_search_filter_fixture(ctx).await; + let thread_filter_results = service + .docs_search_l0(DocsSearchL0Request { + tenant_id: "t".to_string(), + project_id: "p".to_string(), + caller_agent_id: "assistant".to_string(), + scope: None, + status: None, + doc_type: Some("chat".to_string()), + sparse_mode: None, + domain: None, + repo: None, + agent_id: None, + thread_id: Some("shared-chat-thread".to_string()), + updated_after: None, + updated_before: None, + ts_gte: None, + ts_lte: None, + read_profile: "private_plus_project".to_string(), + query: "peregrine".to_string(), + top_k: Some(20), + candidate_k: Some(50), + explain: None, + }) + .await + .expect("Failed to search docs with thread_id filter."); + let thread_filtered_docs = + thread_filter_results.items.into_iter().map(|item| item.doc_id).collect::>(); + + assert!(thread_filtered_docs.contains(&private_chat_doc)); + assert!(!thread_filtered_docs.contains(&shared_knowledge_doc)); + assert!(!thread_filtered_docs.contains(&older_shared_knowledge_doc)); + + docs_extension_v1::cleanup_docs_filter_fixture(test_db, handle, shutdown).await; +} + +#[tokio::test] +#[ignore = "Requires external Postgres and Qdrant. Set ELF_PG_DSN and ELF_QDRANT_URL (or ELF_QDRANT_GRPC_URL) to run this test."] +async fn docs_search_l0_requires_chat_doc_type_for_thread_id() { + let Some(ctx) = docs_extension_v1::setup_docs_context().await else { return }; + let ( + test_db, + service, + _shared_knowledge_doc, + _older_shared_knowledge_doc, + _private_chat_doc, + handle, + shutdown, + ) = docs_extension_v1::create_docs_search_filter_fixture(ctx).await; + let result = service + .docs_search_l0(DocsSearchL0Request { + tenant_id: "t".to_string(), + project_id: "p".to_string(), + caller_agent_id: "assistant".to_string(), + scope: None, + status: None, + doc_type: None, + sparse_mode: None, + domain: None, + repo: None, + agent_id: None, + thread_id: Some("shared-chat-thread".to_string()), + updated_after: None, + updated_before: None, + ts_gte: None, + ts_lte: None, + read_profile: "private_plus_project".to_string(), + query: "peregrine".to_string(), + top_k: Some(20), + candidate_k: Some(50), + explain: None, + }) + .await; + + match result { + Err(Error::InvalidRequest { message }) => { + assert!(message.contains("thread_id requires")); + }, + other => { + panic!("Expected InvalidRequest for thread_id without chat doc_type, got {other:?}") + }, + } + + docs_extension_v1::cleanup_docs_filter_fixture(test_db, handle, shutdown).await; +} + +#[tokio::test] +#[ignore = "Requires external Postgres and Qdrant. Set ELF_PG_DSN and ELF_QDRANT_URL (or ELF_QDRANT_GRPC_URL) to run this test."] +async fn docs_put_applies_write_policy_and_excerpt_by_chunk_id_is_verified() { + let Some(ctx) = docs_extension_v1::setup_docs_context().await else { return }; + let DocsContext { test_db, service } = ctx; + let content = "Alpha normal text then secret sk-abcdef and trailing content."; + let secret = "sk-abcdef"; + let start = content.find(secret).expect("Expected secret in content."); + let end = start + secret.len(); + let write_policy = serde_json::from_value(serde_json::json!({ + "exclusions": [{"start": start, "end": end}], + })) + .expect("Failed to build write_policy."); + let put = service + .docs_put(DocsPutRequest { + tenant_id: "t".to_string(), + project_id: "p".to_string(), + agent_id: "owner".to_string(), + scope: "project_shared".to_string(), + doc_type: None, + title: Some("Docs write_policy sample".to_string()), + write_policy: Some(write_policy), + source_ref: serde_json::json!({ + "schema": "doc_source_ref/v1", + "doc_type": "knowledge", + "ts": "2026-02-25T12:00:00Z", + }), + content: content.to_string(), + }) + .await + .expect("Failed to put doc with write policy."); + let (handle, shutdown) = docs_extension_v1::spawn_doc_worker(&service).await; + + assert!( + docs_extension_v1::wait_for_doc_outbox_done( + &service.db.pool, + put.doc_id, + std::time::Duration::from_secs(15) + ) + .await, + "Expected doc outbox to reach DONE." + ); + + let chunk_id = docs_extension_v1::fetch_first_doc_chunk_id(&service, put.doc_id) + .await + .expect("Expected chunk id from transformed doc."); + let excerpt = service + .docs_excerpts_get(DocsExcerptsGetRequest { + tenant_id: "t".to_string(), + project_id: "p".to_string(), + agent_id: "reader".to_string(), + read_profile: "private_plus_project".to_string(), + doc_id: put.doc_id, + level: "L1".to_string(), + chunk_id: Some(chunk_id), + quote: None, + position: None, + explain: None, + }) + .await + .expect("Failed to hydrate excerpt by chunk_id."); + + assert!(excerpt.verification.verified); + assert!(!excerpt.excerpt.is_empty()); + assert!(!excerpt.excerpt.contains(secret)); + assert!(!excerpt.locator.span_id.is_nil()); + + let captured_chunk_span = put + .source_capture + .source_spans + .iter() + .find(|span| span.chunk_id == Some(chunk_id)) + .expect("Expected captured source span for hydrated chunk."); + + assert_eq!(excerpt.locator.span_id, captured_chunk_span.span_id); + assert_eq!(excerpt.verification.content_hash, put.content_hash); + assert!(put.write_policy_audit.is_some()); + assert_eq!(put.source_capture.policy_spans.len(), 1); + assert_eq!(put.source_capture.policy_spans[0].status, "excluded"); + assert_eq!( + put.source_capture.policy_spans[0].reason_code.as_deref(), + Some("WRITE_POLICY_EXCLUSION") + ); + + let _ = shutdown.send(()); + + handle.abort(); + + let _ = handle.await; + + test_db.cleanup().await.expect("Failed to cleanup test database."); +}