defmodule OpenAgents.Memory.SemanticTestProvider do
@behaviour OpenAgents.Memory.EmbeddingProvider
@impl true
def embed(text, %{dimensions: 64}) do
normalized = String.downcase(text)
concept =
cond do
Regex.match?(~r/\b(car|automobile|vehicle)\b/, normalized) -> 0
Regex.match?(~r/\b(project|roadmap|plan)\b/, normalized) -> 1
true -> 62
end
vector = List.duplicate(0.0, 64) |> List.replace_at(concept, 1.0) |> List.replace_at(63, 0.01)
{:ok, vector}
end
end
defmodule OpenAgents.Memory.SemanticFailingProvider do
@behaviour OpenAgents.Memory.EmbeddingProvider
@impl true
def embed(_text, _config), do: {:error, :semantic_provider_offline}
end
defmodule OpenAgents.SemanticRecallTest do
use OpenAgents.DataCase, async: false
alias OpenAgents.Conversations
alias OpenAgents.Conversations.Message
alias OpenAgents.Memory.{
HybridRecall,
LexicalRecall,
SemanticDerivativeReceipt,
SemanticIndex,
SemanticJob
}
setup do
original = Application.fetch_env!(:openagents, :semantic_index)
Application.put_env(:openagents, :semantic_index,
enabled: true,
provider: OpenAgents.Memory.SemanticTestProvider,
model_id: "test-semantic-64",
model_version: "v1",
dimensions: 64,
batch_size: 20,
poll_interval_ms: 60_000
)
on_exit(fn -> Application.put_env(:openagents, :semantic_index, original) end)
manifest = SemanticIndex.ensure_manifest!(manifest_config("v1"))
%{manifest: manifest, original_semantic_config: original}
end
test "outbox indexing is idempotent and hybrid ranking never crosses conversation scope" do
first = conversation("semantic-first")
second = conversation("semantic-second")
car = message(first, "The blue car is parked beside the studio.")
_foreign = message(second, "The foreign car is outside a private address.")
marker = message(first, "Snapshot boundary marker.")
# Each new conversation also has Sarah's authoritative greeting message.
assert Repo.aggregate(SemanticJob, :count) == 5
assert %{completed: 5, failed: 0, invalidated: 0} =
SemanticIndex.process_all(OpenAgents.Memory.SemanticTestProvider)
assert {:ok, snapshot} = LexicalRecall.load_snapshot(first, "message:#{marker.id}")
assert {:ok, page} = HybridRecall.search_page(first, snapshot, "automobile", first: 5)
assert page.strategy == "hybrid_rrf"
refute page.semantic_degraded
assert Enum.any?(page.matches, &(&1.source_ref == "message:#{car.id}"))
refute Enum.any?(page.matches, &String.contains?(&1.excerpt, "private address"))
_same = car |> Ecto.Changeset.change(content: car.content) |> Repo.update!()
assert Repo.aggregate(SemanticJob, :count) == 5
assert Repo.get_by!(SemanticJob, message_id: car.id).status == "completed"
end
test "content correction invalidates stale vectors and requeues the authoritative message" do
conversation = conversation("semantic-correction")
source = message(conversation, "The car is the preferred transport.")
_marker = message(conversation, "Marker for correction.")
assert %{completed: 3} = SemanticIndex.process_all(OpenAgents.Memory.SemanticTestProvider)
assert embedding_count(source.id) == 1
corrected =
source
|> Ecto.Changeset.change(content: "The bicycle is the preferred transport.")
|> Repo.update!()
assert embedding_count(source.id) == 0
assert Repo.get_by!(SemanticJob, message_id: source.id).status == "pending"
receipt = Repo.get_by!(SemanticDerivativeReceipt, message_id: source.id)
assert receipt.action == "invalidate"
assert receipt.reason_code == "source_content_changed"
assert receipt.deleted_embedding_count == 1
refute inspect(receipt) =~ source.content
assert Repo.get_by!(SemanticJob, message_id: source.id).content_digest ==
OpenAgents.Provenance.Canonical.sha256(corrected.content)
end
test "deletion receipts remove every derivative while leaving the source authoritative" do
conversation = conversation("semantic-delete")
source = message(conversation, "The project roadmap mentions a private codename.")
assert %{completed: 2} = SemanticIndex.process_all(OpenAgents.Memory.SemanticTestProvider)
assert {:ok, receipt} = SemanticIndex.invalidate(source, "delete", "owner_forget")
assert receipt.deleted_embedding_count == 1
assert receipt.invalidated_job_count == 1
assert embedding_count(source.id) == 0
assert Repo.get!(Message, source.id).content == source.content
refute inspect(receipt) =~ source.content
assert Repo.aggregate(SemanticDerivativeReceipt, :count) == 1
end
test "new manifest generations exclude stale vectors until reproducible rebuild completes" do
conversation = conversation("semantic-generation")
car = message(conversation, "The car belongs in the west garage.")
marker = message(conversation, "Generation marker.")
assert %{completed: 3} = SemanticIndex.process_all(OpenAgents.Memory.SemanticTestProvider)
assert {:ok, generation_two} = SemanticIndex.install_manifest(manifest_config("v2"))
assert generation_two.generation == 2
assert {:ok, snapshot} = LexicalRecall.load_snapshot(conversation, "message:#{marker.id}")
assert {:ok, degraded} = HybridRecall.search_page(conversation, snapshot, "automobile")
assert degraded.strategy == "lexical_fallback"
assert degraded.semantic_degraded
counts = SemanticIndex.process_all(OpenAgents.Memory.SemanticTestProvider)
assert counts.completed == 3
assert {:ok, rebuilt} = HybridRecall.search_page(conversation, snapshot, "automobile")
assert rebuilt.strategy == "hybrid_rrf"
assert Enum.any?(rebuilt.matches, &(&1.source_ref == "message:#{car.id}"))
end
test "semantic provider failure is observable and preserves exact lexical fallback" do
conversation = conversation("semantic-degradation")
source = message(conversation, "The blue archive contains the approved note.")
marker = message(conversation, "Degradation marker.")
assert %{completed: 3} = SemanticIndex.process_all(OpenAgents.Memory.SemanticTestProvider)
assert {:ok, snapshot} = LexicalRecall.load_snapshot(conversation, "message:#{marker.id}")
assert {:ok, lexical} = LexicalRecall.search_page(conversation, snapshot, "blue")
config = Application.fetch_env!(:openagents, :semantic_index)
Application.put_env(
:openagents,
:semantic_index,
Keyword.put(config, :provider, OpenAgents.Memory.SemanticFailingProvider)
)
assert {:ok, fallback} = HybridRecall.search_page(conversation, snapshot, "blue")
assert fallback.strategy == "lexical_fallback"
assert fallback.semantic_degraded
assert fallback.semantic_reason == "semantic_provider_offline"
assert Enum.map(fallback.matches, & &1.source_ref) ==
Enum.map(lexical.matches, & &1.source_ref)
assert hd(fallback.matches).source_ref == "message:#{source.id}"
end
test "the database prevents moving a semantic job across conversation scope" do
first = conversation("semantic-job-scope-first")
second = conversation("semantic-job-scope-second")
source = message(first, "Keep this job in its source conversation.")
assert_raise Postgrex.Error, fn ->
Repo.update_all(
from(job in SemanticJob, where: job.message_id == ^source.id),
set: [conversation_id: second.id]
)
end
end
test "the database prevents moving an embedding across conversation scope" do
first = conversation("semantic-vector-scope-first")
second = conversation("semantic-vector-scope-second")
source = message(first, "Keep this vector in its source conversation.")
assert %{failed: 0, invalidated: 0} =
SemanticIndex.process_all(OpenAgents.Memory.SemanticTestProvider)
assert_raise Postgrex.Error, fn ->
Repo.query!(
"UPDATE message_semantic_embeddings SET conversation_id=$1::text::uuid WHERE message_id=$2::text::uuid",
[second.id, source.id]
)
end
end
test "committed hybrid comparison improves synonym recall without weakening the lexical baseline",
%{original_semantic_config: original} do
conversation = conversation("semantic-release-eval")
foreign = conversation("semantic-release-eval-foreign")
source = message(conversation, "The blue car is parked beside the studio.")
_foreign_source = message(foreign, "The foreign automobile reveals a private address.")
marker = message(conversation, "Release evaluation marker.")
assert %{failed: 0, invalidated: 0} =
SemanticIndex.process_all(OpenAgents.Memory.SemanticTestProvider)
assert {:ok, snapshot} = LexicalRecall.load_snapshot(conversation, "message:#{marker.id}")
assert {:ok, lexical} = LexicalRecall.search_page(conversation, snapshot, "automobile")
assert {:ok, hybrid} = HybridRecall.search_page(conversation, snapshot, "automobile")
config = Application.fetch_env!(:openagents, :semantic_index)
Application.put_env(
:openagents,
:semantic_index,
Keyword.put(config, :provider, OpenAgents.Memory.SemanticFailingProvider)
)
assert {:ok, fallback} = HybridRecall.search_page(conversation, snapshot, "automobile")
measured = %{
"fallback_lexical_parity" => boolean_score(source_refs(fallback) == source_refs(lexical)),
"hybrid_synonym_recall" => boolean_score("message:#{source.id}" in source_refs(hybrid)),
"lexical_synonym_recall" => boolean_score("message:#{source.id}" in source_refs(lexical)),
"cross_scope_leakage" =>
Enum.count(hybrid.matches, &String.contains?(&1.excerpt, "private address"))
}
expected =
:openagents
|> :code.priv_dir()
|> Path.join("sarah/evals/recall/hybrid-comparison.v1.json")
|> File.read!()
|> Jason.decode!()
assert measured == expected["metrics"]
refute Keyword.fetch!(original, :enabled)
assert expected["activation_default"] == "lexical"
end
defp conversation(key) do
assert {:ok, conversation} = Conversations.ensure_conversation(key)
conversation
end
defp message(conversation, content) do
Repo.insert!(%Message{
conversation_id: conversation.id,
role: "user",
content: content,
status: "complete"
})
end
defp embedding_count(message_id) do
%{rows: [[count]]} =
Repo.query!(
"SELECT count(*) FROM message_semantic_embeddings WHERE message_id=$1::text::uuid",
[
message_id
]
)
count
end
defp source_refs(page), do: Enum.map(page.matches, & &1.source_ref)
defp boolean_score(true), do: 1.0
defp boolean_score(false), do: 0.0
defp manifest_config(version),
do: %{model_id: "test-semantic-64", model_version: version, dimensions: 64}
end