From 48e3ccab75726807682a6292197877efa491b14b Mon Sep 17 00:00:00 2001 From: Memoryworld Date: Tue, 22 Sep 2026 19:07:05 +1000 Subject: [PATCH] Persist agent memory and isolate prompt evaluation --- .github/workflows/ci.yml | 22 +++ README.md | 56 +++++- examples/07_generate_demo_data.py | 12 +- src/self_evolving/__init__.py | 6 + src/self_evolving/core/agent.py | 55 ++++-- src/self_evolving/core/environment.py | 10 ++ src/self_evolving/dashboard/demo_data.py | 87 +++++---- src/self_evolving/evaluation/benchmark.py | 81 +++++++-- .../evolution/memory/episodic.py | 58 ++++-- src/self_evolving/evolution/prompt/opro.py | 4 + .../mechanisms/reflection/reflexion.py | 76 ++++---- src/self_evolving/persistence/sqlite_store.py | 100 ++++++++++- src/self_evolving/service/api.py | 8 +- tests/conftest.py | 51 ++++++ tests/test_api.py | 66 +++++++ tests/test_benchmark_runner.py | 2 + tests/test_demo_data.py | 46 +++++ tests/test_evaluation_protocol.py | 114 ++++++++++++ tests/test_memory_lifecycle.py | 165 ++++++++++++++++++ tests/test_reflexion.py | 90 ++++++++++ tests/test_sqlite_store.py | 35 ++++ tests/test_vector_memory.py | 3 +- 22 files changed, 1024 insertions(+), 123 deletions(-) create mode 100644 .github/workflows/ci.yml create mode 100644 tests/conftest.py create mode 100644 tests/test_evaluation_protocol.py create mode 100644 tests/test_memory_lifecycle.py create mode 100644 tests/test_reflexion.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..d17ab00 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,22 @@ +name: Offline regressions + +on: [push, pull_request] + +jobs: + tests: + strategy: + matrix: + os: [ubuntu-latest, windows-latest] + runs-on: ${{ matrix.os }} + env: + LITELLM_LOCAL_MODEL_COST_MAP: "True" + DO_NOT_TRACK: "1" + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + - run: python -m pip install -e ".[dev]" + - run: python -m ruff check --select E9,F63,F7,F82 src tests examples/07_generate_demo_data.py + - run: python -m pytest tests -q + - run: python examples/07_generate_demo_data.py --db-path .data/ci-demo.db --benchmark-dir .data/ci-benchmarks diff --git a/README.md b/README.md index 16fa903..bc316de 100644 --- a/README.md +++ b/README.md @@ -75,6 +75,9 @@ src/self_evolving/ - Retrieval now uses vector similarity as the primary score. - Default setup uses a lightweight local hashing embedder so the project works without extra model downloads. - The embedder backend is pluggable, so this can later be swapped to `sentence-transformers` or an external embedding model. +- Retrieval is `cosine × importance + 0.15 × lexical overlap`, with a linear scan of the retained entries. Access count is tracked, not used as a ranking factor. +- Empty memory now accumulates its first lesson. Completed episodes checkpoint the **full** memory: summaries, evictions, access counts and the stored-episode count survive reconstruction with the same database and `agent_id`. +- Retrieval outside `agent.run` requires an explicit `agent.save_memory()` to checkpoint its updated access counts. Failed episodes that raise before checkpoint do not save these changes. ### Embedders - `src/self_evolving/evolution/memory/embedders.py` @@ -87,7 +90,10 @@ src/self_evolving/ - Persists: - runs - steps - - episodic memory entries +- episodic memory entries +- Memory checkpoints replace one agent's snapshot atomically with an optimistic revision check. Concurrent stale writers fail with `MemoryConflictError` instead of overwriting a newer snapshot; API jobs expose this as `failed` with an error message. +- `save_memory_entries` and `list_memory` retain their append/query interfaces. Internal snapshot loads preserve oldest-first order; legacy entries migrate automatically, but their unknown historical store count starts at zero. +- For read-only QA, reload with `agent.load_memory()` and rerun after a conflict. For tools with external side effects, reconcile those effects before retrying. Memory and trajectory writes are separate transactions, not an exactly-once workflow. ### API service - `src/self_evolving/service/api.py` @@ -109,6 +115,8 @@ src/self_evolving/ ### Reflection - `src/self_evolving/mechanisms/reflection/reflexion.py` - Adds post-failure reflection and retry behavior. +- Retries use a fresh conversation and the previous attempt's reflection. The original prompt and reflector are restored on success, exhaustion, or exceptions. +- Each attempt is persisted separately; the final run also records attempt IDs/counts and total attempt steps. Reflection text is included as unverified context when distilling a lesson, not treated as verified knowledge or a model-weight update. ### Reward scoring - `src/self_evolving/mechanisms/reward/scorer.py` @@ -130,6 +138,9 @@ src/self_evolving/ - reflexion - prompt optimization - Writes JSON artifacts for each variant plus a session summary. +- Explicit `tuning_tasks` are used only to score OPRO prompt candidates; the selected prompt is then evaluated on `tasks`. Exact normalized goal overlap is rejected. Without `tuning_tasks`, the protocol is labeled `resubstitution`, never held-out evaluation. +- Sessions have unique directories and agent IDs, so concurrent sessions do not overwrite artifacts or inherit a previous benchmark's memories. Summaries include the task manifest, protocol and data source. +- The QA scorer is a case-insensitive reference substring smoke test, not exact-match accuracy. Memory uses sequential online adaptation; Reflexion allows up to two attempts. Final-attempt mean steps and total retry steps are both identified, but reflection/OPRO token cost and latency are **not** measured. These variants do not represent equal-cost quality comparisons. ### Dashboard - `app.py` @@ -166,8 +177,8 @@ That is not yet enough for: ### Engineering gaps - no container delivery files -- no CI workflow -- no background job execution for long-running runs and benchmarks +- CI runs the offline regression suite and synthetic demo on Windows and Linux +- background jobs exist, but job state is in memory and is lost on API restart ### Systems / platform gaps - no safe tool sandbox @@ -212,6 +223,32 @@ cp .env.example .env ## Quick Start +For a reproducible offline acceptance run, no API keys or `.env` file are needed: + +```bash +python -m pip install -e ".[dev]" +python -m pytest tests -q +python examples/07_generate_demo_data.py --db-path .data/offline-demo.db --benchmark-dir .data/offline-benchmarks +``` + +The tests block provider calls and socket connections (except Windows' internal asyncio socket-pair setup). The demo uses explicit deterministic agent/memory/reflection/optimizer implementations, never global patches; its scores are **synthetic fixtures**, not real model results. Package imports use LiteLLM's bundled cost map by default to avoid a metadata download. + +For an actual model-backed prompt experiment, provide independent tuning and evaluation tasks: + +```python +from self_evolving.evaluation.benchmark import BenchmarkRunner + +runner = BenchmarkRunner( + tuning_tasks=[("What is 3 * 7?", "21"), ("Who wrote Hamlet?", "Shakespeare")], + tasks=[("What is 8 * 9?", "72"), ("What is the capital of France?", "Paris")], + model="your/configured-model", +) +summary = runner.run(["baseline", "prompt_optimization"]) +print(summary["evaluation_protocol"]) +``` + +These tiny examples demonstrate split wiring only. Disjoint question strings do not establish semantic independence; review task families and answer leakage before reporting a real benchmark result. Model-backed examples below require your own provider configuration and may incur API charges. + ```python from dotenv import load_dotenv load_dotenv() @@ -263,12 +300,17 @@ curl -X POST http://127.0.0.1:8000/benchmarks/qa \ {"goal": "What is the capital of France?", "reference_answer": "Paris"}, {"goal": "What is 12 * 7?", "reference_answer": "84"} ], - "variants": ["baseline", "memory", "reflexion"] + "tuning_tasks": [ + {"goal": "What is 3 * 7?", "reference_answer": "21"} + ], + "variants": ["baseline", "memory", "reflexion", "prompt_optimization"] }' ``` Inspect recent jobs: +Completed benchmark jobs return `evaluation_protocol` and `data_source` alongside their variant results; the dashboard's session summary displays the same fields. If `tuning_tasks` is omitted, OPRO output is explicitly marked `resubstitution`. + ```bash curl http://127.0.0.1:8000/jobs ``` @@ -334,6 +376,8 @@ This populates: Use it when you want the dashboard to have data immediately without calling a real external model. +The generated runs use model label `offline/deterministic-fixture` and benchmark JSON uses `data_source=synthetic_deterministic_fixture`. No learned-performance claim can be inferred from those values. + ## Next Step The next practical step after the current background-job control plane is: @@ -354,6 +398,10 @@ In other words, the next upgrade is turning the current local control plane into pytest tests/ -v ``` +Regression coverage includes empty-memory cold start; app/agent reconstruction and memory injection; summary/eviction/counter restoration; snapshot rollback and stale-writer conflicts; actual Reflexion failure/retry/cleanup and persisted attempt evidence; OPRO tuning/evaluation separation; unique same-second artifacts; and the fully offline demo. + +Remaining limits: the default hashing vectors are not a transformer semantic model; embedding model/version migration is not automatic. Memory lacks trusted-source filtering, semantic duplicate detection and tenant authorization. Tool learning is an experimental Python execution path, not a safe sandbox. Background job state remains process-local. No real-model improvement, training result, throughput or token-cost benchmark has been measured by the offline suite. + ## Project Direction The target outcome for this repo is no longer just: diff --git a/examples/07_generate_demo_data.py b/examples/07_generate_demo_data.py index de6edfe..bbaf23f 100644 --- a/examples/07_generate_demo_data.py +++ b/examples/07_generate_demo_data.py @@ -9,15 +9,21 @@ python examples/07_generate_demo_data.py """ -from dotenv import load_dotenv +import argparse +import os -load_dotenv() +# An offline CLI must also avoid import-time provider metadata refreshes. +os.environ["LITELLM_LOCAL_MODEL_COST_MAP"] = "True" from self_evolving.dashboard.demo_data import generate_demo_data def main(): - result = generate_demo_data() + parser = argparse.ArgumentParser(description="Generate synthetic offline data, not model scores") + parser.add_argument("--db-path", default=".data/sea.db") + parser.add_argument("--benchmark-dir", default="runs/benchmarks") + args = parser.parse_args() + result = generate_demo_data(args.db_path, args.benchmark_dir) print(result) diff --git a/src/self_evolving/__init__.py b/src/self_evolving/__init__.py index a459454..b4b1771 100644 --- a/src/self_evolving/__init__.py +++ b/src/self_evolving/__init__.py @@ -3,6 +3,12 @@ Based on: "A Survey of Self-Evolving Agents" (arXiv:2507.21046) and "A Comprehensive Survey of Self-Evolving AI Agents" (arXiv:2508.07407) """ +import os + +# Importing the framework should not fetch provider metadata. Real completions +# still require an explicit call; users can opt into a refreshed cost map. +os.environ.setdefault("LITELLM_LOCAL_MODEL_COST_MAP", "True") + from self_evolving.core.agent import BaseAgent from self_evolving.core.environment import Environment from self_evolving.core.types import AgentState, Trajectory, Feedback diff --git a/src/self_evolving/core/agent.py b/src/self_evolving/core/agent.py index bbb1dfc..51a55e4 100644 --- a/src/self_evolving/core/agent.py +++ b/src/self_evolving/core/agent.py @@ -8,6 +8,7 @@ import os import uuid import logging +from copy import deepcopy from typing import Any, Callable, Optional, TYPE_CHECKING import litellm @@ -15,7 +16,7 @@ from self_evolving.core.types import ( AgentState, Trajectory, Step, Feedback, FeedbackType, Message, - EvolutionRecord, EvolutionTarget, EvolutionStage, + EvolutionRecord, ) from self_evolving.core.environment import Environment @@ -37,7 +38,7 @@ class BaseAgent: action = agent.act(obs) obs, feedback, done = env.step(action) agent.observe(feedback) - agent.evolve(trajectory) ← inter-test-time evolution hook + optionally reflect, distil memory, and persist the episode """ DEFAULT_SYSTEM = ( @@ -52,6 +53,8 @@ def __init__( max_steps: int = 20, agent_id: Optional[str] = None, ): + if max_steps < 1: + raise ValueError("max_steps must be positive") self.agent_id = agent_id or str(uuid.uuid4())[:8] self.model = model or os.getenv("SEA_MODEL", "deepseek/deepseek-chat") self.state = AgentState( @@ -67,6 +70,7 @@ def __init__( self._conversation: list[Message] = [] self._current_trajectory: Optional[Trajectory] = None + self._memory_revision: Optional[int] = None # ------------------------------------------------------------------ # Public API @@ -80,6 +84,8 @@ def run( progress_callback: Optional[Callable[[float, str, dict[str, Any]], None]] = None, ) -> Trajectory: """Execute a full task episode and return the trajectory.""" + if self.memory is not None and self.store is not None and self._memory_revision is None: + self.load_memory(preserve_initial=True) task_id = task_id or str(uuid.uuid4())[:8] obs = env.reset(goal) trajectory = Trajectory(task_id=task_id, goal=goal) @@ -124,10 +130,9 @@ def run( trajectory = self.reflector.reflect(trajectory) # Store episode in memory - if self.memory: - memory_entries = self.memory.store(trajectory) - if self.store: - self.store.save_memory_entries(self.agent_id, memory_entries) + if self.memory is not None: + self.memory.store(trajectory) + self.save_memory() if self.store: trajectory.metadata["run_id"] = self.store.save_trajectory( @@ -218,13 +223,43 @@ def _call_llm(self, messages: list[dict]) -> str: # State management # ------------------------------------------------------------------ + def load_memory(self, *, preserve_initial: bool = False) -> None: + """Reload a persisted snapshot before running; safe for an empty memory.""" + if self.memory is None or self.store is None: + return + snapshot = self.store.load_memory_snapshot(self.agent_id) + if not preserve_initial or snapshot["revision"] or snapshot["entries"]: + self.memory.load(snapshot["entries"], stored_count=snapshot["stored_count"]) + self._memory_revision = snapshot["revision"] + + def save_memory(self) -> None: + """Checkpoint the complete memory, including retrieval access counts. + + Concurrent stale snapshots raise MemoryConflictError; they are never + merged blindly. Call load_memory and rerun the task after a conflict. + """ + if self.memory is None or self.store is None: + return + expected = self._memory_revision if self._memory_revision is not None else 0 + self._memory_revision = self.store.save_memory_snapshot( + self.agent_id, self.memory.dump(), stored_count=self.memory.stored_count, + expected_revision=expected, + ) + def get_state(self) -> AgentState: - return self.state + state = deepcopy(self.state) + if self.memory is not None: + state.memory_entries = self.memory.dump() + state.metadata["memory_stored_count"] = self.memory.stored_count + return state def load_state(self, state: AgentState) -> None: - self.state = state - if self.memory and state.memory_entries: - self.memory.load(state.memory_entries) + self.state = deepcopy(state) + self.agent_id = state.agent_id + self._memory_revision = None + if self.memory is not None: + self.memory.load(state.memory_entries, + stored_count=state.metadata.get("memory_stored_count", 0)) def __repr__(self) -> str: return f"BaseAgent(id={self.agent_id}, model={self.model})" diff --git a/src/self_evolving/core/environment.py b/src/self_evolving/core/environment.py index a9722b9..736a04f 100644 --- a/src/self_evolving/core/environment.py +++ b/src/self_evolving/core/environment.py @@ -53,6 +53,12 @@ class SimpleQAEnvironment(Environment): def __init__(self, qa_pairs: list[tuple[str, str]]): super().__init__("simple_qa") + if not qa_pairs or any(not question.strip() or not answer.strip() + for question, answer in qa_pairs): + raise ValueError("QA pairs require nonempty questions and reference answers") + questions = [question.strip().casefold() for question, _ in qa_pairs] + if len(set(questions)) != len(questions): + raise ValueError("QA questions must be unique") self._qa_pairs = qa_pairs self._current_answer: str = "" self._done: bool = False @@ -65,9 +71,13 @@ def reset(self, goal: str) -> str: if question.strip().lower() == goal.strip().lower(): self._current_answer = answer break + else: + raise ValueError(f"No reference answer for goal: {goal!r}") return f"Question: {goal}" def step(self, action: str) -> tuple[str, Feedback, bool]: + if not self._current_answer: + raise RuntimeError("reset must select a valid QA goal before step") self._step_count += 1 correct = self._current_answer.lower() in action.lower() self._done = True diff --git a/src/self_evolving/dashboard/demo_data.py b/src/self_evolving/dashboard/demo_data.py index a9435eb..a0d2a52 100644 --- a/src/self_evolving/dashboard/demo_data.py +++ b/src/self_evolving/dashboard/demo_data.py @@ -1,20 +1,21 @@ -"""Offline demo data generation for the dashboard.""" +"""Deterministic synthetic dashboard data; no providers or global patches.""" from __future__ import annotations -from contextlib import contextmanager from pathlib import Path -from typing import Iterator from self_evolving.core.agent import BaseAgent from self_evolving.core.environment import SimpleQAEnvironment from self_evolving.evaluation.benchmark import BenchmarkRunner, BenchmarkTask from self_evolving.evolution.memory.episodic import EpisodicMemory +from self_evolving.evolution.memory.embedders import HashingEmbedder from self_evolving.evolution.prompt.opro import OPROOptimizer +from self_evolving.mechanisms.reflection.reflexion import ReflexionReflector from self_evolving.persistence.sqlite_store import SQLiteStore def _demo_answer(question: str) -> str: + question = question.rsplit("[Current observation]:", 1)[-1] q = question.lower() if "capital of france" in q: return "ANSWER: Paris" @@ -29,34 +30,44 @@ def _demo_answer(question: str) -> str: return "ANSWER: unknown" -@contextmanager -def patched_demo_llm() -> Iterator[None]: - original_call = BaseAgent._call_llm - original_optimize = OPROOptimizer.optimize +class DemoAgent(BaseAgent): + def __init__(self, **kwargs): + kwargs["model"] = "offline/deterministic-fixture" + super().__init__(**kwargs) - def fake_call(self, messages): + def _call_llm(self, messages): return _demo_answer(messages[-1]["content"]) - def fake_optimize(self, initial_prompt, eval_fn, task_description="general agent task"): - self._history = [ - (initial_prompt, eval_fn(initial_prompt)), - (initial_prompt + " [optimized]", eval_fn(initial_prompt)), - ] - return initial_prompt + " [optimized]" - BaseAgent._call_llm = fake_call - OPROOptimizer.optimize = fake_optimize - try: - yield - finally: - BaseAgent._call_llm = original_call - OPROOptimizer.optimize = original_optimize +class DemoMemory(EpisodicMemory): + def __init__(self, **kwargs): + kwargs["embedder"] = HashingEmbedder() + super().__init__(**kwargs) + + def _distil(self, trajectory): + return [f"Synthetic lesson for {trajectory.goal}: inspect the current question."] + + def _summarize(self, combined): + return f"Synthetic summary: {combined[:120]}" + + +class DemoReflector(ReflexionReflector): + def reflect(self, trajectory): + if not trajectory.success: + trajectory.metadata["reflection"] = "Synthetic reflection: inspect the question." + return trajectory + + +class DemoOptimizer(OPROOptimizer): + def _propose(self, task_description): + return BaseAgent.DEFAULT_SYSTEM + " Check the current question carefully." def generate_demo_data( db_path: str = ".data/sea.db", benchmark_dir: str = "runs/benchmarks", ) -> dict: + """Run actual control flow with canned local answers, not model-quality scores.""" Path(db_path).parent.mkdir(parents=True, exist_ok=True) Path(benchmark_dir).mkdir(parents=True, exist_ok=True) @@ -67,22 +78,28 @@ def generate_demo_data( BenchmarkTask("What planet is known as the Red Planet?", "Mars"), ] - with patched_demo_llm(): - store = SQLiteStore(db_path) - agent = BaseAgent(agent_id="demo-agent") - agent.store = store - agent.memory = EpisodicMemory() - env = SimpleQAEnvironment([(task.goal, task.reference_answer) for task in tasks]) - - generated_runs = [] - for index, task in enumerate(tasks): - trajectory = agent.run(env, goal=task.goal, task_id=f"demo_task_{index}") - generated_runs.append(trajectory.metadata.get("run_id")) - - runner = BenchmarkRunner(tasks, output_dir=benchmark_dir) - benchmark_summary = runner.run() + store = SQLiteStore(db_path) + agent = DemoAgent(agent_id="demo-agent") + agent.store = store + agent.memory = DemoMemory() + env = SimpleQAEnvironment([(task.goal, task.reference_answer) for task in tasks]) + + generated_runs = [] + for index, task in enumerate(tasks): + trajectory = agent.run(env, goal=task.goal, task_id=f"demo_task_{index}") + generated_runs.append(trajectory.metadata.get("run_id")) + + runner = BenchmarkRunner( + tasks, output_dir=benchmark_dir, + tuning_tasks=[BenchmarkTask("Which element has atomic number 79?", "Gold")], + agent_factory=DemoAgent, memory_factory=DemoMemory, + reflector_factory=DemoReflector, optimizer_factory=DemoOptimizer, + data_source="synthetic_deterministic_fixture", + ) + benchmark_summary = runner.run() return { + "data_source": "synthetic_deterministic_fixture", "db_path": db_path, "benchmark_dir": benchmark_dir, "generated_runs": generated_runs, diff --git a/src/self_evolving/evaluation/benchmark.py b/src/self_evolving/evaluation/benchmark.py index 8d63d30..cf7b7ff 100644 --- a/src/self_evolving/evaluation/benchmark.py +++ b/src/self_evolving/evaluation/benchmark.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import uuid from dataclasses import asdict, dataclass, field from datetime import UTC, datetime from pathlib import Path @@ -49,11 +50,33 @@ def __init__( max_steps: int = 20, store: Optional[SQLiteStore] = None, agent_id_prefix: str = "benchmark", + *, + tuning_tasks: Optional[Iterable[BenchmarkTask | tuple[str, str]]] = None, + agent_factory: Callable[..., BaseAgent] = BaseAgent, + memory_factory: Callable[..., EpisodicMemory] = EpisodicMemory, + reflector_factory: Callable[..., ReflexionReflector] = ReflexionReflector, + optimizer_factory: Callable[..., OPROOptimizer] = OPROOptimizer, + data_source: str = "model_execution", ): self.tasks = [ task if isinstance(task, BenchmarkTask) else BenchmarkTask(*task) for task in tasks ] + self.tuning_tasks = [task if isinstance(task, BenchmarkTask) else BenchmarkTask(*task) + for task in tuning_tasks] if tuning_tasks is not None else None + # Validate before making model calls or writing artifacts. + SimpleQAEnvironment([(task.goal, task.reference_answer) for task in self.tasks]) + if self.tuning_tasks is not None: + SimpleQAEnvironment([(task.goal, task.reference_answer) for task in self.tuning_tasks]) + tuning_goals = {task.goal.strip().casefold() for task in self.tuning_tasks} + if tuning_goals & {task.goal.strip().casefold() for task in self.tasks}: + raise ValueError("tuning_tasks and evaluation tasks must have disjoint goals") + self.agent_factory = agent_factory + self.memory_factory = memory_factory + self.reflector_factory = reflector_factory + self.optimizer_factory = optimizer_factory + self.data_source = data_source + self._session_id = "" self.output_dir = Path(output_dir) self.output_dir.mkdir(parents=True, exist_ok=True) self.model = model @@ -67,6 +90,9 @@ def run( progress_callback: Optional[Callable[[float, str, dict[str, Any]], None]] = None, ) -> dict: variants = variants or list(self.DEFAULT_VARIANTS) + unknown = set(variants) - set(self.DEFAULT_VARIANTS) + if unknown: + raise ValueError(f"Unsupported benchmark variants: {sorted(unknown)}") ordered_variants = ["baseline"] + [variant for variant in variants if variant != "baseline"] session_dir = self._make_session_dir() if progress_callback: @@ -110,6 +136,13 @@ def callback(progress: float, stage: str, detail: dict[str, Any]) -> None: "generated_at": datetime.now(UTC).isoformat(), "session_dir": str(session_dir), "task_count": len(self.tasks), + "data_source": self.data_source, + "evaluation_protocol": self._evaluation_protocol(), + "task_manifest": { + "evaluation": [asdict(task) for task in self.tasks], + "tuning": [asdict(task) for task in self.tuning_tasks] + if self.tuning_tasks is not None else None, + }, "variants": {name: asdict(result) for name, result in results.items()}, } self._write_summary_artifact(session_dir, summary) @@ -155,7 +188,7 @@ def _run_memory( progress_callback: Optional[Callable[[float, str, dict[str, Any]], None]] = None, ) -> VariantResult: agent = self._make_agent("memory") - agent.memory = EpisodicMemory() + agent.memory = self.memory_factory(model=agent.model) metrics = EvolutionMetrics(baseline_success_rate=baseline_success_rate) episodes = self._run_tasks_with_agent( agent, @@ -164,7 +197,10 @@ def _run_memory( progress_callback=progress_callback, ) report = metrics.report() - return self._build_result("memory", report, episodes) + return self._build_result("memory", report, episodes, metadata={ + "protocol": "sequential_online_adaptation", + "initial_memory": "empty", "memory_updates_during_evaluation": True, + }) def _run_reflexion( self, @@ -172,7 +208,7 @@ def _run_reflexion( progress_callback: Optional[Callable[[float, str, dict[str, Any]], None]] = None, ) -> VariantResult: agent = self._make_agent("reflexion") - wrapped = ReflexionAgent(agent, ReflexionReflector(model=agent.model, max_rounds=2)) + wrapped = ReflexionAgent(agent, self.reflector_factory(model=agent.model, max_rounds=2)) metrics = EvolutionMetrics(baseline_success_rate=baseline_success_rate) episodes = self._run_tasks_with_runner( wrapped, @@ -181,7 +217,12 @@ def _run_reflexion( progress_callback=progress_callback, ) report = metrics.report() - return self._build_result("reflexion", report, episodes) + return self._build_result("reflexion", report, episodes, metadata={ + "protocol": "same_task_retry", "max_attempts": 2, + "mean_steps_scope": "final_attempt_only", + "total_attempt_steps": sum(e["metadata"].get("total_attempt_steps", e["num_steps"]) + for e in episodes), + }) def _run_prompt_optimization( self, @@ -191,7 +232,9 @@ def _run_prompt_optimization( initial_prompt = BaseAgent.DEFAULT_SYSTEM if progress_callback: progress_callback(5.0, "optimizing_prompt", {"variant": "prompt_optimization"}) - optimizer = OPROOptimizer(model=self.model, max_iterations=3, batch_size=min(4, len(self.tasks))) + tuning_tasks = self.tuning_tasks if self.tuning_tasks is not None else self.tasks + optimizer = self.optimizer_factory(model=self.model, max_iterations=3, + batch_size=min(4, len(tuning_tasks))) eval_fn = self._make_eval_fn() best_prompt = optimizer.optimize( initial_prompt=initial_prompt, @@ -221,6 +264,7 @@ def on_task_progress(progress: float, stage: str, detail: dict[str, Any]) -> Non report, episodes, metadata={ + "evaluation_protocol": self._evaluation_protocol(), "best_prompt": best_prompt, "history": [ {"prompt": prompt, "score": score} @@ -230,14 +274,15 @@ def on_task_progress(progress: float, stage: str, detail: dict[str, Any]) -> Non ) def _make_eval_fn(self): - tasks = list(self.tasks) + tasks = list(self.tuning_tasks if self.tuning_tasks is not None else self.tasks) def eval_fn(prompt: str) -> float: - agent = BaseAgent(model=self.model, max_steps=self.max_steps, system_prompt=prompt) + agent = self.agent_factory(model=self.model, max_steps=self.max_steps, + system_prompt=prompt) env = SimpleQAEnvironment([(task.goal, task.reference_answer) for task in tasks]) successes = 0 for index, task in enumerate(tasks): - trajectory = agent.run(env, goal=task.goal, task_id=f"opro_eval_{index}") + trajectory = agent.run(env, goal=task.goal, task_id=f"opro_tuning_{index}") if trajectory.success: successes += 1 return successes / len(tasks) if tasks else 0.0 @@ -245,11 +290,11 @@ def eval_fn(prompt: str) -> float: return eval_fn def _make_agent(self, variant: str, system_prompt: Optional[str] = None) -> BaseAgent: - agent = BaseAgent( + agent = self.agent_factory( model=self.model, max_steps=self.max_steps, system_prompt=system_prompt, - agent_id=f"{self.agent_id_prefix}-{variant}", + agent_id=f"{self.agent_id_prefix}-{self._session_id}-{variant}", ) agent.store = self.store return agent @@ -353,10 +398,22 @@ def _build_result(name: str, report, episodes: list[dict], metadata: Optional[di def _make_session_dir(self) -> Path: stamp = datetime.now(UTC).strftime("%Y%m%d-%H%M%S") - session_dir = self.output_dir / stamp - session_dir.mkdir(parents=True, exist_ok=True) + self._session_id = f"{stamp}-{uuid.uuid4().hex}" + session_dir = self.output_dir / self._session_id + session_dir.mkdir(parents=True, exist_ok=False) return session_dir + def _evaluation_protocol(self) -> dict[str, Any]: + return { + "prompt_optimization": "heldout" if self.tuning_tasks is not None else "resubstitution", + "tuning_task_count": len(self.tuning_tasks) if self.tuning_tasks is not None else len(self.tasks), + "evaluation_task_count": len(self.tasks), + "overlap_check": "normalized_exact_goal; semantic overlap requires dataset review", + "scoring": "case_insensitive_reference_substring_smoke_test", + "memory": "sequential_online_adaptation", + "reflexion": "same_task_retry_up_to_2_attempts", + } + @staticmethod def _write_variant_artifact(session_dir: Path, result: VariantResult) -> None: path = session_dir / f"{result.name}.json" diff --git a/src/self_evolving/evolution/memory/episodic.py b/src/self_evolving/evolution/memory/episodic.py index 88adc0b..eaa499a 100644 --- a/src/self_evolving/evolution/memory/episodic.py +++ b/src/self_evolving/evolution/memory/episodic.py @@ -9,7 +9,7 @@ import logging import os -from dataclasses import dataclass, field +from dataclasses import asdict, dataclass, field from typing import Optional import litellm @@ -62,6 +62,8 @@ def __init__( embedder: Optional[BaseEmbedder] = None, lexical_weight: float = 0.15, ): + if max_entries < 1 or summarize_after < 1: + raise ValueError("max_entries and summarize_after must be positive") self.model = model or os.getenv("SEA_WEAK_MODEL", "deepseek/deepseek-chat") self.max_entries = max_entries self.summarize_after = summarize_after @@ -92,7 +94,7 @@ def store(self, trajectory: Trajectory) -> list[MemoryEntry]: def retrieve(self, query: str, top_k: int = 3) -> list[str]: """Return top-k relevant memory entries for a query.""" - if not self._entries: + if not self._entries or top_k <= 0: return [] query_embedding = self.embedder.embed(query) @@ -113,18 +115,28 @@ def retrieve(self, query: str, top_k: int = 3) -> list[str]: results.append(entry.content) return results - def load(self, raw_entries: list[dict]) -> None: + def load(self, raw_entries: list[dict], *, stored_count: int = 0) -> None: + """Load entries in oldest-first order, preserving the summary cadence.""" + if stored_count < 0: + raise ValueError("stored_count must be nonnegative") self._entries = [MemoryEntry(**entry) for entry in raw_entries] + self._stored_count = stored_count + self._trim() def dump(self) -> list[dict]: - return [entry.__dict__ for entry in self._entries] + return [asdict(entry) for entry in self._entries] + + @property + def stored_count(self) -> int: + return self._stored_count def __len__(self) -> int: return len(self._entries) def _distil(self, trajectory: Trajectory) -> list[str]: steps_summary = "\n".join( - f" Step {step.step_index}: action={step.action[:120]!r}" + f" Step {step.step_index}: action={step.action[:120]!r}; " + f"feedback={str(step.feedback.value)[:120] if step.feedback else ''!r}" for step in trajectory.steps[:6] ) outcome = "SUCCESS" if trajectory.success else "FAILURE" @@ -133,6 +145,9 @@ def _distil(self, trajectory: Trajectory) -> list[str]: outcome=outcome, steps=steps_summary, ) + reflection = trajectory.metadata.get("reflection") + if reflection: + prompt += f"\nPost-attempt reflection (unverified): {str(reflection)[:600]}" try: resp = litellm.completion( model=self.model, @@ -144,7 +159,7 @@ def _distil(self, trajectory: Trajectory) -> list[str]: return [ line.replace("LESSON:", "").strip() for line in raw.splitlines() - if line.strip().startswith("LESSON:") + if line.strip().startswith("LESSON:") and line.split("LESSON:", 1)[1].strip() ] except Exception as exc: logger.warning(f"Memory distillation failed: {exc}") @@ -159,8 +174,19 @@ def _maybe_summarize(self) -> None: n_old = max(1, len(self._entries) // 4) old_entries = self._entries[:n_old] - self._entries = self._entries[n_old:] combined = " | ".join(entry.content for entry in old_entries) + summary = self._summarize(combined) + entry = MemoryEntry( + content=f"[Summary] {summary}", + source_task="summarized", + success=all(item.success for item in old_entries), + importance=1.2, + access_count=sum(item.access_count for item in old_entries), + embedding=self.embedder.embed(summary), + ) + self._entries = [entry] + self._entries[n_old:] + + def _summarize(self, combined: str) -> str: summary_prompt = f"Compress these agent lessons into 2 concise sentences:\n{combined}" try: @@ -174,21 +200,15 @@ def _maybe_summarize(self) -> None: except Exception: summary = combined[:200] - self._entries.insert( - 0, - MemoryEntry( - content=f"[Summary] {summary}", - source_task="summarized", - success=True, - importance=1.2, - embedding=self.embedder.embed(summary), - ), - ) + return summary def _trim(self) -> None: if len(self._entries) > self.max_entries: - self._entries.sort(key=lambda entry: entry.importance, reverse=True) - self._entries = self._entries[: self.max_entries] + # Select by importance without changing chronological order. + keep = set(sorted(range(len(self._entries)), + key=lambda i: self._entries[i].importance, + reverse=True)[:self.max_entries]) + self._entries = [entry for i, entry in enumerate(self._entries) if i in keep] @staticmethod def _lexical_overlap(query_words: set[str], content: str) -> float: diff --git a/src/self_evolving/evolution/prompt/opro.py b/src/self_evolving/evolution/prompt/opro.py index ed23c26..2b359be 100644 --- a/src/self_evolving/evolution/prompt/opro.py +++ b/src/self_evolving/evolution/prompt/opro.py @@ -59,6 +59,8 @@ def __init__( max_iterations: int = 5, batch_size: int = 4, ): + if max_iterations < 0 or batch_size < 1: + raise ValueError("max_iterations must be nonnegative and batch_size positive") self.model = model or os.getenv("SEA_MODEL", "deepseek/deepseek-chat") self.max_iterations = max_iterations self.batch_size = batch_size @@ -74,6 +76,8 @@ def optimize( Run OPRO optimisation loop. Returns the best prompt found. """ + # Each optimization run is independent; old task scores must not leak. + self._history = [] current_prompt = initial_prompt current_score = eval_fn(current_prompt) self._history.append((current_prompt, current_score)) diff --git a/src/self_evolving/mechanisms/reflection/reflexion.py b/src/self_evolving/mechanisms/reflection/reflexion.py index 03cf287..07a6704 100644 --- a/src/self_evolving/mechanisms/reflection/reflexion.py +++ b/src/self_evolving/mechanisms/reflection/reflexion.py @@ -53,6 +53,8 @@ class ReflexionReflector(BaseReflector): """ def __init__(self, model: Optional[str] = None, max_rounds: int = 3): + if max_rounds < 1: + raise ValueError("max_rounds must be positive") self.model = model or os.getenv("SEA_WEAK_MODEL", "deepseek/deepseek-chat") self.max_rounds = max_rounds @@ -74,7 +76,7 @@ def reflect(self, trajectory: Trajectory) -> Trajectory: temperature=0.5, max_tokens=256, ) - reflection = resp.choices[0].message.content.strip() + reflection = (resp.choices[0].message.content or "").strip() trajectory.metadata["reflection"] = reflection logger.info(f"Reflexion: {reflection[:100]}") except Exception as e: @@ -100,7 +102,6 @@ def __init__(self, agent, reflector: Optional[ReflexionReflector] = None): from self_evolving.core.agent import BaseAgent self.agent: BaseAgent = agent self.reflector = reflector or ReflexionReflector(model=agent.model) - self.agent.reflector = self.reflector def run( self, @@ -109,35 +110,44 @@ def run( task_id: Optional[str] = None, progress_callback: Optional[Callable[[float, str, dict[str, Any]], None]] = None, ): - from self_evolving.core.environment import Environment original_prompt = self.agent.state.system_prompt - - for attempt in range(self.reflector.max_rounds): - def on_progress(progress: float, stage: str, detail: dict[str, Any]) -> None: - if progress_callback is None: - return - scaled = ((attempt + (progress / 100.0)) / self.reflector.max_rounds) * 100.0 - progress_callback( - scaled, - stage, - { - **detail, - "attempt": attempt + 1, - "max_attempts": self.reflector.max_rounds, - }, - ) - - trajectory = self.agent.run(env, goal, task_id, progress_callback=on_progress) - if trajectory.success: - self.agent.state.system_prompt = original_prompt - return trajectory - - reflection = trajectory.metadata.get("reflection", "") - if reflection: - self.agent.state.system_prompt = ( - original_prompt + f"\n\n[Reflection from attempt {attempt+1}]: {reflection}" - ) - logger.info(f"Reflexion retry {attempt+2}/{self.reflector.max_rounds}") - - self.agent.state.system_prompt = original_prompt - return trajectory + original_reflector = self.agent.reflector + self.agent.reflector = self.reflector + attempts = [] + try: + for attempt in range(self.reflector.max_rounds): + def on_progress(progress: float, stage: str, detail: dict[str, Any]) -> None: + if progress_callback is None: + return + scaled = ((attempt + (progress / 100.0)) / self.reflector.max_rounds) * 100.0 + progress_callback(scaled, "attempt_completed" if stage == "completed" else stage, + {**detail, "attempt": attempt + 1, + "max_attempts": self.reflector.max_rounds}) + + trajectory = self.agent.run(env, goal, task_id, progress_callback=on_progress) + attempts.append({"attempt": attempt + 1, "success": trajectory.success, + "num_steps": len(trajectory.steps), + "total_reward": trajectory.total_reward, + "run_id": trajectory.metadata.get("run_id"), + "reflection": trajectory.metadata.get("reflection")}) + if trajectory.success: + break + reflection = trajectory.metadata.get("reflection", "") + if reflection: + self.agent.state.system_prompt = ( + original_prompt + f"\n\n[Reflection from attempt {attempt+1}]: {reflection}" + ) + if attempt + 1 < self.reflector.max_rounds: + logger.info(f"Reflexion retry {attempt+2}/{self.reflector.max_rounds}") + trajectory.metadata["reflexion_attempts"] = attempts + trajectory.metadata["attempt_count"] = len(attempts) + trajectory.metadata["total_attempt_steps"] = sum(a["num_steps"] for a in attempts) + if self.agent.store is not None and trajectory.metadata.get("run_id"): + self.agent.store.update_run_metadata(trajectory.metadata["run_id"], trajectory.metadata) + if progress_callback: + progress_callback(100.0, "completed", {"attempt_count": len(attempts)}) + return trajectory + finally: + # Also restore state if execution, persistence, or callbacks fail. + self.agent.state.system_prompt = original_prompt + self.agent.reflector = original_reflector diff --git a/src/self_evolving/persistence/sqlite_store.py b/src/self_evolving/persistence/sqlite_store.py index 1536d93..46dca77 100644 --- a/src/self_evolving/persistence/sqlite_store.py +++ b/src/self_evolving/persistence/sqlite_store.py @@ -8,7 +8,12 @@ import time import uuid from pathlib import Path -from typing import Any +from contextlib import contextmanager +from typing import Any, Iterator + + +class MemoryConflictError(RuntimeError): + """A stale agent tried to replace a newer memory snapshot.""" class SQLiteStore: @@ -20,10 +25,15 @@ def __init__(self, db_path: str | None = None): path.parent.mkdir(parents=True, exist_ok=True) self._init_db() - def _connect(self) -> sqlite3.Connection: + @contextmanager + def _connect(self) -> Iterator[sqlite3.Connection]: conn = sqlite3.connect(self.db_path) conn.row_factory = sqlite3.Row - return conn + try: + with conn: + yield conn + finally: + conn.close() def _init_db(self) -> None: with self._connect() as conn: @@ -75,6 +85,12 @@ def _init_db(self) -> None: CREATE INDEX IF NOT EXISTS idx_memories_agent_id ON memories (agent_id, created_at DESC); + + CREATE TABLE IF NOT EXISTS memory_state ( + agent_id TEXT PRIMARY KEY, + stored_count INTEGER NOT NULL DEFAULT 0, + revision INTEGER NOT NULL DEFAULT 0 + ); """ ) columns = { @@ -149,9 +165,11 @@ def save_trajectory(self, *, agent: Any, env_name: str, trajectory: Any) -> str: return run_id def save_memory_entries(self, agent_id: str, entries: list[Any]) -> None: + """Legacy append interface; agent checkpoints use save_memory_snapshot.""" if not entries: return with self._connect() as conn: + conn.execute("BEGIN IMMEDIATE") conn.executemany( """ INSERT INTO memories ( @@ -173,6 +191,67 @@ def save_memory_entries(self, agent_id: str, entries: list[Any]) -> None: for entry in entries ], ) + conn.execute( + "INSERT INTO memory_state (agent_id, revision) VALUES (?, 1) " + "ON CONFLICT(agent_id) DO UPDATE SET revision = revision + 1", + (agent_id,), + ) + + def save_memory_snapshot( + self, agent_id: str, entries: list[dict[str, Any]], *, + stored_count: int, expected_revision: int, + ) -> int: + """Atomically replace one agent's entries and lifecycle counters. + + Optimistic versioning rejects stale writers instead of silently losing + another run's memories. A caller must reload and rerun after a conflict. + """ + if stored_count < 0: + raise ValueError("stored_count must be nonnegative") + with self._connect() as conn: + conn.execute("BEGIN IMMEDIATE") + state = conn.execute( + "SELECT revision FROM memory_state WHERE agent_id = ?", (agent_id,) + ).fetchone() + revision = state["revision"] if state else 0 + if revision != expected_revision: + raise MemoryConflictError( + f"Memory for {agent_id!r} changed: expected revision " + f"{expected_revision}, found {revision}; reload and rerun" + ) + conn.execute("DELETE FROM memories WHERE agent_id = ?", (agent_id,)) + conn.executemany( + "INSERT INTO memories (agent_id, source_task, content, success, " + "importance, access_count, embedding_json, created_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + [(agent_id, item["source_task"], item["content"], int(item["success"]), + float(item["importance"]), int(item["access_count"]), + self._json_dump(item.get("embedding", [])), time.time()) for item in entries], + ) + conn.execute( + "INSERT INTO memory_state (agent_id, stored_count, revision) VALUES (?, ?, ?) " + "ON CONFLICT(agent_id) DO UPDATE SET " + "stored_count = excluded.stored_count, revision = excluded.revision", + (agent_id, stored_count, revision + 1), + ) + return revision + 1 + + def load_memory_snapshot(self, agent_id: str) -> dict[str, Any]: + """Read one consistent, oldest-first snapshot (including legacy rows).""" + with self._connect() as conn: + conn.execute("BEGIN") + state = conn.execute( + "SELECT stored_count, revision FROM memory_state WHERE agent_id = ?", (agent_id,) + ).fetchone() + rows = conn.execute( + "SELECT source_task, content, success, importance, access_count, embedding_json " + "FROM memories WHERE agent_id = ? ORDER BY id ASC", (agent_id,), + ).fetchall() + return { + "entries": self._memory_rows(rows), + "stored_count": state["stored_count"] if state else 0, + "revision": state["revision"] if state else 0, + } def list_runs(self, limit: int = 50) -> list[dict[str, Any]]: with self._connect() as conn: @@ -181,7 +260,7 @@ def list_runs(self, limit: int = 50) -> list[dict[str, Any]]: SELECT run_id, task_id, agent_id, env_name, goal, model, success, total_reward, num_steps, created_at FROM runs - ORDER BY created_at DESC + ORDER BY created_at DESC, run_id DESC LIMIT ? """, (limit,), @@ -220,6 +299,14 @@ def get_run(self, run_id: str) -> dict[str, Any] | None: run["steps"].append(item) return run + def update_run_metadata(self, run_id: str, metadata: dict[str, Any]) -> None: + """Persist post-run wrapper evidence, such as Reflexion attempt links.""" + with self._connect() as conn: + result = conn.execute("UPDATE runs SET metadata_json = ? WHERE run_id = ?", + (self._json_dump(metadata), run_id)) + if result.rowcount != 1: + raise KeyError(f"Run not found: {run_id}") + def list_memory(self, agent_id: str, limit: int = 100) -> list[dict[str, Any]]: with self._connect() as conn: rows = conn.execute( @@ -227,11 +314,14 @@ def list_memory(self, agent_id: str, limit: int = 100) -> list[dict[str, Any]]: SELECT source_task, content, success, importance, access_count, embedding_json FROM memories WHERE agent_id = ? - ORDER BY created_at DESC + ORDER BY created_at DESC, id DESC LIMIT ? """, (agent_id, limit), ).fetchall() + return self._memory_rows(rows) + + def _memory_rows(self, rows) -> list[dict[str, Any]]: result = [] for row in rows: item = dict(row) diff --git a/src/self_evolving/service/api.py b/src/self_evolving/service/api.py index c0fb999..3c6f692 100644 --- a/src/self_evolving/service/api.py +++ b/src/self_evolving/service/api.py @@ -46,6 +46,7 @@ class BenchmarkTaskRequest(BaseModel): class BenchmarkRequest(BaseModel): tasks: list[BenchmarkTaskRequest] = Field(..., min_length=1) + tuning_tasks: Optional[list[BenchmarkTaskRequest]] = Field(default=None, min_length=1) variants: list[str] = Field(default_factory=lambda: list(BenchmarkRunner.DEFAULT_VARIANTS)) model: Optional[str] = None max_steps: int = Field(default=20, ge=1, le=100) @@ -75,8 +76,8 @@ def _build_agent(request: QARunRequest, store: SQLiteStore) -> BaseAgent: if request.use_memory: memory = EpisodicMemory() - memory.load(store.list_memory(agent.agent_id)) agent.memory = memory + agent.load_memory() return agent @@ -188,6 +189,9 @@ def job_fn(progress_callback): model=request.model, max_steps=request.max_steps, store=app.state.store, + tuning_tasks=[BenchmarkTask(task.goal, task.reference_answer) + for task in request.tuning_tasks] + if request.tuning_tasks is not None else None, ) summary = runner.run( variants=request.variants, @@ -197,6 +201,8 @@ def job_fn(progress_callback): "session_dir": summary["session_dir"], "task_count": summary["task_count"], "variants": summary["variants"], + "evaluation_protocol": summary["evaluation_protocol"], + "data_source": summary["data_source"], } job = jobs.submit( diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..4e98327 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,51 @@ +"""All tests are offline: providers and socket connections need explicit mocks.""" +import os +import socket +import threading + +# Prevent LiteLLM's import-time refresh of its public price map. +os.environ["LITELLM_LOCAL_MODEL_COST_MAP"] = "True" +os.environ["DO_NOT_TRACK"] = "1" + +import litellm +import pytest + + +@pytest.fixture(autouse=True) +def no_external_calls(monkeypatch): + attempted = [] + pair_context = threading.local() + original_pair = socket.socketpair + original_connect = socket.socket.connect + + def offline_socketpair(*args, **kwargs): + # Windows implements asyncio's self-pipe with a local socket pair. + # Only that construction may connect; ordinary localhost HTTP is blocked. + pair_context.active = True + try: + return original_pair(*args, **kwargs) + finally: + pair_context.active = False + + def guarded_connect(sock, address): + if getattr(pair_context, "active", False) and address[0] in {"127.0.0.1", "::1"}: + return original_connect(sock, address) + return blocked_network() + + def blocked_network(*args, **kwargs): + attempted.append("network") + raise AssertionError("Network calls are forbidden in offline tests") + + def blocked_provider(*args, **kwargs): + attempted.append("provider") + raise AssertionError("Mock the model boundary explicitly") + + monkeypatch.setattr(socket, "socketpair", offline_socketpair) + monkeypatch.setattr(socket.socket, "connect", guarded_connect) + monkeypatch.setattr(socket.socket, "connect_ex", blocked_network) + monkeypatch.setattr(socket, "create_connection", blocked_network) + monkeypatch.setattr(litellm, "completion", blocked_provider) + monkeypatch.setenv("SEA_MEMORY_EMBEDDER", "hashing") + yield + # Model code can catch exceptions: fail even if the call was swallowed. + assert not attempted, f"Unexpected external calls: {attempted}" diff --git a/tests/test_api.py b/tests/test_api.py index f1870fd..3e91a8c 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -6,6 +6,7 @@ from self_evolving.core.agent import BaseAgent from self_evolving.evolution.prompt.opro import OPROOptimizer +from self_evolving.evolution.memory.episodic import EpisodicMemory from self_evolving.service.api import create_app @@ -58,6 +59,7 @@ def test_run_qa_and_fetch_persisted_data(tmp_path, monkeypatch): def test_run_benchmark_endpoint(tmp_path, monkeypatch): + monkeypatch.setattr(EpisodicMemory, "_distil", lambda self, trajectory: [trajectory.goal]) monkeypatch.setenv("SEA_BENCHMARK_DIR", str(tmp_path / "benchmarks")) def fake_call(self, messages): @@ -98,7 +100,71 @@ def fake_call(self, messages): assert payload["task_count"] == 2 assert "baseline" in payload["variants"] assert "memory" in payload["variants"] + assert payload["evaluation_protocol"]["prompt_optimization"] == "resubstitution" + assert payload["data_source"] == "model_execution" jobs = client.get("/jobs") assert jobs.status_code == 200 assert len(jobs.json()) >= 1 + + +def test_api_rebuilds_agent_with_prior_memory_and_persisted_accesses(tmp_path, monkeypatch): + monkeypatch.setattr(BaseAgent, "_call_llm", lambda self, messages: "ANSWER: Paris") + monkeypatch.setattr(EpisodicMemory, "_distil", lambda self, trajectory: ["France capital Paris"]) + path = str(tmp_path / "sea.db") + # Recreate the app as well as the agent, keeping only the SQLite path and agent ID. + for task_id in ["first", "second"]: + with TestClient(create_app(path)) as client: + response = client.post("/runs/qa", json={"goal": "France capital?", + "reference_answer": "Paris", "agent_id": "stable", + "task_id": task_id, "use_memory": True}) + job = _wait_for_job_completion(client, response.json()["job_id"]) + assert job["status"] == "completed", job.get("error") + detail = client.get(f"/runs/{job['result']['run_id']}").json() + if task_id == "second": + assert "[Past experience 1]: France capital Paris" in detail["steps"][0]["observation"] + entries = client.get("/agents/stable/memory").json() + assert len(entries) == 2 + assert sum(item["access_count"] for item in entries) == 1 + + +def test_api_same_agent_concurrent_run_reports_conflict(tmp_path, monkeypatch): + from threading import Barrier + barrier = Barrier(2) + + def answer(self, messages): + barrier.wait(timeout=5) + return "ANSWER: Paris" + + monkeypatch.setattr(BaseAgent, "_call_llm", answer) + monkeypatch.setattr(EpisodicMemory, "_distil", lambda self, trajectory: [trajectory.task_id]) + with TestClient(create_app(str(tmp_path / "sea.db"))) as client: + jobs = [] + for task_id in ["one", "two"]: + response = client.post("/runs/qa", json={"goal": "France capital?", + "reference_answer": "Paris", "agent_id": "shared", + "task_id": task_id}) + jobs.append(response.json()["job_id"]) + results = [_wait_for_job_completion(client, job_id, timeout=8) for job_id in jobs] + assert sorted(job["status"] for job in results) == ["completed", "failed"] + failed = next(job for job in results if job["status"] == "failed") + assert "reload and rerun" in failed["error"] + entries = client.get("/agents/shared/memory").json() + winner = next(job for job in results if job["status"] == "completed") + assert [entry["content"] for entry in entries] == [winner["result"]["task_id"]] + + +def test_api_accepts_disjoint_tuning_tasks_and_keeps_protocol(tmp_path, monkeypatch): + monkeypatch.setenv("SEA_BENCHMARK_DIR", str(tmp_path / "benchmarks")) + monkeypatch.setattr(BaseAgent, "_call_llm", lambda self, messages: "ANSWER: correct") + monkeypatch.setattr(OPROOptimizer, "_propose", lambda self, description: "candidate") + with TestClient(create_app(str(tmp_path / "sea.db"))) as client: + response = client.post("/benchmarks/qa", json={ + "tasks": [{"goal": "evaluation question", "reference_answer": "correct"}], + "tuning_tasks": [{"goal": "tuning question", "reference_answer": "correct"}], + "variants": ["prompt_optimization"], + }) + job = _wait_for_job_completion(client, response.json()["job_id"]) + assert job["status"] == "completed", job.get("error") + assert job["result"]["evaluation_protocol"]["prompt_optimization"] == "heldout" + assert job["result"]["variants"]["prompt_optimization"]["metadata"]["evaluation_protocol"]["tuning_task_count"] == 1 diff --git a/tests/test_benchmark_runner.py b/tests/test_benchmark_runner.py index c220e76..2148ef1 100644 --- a/tests/test_benchmark_runner.py +++ b/tests/test_benchmark_runner.py @@ -5,9 +5,11 @@ from self_evolving.core.agent import BaseAgent from self_evolving.evaluation.benchmark import BenchmarkRunner, BenchmarkTask from self_evolving.evolution.prompt.opro import OPROOptimizer +from self_evolving.evolution.memory.episodic import EpisodicMemory def test_benchmark_runner_writes_artifacts(tmp_path, monkeypatch): + monkeypatch.setattr(EpisodicMemory, "_distil", lambda self, trajectory: [trajectory.goal]) def fake_call_llm(self, messages): prompt = messages[0]["content"] question = messages[-1]["content"] diff --git a/tests/test_demo_data.py b/tests/test_demo_data.py index 8549f96..4a2be86 100644 --- a/tests/test_demo_data.py +++ b/tests/test_demo_data.py @@ -19,3 +19,49 @@ def test_generate_demo_data(tmp_path): session_dirs = list((tmp_path / "benchmarks").iterdir()) assert len(session_dirs) == 1 assert (session_dirs[0] / "summary.json").exists() + assert result["data_source"] == "synthetic_deterministic_fixture" + assert result["benchmark_summary"]["data_source"] == "synthetic_deterministic_fixture" + assert result["benchmark_summary"]["evaluation_protocol"]["prompt_optimization"] == "heldout" + memories = store.list_memory("demo-agent") + assert len(memories) == 4 + assert all("Synthetic lesson" in item["content"] for item in memories) + + +def test_demo_cannot_mutate_live_provider_methods(tmp_path, monkeypatch): + from self_evolving.core.agent import BaseAgent + from self_evolving.evolution.prompt.opro import OPROOptimizer + original_call = BaseAgent._call_llm + original_optimize = OPROOptimizer.optimize + # A user's choice of external embedder must not make the offline demo download a model. + monkeypatch.setenv("SEA_MEMORY_EMBEDDER", "sentence_transformers") + generate_demo_data(str(tmp_path / "demo.db"), str(tmp_path / "runs")) + assert BaseAgent._call_llm is original_call + assert OPROOptimizer.optimize is original_optimize + + +def test_demo_cli_is_offline_even_during_package_import(tmp_path): + import subprocess + import sys + from pathlib import Path + + probe = ''' +import contextlib, io, runpy, sys +def guard(event, args): + if event in {"socket.connect", "socket.getaddrinfo"}: + raise AssertionError("Offline CLI attempted network access") +sys.addaudithook(guard) +sys.argv = ["examples/07_generate_demo_data.py", "--db-path", sys.argv[1], + "--benchmark-dir", sys.argv[2]] +output = io.StringIO() +with contextlib.redirect_stdout(output): + runpy.run_path("examples/07_generate_demo_data.py", run_name="__main__") +assert "synthetic_deterministic_fixture" in output.getvalue() +print("offline CLI completed with network audit guard") +''' + result = subprocess.run( + [sys.executable, "-c", probe, str(tmp_path / "cli.db"), str(tmp_path / "benchmarks")], + cwd=Path(__file__).resolve().parents[1], capture_output=True, text=True, timeout=45, + ) + assert result.returncode == 0, result.stderr + assert "offline CLI completed" in result.stdout + assert (tmp_path / "cli.db").exists() diff --git a/tests/test_evaluation_protocol.py b/tests/test_evaluation_protocol.py new file mode 100644 index 0000000..3ad8a79 --- /dev/null +++ b/tests/test_evaluation_protocol.py @@ -0,0 +1,114 @@ +"""Tuning/evaluation separation, artifacts and score interpretation.""" +import json +from datetime import datetime as RealDatetime +from pathlib import Path + +import pytest + +from self_evolving.core.agent import BaseAgent +from self_evolving.core.environment import SimpleQAEnvironment +from self_evolving.evaluation.benchmark import BenchmarkRunner +from self_evolving.evolution.prompt.opro import OPROOptimizer +from self_evolving.persistence.sqlite_store import SQLiteStore + + +def test_optimizer_only_scores_tuning_tasks_before_heldout_evaluation(tmp_path, monkeypatch): + calls = [] + proposals = [] + + def answer(self, messages): + calls.append((messages[0]["content"], messages[-1]["content"])) + return "ANSWER: correct" if messages[0]["content"] == "candidate" else "wrong" + + def propose(self, task_description): + proposals.append(list(calls)) + return "candidate" + + monkeypatch.setattr(BaseAgent, "_call_llm", answer) + monkeypatch.setattr(OPROOptimizer, "_propose", propose) + runner = BenchmarkRunner(tasks=[("heldout question", "correct")], + tuning_tasks=[("tuning question", "correct")], + output_dir=str(tmp_path)) + summary = runner.run(["prompt_optimization"]) + # The baseline evaluates heldout first, but those scores are never supplied to OPRO. + assert calls[0][1] == "Question: heldout question" + assert all(question == "Question: tuning question" for _, question in calls[1:-1]) + assert calls[-1] == ("candidate", "Question: heldout question") + result = summary["variants"]["prompt_optimization"] + assert result["metadata"]["history"][0]["score"] == 0 + assert result["metadata"]["best_prompt"] == "candidate" + assert summary["evaluation_protocol"]["prompt_optimization"] == "heldout" + assert len(proposals) == 3 + + +def test_default_protocol_never_claims_heldout(tmp_path, monkeypatch): + monkeypatch.setattr(BaseAgent, "_call_llm", lambda self, messages: "ANSWER: correct") + monkeypatch.setattr(OPROOptimizer, "_propose", lambda self, description: "candidate") + summary = BenchmarkRunner([("question", "correct")], output_dir=str(tmp_path)).run( + ["prompt_optimization"]) + protocol = summary["evaluation_protocol"] + assert protocol["prompt_optimization"] == "resubstitution" + assert "substring_smoke_test" in protocol["scoring"] + artifact = json.loads((Path(summary["session_dir"]) / "summary.json").read_text()) + assert artifact["evaluation_protocol"] == protocol + + +@pytest.mark.parametrize("tuning", [[(" QUESTION ", "different reference")], []]) +def test_overlap_or_empty_tuning_rejected(tmp_path, tuning): + with pytest.raises(ValueError): + BenchmarkRunner([("question", "answer")], tuning_tasks=tuning, output_dir=str(tmp_path)) + + +def test_new_optimizer_run_does_not_use_prior_task_history(monkeypatch): + optimizer = OPROOptimizer(max_iterations=1) + histories = [] + + def propose(description): + histories.append(optimizer.history) + return "candidate" + + monkeypatch.setattr(optimizer, "_propose", propose) + optimizer.optimize("first", lambda prompt: 0.25) + optimizer.optimize("second", lambda prompt: 0.5) + assert histories[1] == [("second", 0.5)] + assert all(prompt != "first" for prompt, _ in optimizer.history) + + +def test_same_second_sessions_have_unique_artifacts_and_agents(tmp_path, monkeypatch): + from self_evolving.evaluation import benchmark + + class FrozenDatetime: + @staticmethod + def now(tz): + return RealDatetime(2026, 9, 22, tzinfo=tz) + + monkeypatch.setattr(benchmark, "datetime", FrozenDatetime) + monkeypatch.setattr(BaseAgent, "_call_llm", lambda self, messages: "ANSWER: correct") + store = SQLiteStore(str(tmp_path / "runs.db")) + runner = BenchmarkRunner([("question", "correct")], output_dir=str(tmp_path / "runs"), store=store) + first = runner.run(["baseline"]) + second = runner.run(["baseline"]) + assert first["session_dir"] != second["session_dir"] + assert len({run["agent_id"] for run in store.list_runs()}) == 2 + assert len(list((tmp_path / "runs").glob("*/summary.json"))) == 2 + + +@pytest.mark.parametrize("pairs", [[], [("q", " ")], [("", "a")], [("Q", "a"), (" q ", "b")]]) +def test_qa_rejects_invalid_reference_data(pairs): + with pytest.raises(ValueError): + SimpleQAEnvironment(pairs) + + +def test_qa_missing_goal_and_step_before_reset_cannot_pass(): + env = SimpleQAEnvironment([("q", "a")]) + with pytest.raises(RuntimeError): + env.step("anything") + with pytest.raises(ValueError): + env.reset("missing") + + +def test_qa_substring_scoring_is_explicitly_only_a_smoke_test(): + env = SimpleQAEnvironment([("two plus two", "4")]) + env.reset("two plus two") + # Documents a known limitation; this must not be called exact-match accuracy. + assert env.step("ANSWER: 42")[1].value is True diff --git a/tests/test_memory_lifecycle.py b/tests/test_memory_lifecycle.py new file mode 100644 index 0000000..dec8ccc --- /dev/null +++ b/tests/test_memory_lifecycle.py @@ -0,0 +1,165 @@ +"""Cold start, checkpoint/reload, compaction and stale-writer regressions.""" +from concurrent.futures import ThreadPoolExecutor +from threading import Barrier + +import pytest + +from self_evolving.core.agent import BaseAgent +from self_evolving.core.environment import SimpleQAEnvironment +from self_evolving.core.types import AgentState, Feedback, FeedbackType, Trajectory +from self_evolving.evolution.memory.episodic import EpisodicMemory, MemoryEntry +from self_evolving.persistence.sqlite_store import SQLiteStore, MemoryConflictError + + +class ConstantEmbedder: + def embed(self, text): + return [1.0, 0.0] + + +def make_agent(path, monkeypatch, *, agent_id="same-agent", max_entries=100, summarize_after=10): + agent = BaseAgent(model="test", agent_id=agent_id) + agent.memory = EpisodicMemory(embedder=ConstantEmbedder(), max_entries=max_entries, + summarize_after=summarize_after) + agent.store = SQLiteStore(str(path)) + monkeypatch.setattr(agent.memory, "_distil", lambda trajectory: [f"lesson {trajectory.task_id}"]) + monkeypatch.setattr(agent.memory, "_summarize", lambda combined: f"compressed {combined}") + monkeypatch.setattr(agent, "_call_llm", lambda messages: "ANSWER: Paris") + return agent + + +def run_task(agent, task_id): + return agent.run(SimpleQAEnvironment([("France capital?", "Paris")]), + "France capital?", task_id=task_id) + + +def test_empty_memory_writes_then_new_agent_loads_and_injects(tmp_path, monkeypatch): + path = tmp_path / "memory.db" + first = make_agent(path, monkeypatch) + assert not first.memory + run_task(first, "first") + assert len(first.memory) == 1 + second = make_agent(path, monkeypatch) + trajectory = run_task(second, "second") + assert "[Past experience 1]: lesson first" in trajectory.steps[0].observation + snapshot = SQLiteStore(str(path)).load_memory_snapshot("same-agent") + assert [entry["content"] for entry in snapshot["entries"]] == ["lesson first", "lesson second"] + assert snapshot["entries"][0]["access_count"] == 1 + assert snapshot["stored_count"] == 2 + assert snapshot["revision"] == 2 + isolated = make_agent(path, monkeypatch, agent_id="different-agent") + assert "Past experience" not in run_task(isolated, "third").steps[0].observation + + +def test_summary_cadence_and_contents_survive_new_agent_each_run(tmp_path, monkeypatch): + path = tmp_path / "memory.db" + # Eight entries ensure summarization replaces two entries, not just renames one. + for index in range(8): + agent = make_agent(path, monkeypatch, summarize_after=8) + run_task(agent, str(index)) + snapshot = SQLiteStore(str(path)).load_memory_snapshot("same-agent") + assert snapshot["stored_count"] == 8 + assert len(snapshot["entries"]) == 7 + assert snapshot["entries"][0]["content"] == "[Summary] compressed lesson 0 | lesson 1" + assert all(item["content"] not in {"lesson 0", "lesson 1"} for item in snapshot["entries"]) + restored = make_agent(path, monkeypatch, summarize_after=8) + restored.load_memory() + assert restored.memory.dump() == snapshot["entries"] + assert restored.memory.stored_count == 8 + + +def test_trim_preserves_order_and_does_not_resurrect_evicted_rows(tmp_path, monkeypatch): + path = tmp_path / "memory.db" + agent = make_agent(path, monkeypatch, max_entries=2) + agent.memory.load([ + MemoryEntry("low", "old", False, importance=0.1).__dict__, + MemoryEntry("high", "middle", True, importance=2).__dict__, + ]) + run_task(agent, "new") + restored = make_agent(path, monkeypatch, max_entries=2) + restored.load_memory() + assert [item["content"] for item in restored.memory.dump()] == ["high", "lesson new"] + restored.memory.load([], stored_count=3) + restored.save_memory() + assert restored.store.load_memory_snapshot("same-agent")["entries"] == [] + + +def test_snapshot_rolls_back_failed_replacement(tmp_path): + store = SQLiteStore(str(tmp_path / "memory.db")) + valid = MemoryEntry("kept", "t1", True).__dict__ + store.save_memory_snapshot("a", [valid], stored_count=1, expected_revision=0) + broken = {**valid, "content": None} + with pytest.raises(Exception): + store.save_memory_snapshot("a", [broken], stored_count=2, expected_revision=1) + snapshot = store.load_memory_snapshot("a") + assert snapshot["revision"] == 1 + assert snapshot["entries"][0]["content"] == "kept" + + +def test_concurrent_snapshots_reject_stale_writer_without_losing_winner(tmp_path): + path = str(tmp_path / "memory.db") + stores = [SQLiteStore(path), SQLiteStore(path)] + barrier = Barrier(2) + + def write(index): + revision = stores[index].load_memory_snapshot("a")["revision"] + barrier.wait(timeout=5) + try: + stores[index].save_memory_snapshot("a", [MemoryEntry(str(index), "task", True).__dict__], + stored_count=1, expected_revision=revision) + return "saved", str(index) + except MemoryConflictError: + return "conflict", str(index) + + with ThreadPoolExecutor(max_workers=2) as executor: + results = list(executor.map(write, [0, 1])) + assert sorted(result[0] for result in results) == ["conflict", "saved"] + winner = next(value for status, value in results if status == "saved") + snapshot = stores[0].load_memory_snapshot("a") + assert snapshot["revision"] == 1 + assert snapshot["entries"][0]["content"] == winner + + +def test_legacy_append_invalidates_existing_snapshot_and_remains_readable(tmp_path): + store = SQLiteStore(str(tmp_path / "memory.db")) + store.save_memory_entries("a", [MemoryEntry("first", "task", True)]) + snapshot = store.load_memory_snapshot("a") + assert snapshot["entries"][0]["content"] == "first" + store.save_memory_entries("a", [MemoryEntry("second", "task", True)]) + with pytest.raises(MemoryConflictError): + store.save_memory_snapshot("a", [], stored_count=0, expected_revision=snapshot["revision"]) + assert [item["content"] for item in store.list_memory("a")] == ["second", "first"] + + +def test_state_loads_into_empty_memory_and_dump_does_not_alias(): + agent = BaseAgent(model="test") + agent.memory = EpisodicMemory(embedder=ConstantEmbedder()) + state = AgentState(agent_id="restored", system_prompt="prompt", + memory_entries=[MemoryEntry("kept", "task", True, embedding=[1, 0]).__dict__], + metadata={"memory_stored_count": 6}) + agent.load_state(state) + assert agent.agent_id == "restored" + assert agent.memory.stored_count == 6 + exported = agent.get_state() + exported.memory_entries[0]["embedding"][0] = 999 + assert agent.memory.dump()[0]["embedding"] == [1, 0] + agent.load_state(AgentState(agent_id="restored", system_prompt="empty")) + assert len(agent.memory) == 0 + + +def test_distillation_receives_failure_feedback_and_reflection(monkeypatch): + from types import SimpleNamespace + import litellm + prompts = [] + + def completion(**kwargs): + prompts.append(kwargs["messages"][0]["content"]) + return SimpleNamespace(choices=[SimpleNamespace(message=SimpleNamespace(content="LESSON: check units"))]) + + monkeypatch.setattr(litellm, "completion", completion) + memory = EpisodicMemory(embedder=ConstantEmbedder()) + trajectory = Trajectory(task_id="failed", goal="convert units", + final_feedback=Feedback(FeedbackType.BINARY, False), + metadata={"reflection": "Confused meters with centimeters"}) + memory.store(trajectory) + assert "Confused meters with centimeters" in prompts[0] + assert memory.dump()[0]["success"] is False diff --git a/tests/test_reflexion.py b/tests/test_reflexion.py new file mode 100644 index 0000000..7f7b91b --- /dev/null +++ b/tests/test_reflexion.py @@ -0,0 +1,90 @@ +"""Actual failure/retry behavior and cleanup, without model requests.""" +import pytest + +from self_evolving.core.agent import BaseAgent +from self_evolving.core.environment import SimpleQAEnvironment +from self_evolving.mechanisms.reflection.reflexion import ReflexionAgent, ReflexionReflector + + +class FixedReflector(ReflexionReflector): + def reflect(self, trajectory): + if not trajectory.success: + trajectory.metadata["reflection"] = "Check the capital, not the country." + return trajectory + + +def test_failed_attempt_retries_with_reflection_and_resets_for_next_task(monkeypatch): + agent = BaseAgent(model="test", system_prompt="original") + original_reflector = object() + agent.reflector = original_reflector + wrapper = ReflexionAgent(agent, FixedReflector(max_rounds=3)) + seen = [] + + def answer(messages): + seen.append(messages) + return "ANSWER: Paris" if "Reflection from attempt" in messages[0]["content"] else "ANSWER: wrong" + + monkeypatch.setattr(agent, "_call_llm", answer) + progress = [] + trajectory = wrapper.run(SimpleQAEnvironment([("capital?", "Paris")]), "capital?", + progress_callback=lambda value, stage, detail: progress.append(value)) + assert trajectory.success + assert trajectory.metadata["attempt_count"] == 2 + assert trajectory.metadata["total_attempt_steps"] == 2 + assert "Check the capital" in seen[1][0]["content"] + assert len(seen[1]) == 2 # new episode has no old conversation messages + assert agent.state.system_prompt == "original" + assert agent.reflector is original_reflector + assert progress[-1] == 100 + assert progress == sorted(progress) + wrapper.run(SimpleQAEnvironment([("capital?", "Paris")]), "capital?") + assert seen[2][0]["content"] == "original" + + +def test_exhaustion_is_bounded_and_restores_prompt(monkeypatch): + agent = BaseAgent(model="test", system_prompt="original") + monkeypatch.setattr(agent, "_call_llm", lambda messages: "wrong") + trajectory = ReflexionAgent(agent, FixedReflector(max_rounds=2)).run( + SimpleQAEnvironment([("capital?", "Paris")]), "capital?") + assert not trajectory.success + assert trajectory.metadata["attempt_count"] == 2 + assert agent.state.system_prompt == "original" + assert agent.reflector is None + + +def test_exception_during_retry_restores_prompt_and_reflector(monkeypatch): + agent = BaseAgent(model="test", system_prompt="original") + calls = 0 + + def answer(messages): + nonlocal calls + calls += 1 + if calls == 2: + raise RuntimeError("simulated execution failure") + return "wrong" + + monkeypatch.setattr(agent, "_call_llm", answer) + with pytest.raises(RuntimeError, match="simulated"): + ReflexionAgent(agent, FixedReflector(max_rounds=3)).run( + SimpleQAEnvironment([("capital?", "Paris")]), "capital?") + assert calls == 2 + assert agent.state.system_prompt == "original" + assert agent.reflector is None + + +def test_zero_rounds_rejected(): + with pytest.raises(ValueError, match="max_rounds"): + ReflexionReflector(max_rounds=0) + + +def test_attempt_aggregate_is_saved_in_final_run_metadata(tmp_path, monkeypatch): + from self_evolving.persistence.sqlite_store import SQLiteStore + agent = BaseAgent(model="test") + agent.store = SQLiteStore(str(tmp_path / "runs.db")) + monkeypatch.setattr(agent, "_call_llm", lambda messages: "wrong") + trajectory = ReflexionAgent(agent, FixedReflector(max_rounds=2)).run( + SimpleQAEnvironment([("capital?", "Paris")]), "capital?") + persisted = agent.store.get_run(trajectory.metadata["run_id"]) + assert persisted["metadata"]["attempt_count"] == 2 + assert persisted["metadata"]["total_attempt_steps"] == 2 + assert len({attempt["run_id"] for attempt in persisted["metadata"]["reflexion_attempts"]}) == 2 diff --git a/tests/test_sqlite_store.py b/tests/test_sqlite_store.py index ea69775..38ed096 100644 --- a/tests/test_sqlite_store.py +++ b/tests/test_sqlite_store.py @@ -56,3 +56,38 @@ class Entry: assert memories[0]["content"] == "Always answer with the known fact." assert memories[0]["success"] is True assert memories[0]["embedding"] == [0.1, 0.2, 0.3] + + +def test_legacy_memory_schema_migrates_without_losing_entries(tmp_path): + import sqlite3 + path = str(tmp_path / "legacy.db") + conn = sqlite3.connect(path) + conn.execute("CREATE TABLE memories (id INTEGER PRIMARY KEY AUTOINCREMENT, " + "agent_id TEXT, source_task TEXT, content TEXT, success INTEGER, " + "importance REAL, access_count INTEGER, created_at REAL)") + conn.execute("INSERT INTO memories (agent_id, source_task, content, success, importance, " + "access_count, created_at) VALUES ('a', 'old', 'kept', 1, 1.5, 2, 1)") + conn.commit() + conn.close() + store = SQLiteStore(path) + snapshot = store.load_memory_snapshot("a") + assert snapshot["entries"][0]["content"] == "kept" + assert snapshot["entries"][0]["embedding"] == [] + assert snapshot["stored_count"] == 0 # old schema did not record this counter + store.save_memory_snapshot("a", snapshot["entries"], stored_count=1, expected_revision=0) + assert store.list_memory("a")[0]["access_count"] == 2 + + +def test_connections_close_after_success_and_error(tmp_path): + import sqlite3 + import pytest + store = SQLiteStore(str(tmp_path / "connections.db")) + with store._connect() as connection: + connection.execute("SELECT 1") + with pytest.raises(sqlite3.ProgrammingError, match="closed"): + connection.execute("SELECT 1") + with pytest.raises(RuntimeError): + with store._connect() as failed_connection: + raise RuntimeError("simulate failure") + with pytest.raises(sqlite3.ProgrammingError, match="closed"): + failed_connection.execute("SELECT 1") diff --git a/tests/test_vector_memory.py b/tests/test_vector_memory.py index 2ae200f..a3d90c3 100644 --- a/tests/test_vector_memory.py +++ b/tests/test_vector_memory.py @@ -40,7 +40,8 @@ def test_vector_retrieval_prefers_semantic_match(): assert hits == ["Capital of France is Paris."] -def test_store_adds_embedding(): +def test_store_adds_embedding(monkeypatch): + monkeypatch.setattr(EpisodicMemory, "_distil", lambda self, trajectory: ["Paris is in France"]) class Trajectory: task_id = "task-1" goal = "What is the capital of France?"