from __future__ import annotations from types import SimpleNamespace from opensquilla.bootstrap_types import BootstrapFileReport from opensquilla.engine.runtime import BootstrapSnapshot, MemorySnapshot, TurnRunner from opensquilla.identity.workspace import filter_workspace_filenames_for_session from opensquilla.tools.types import ToolContext def test_bootstrap_write_evicts_only_matching_agent_snapshots() -> None: runner = TurnRunner(provider_selector=None) report = [BootstrapFileReport(filename="USER.md", raw_chars=4, injected_chars=4)] runner._bootstrap_snapshots[("main", "session-a", "full")] = BootstrapSnapshot( workspace_files={"USER.md": "main"}, report=report, ) runner._bootstrap_snapshots[("main", "session-b", "minimal")] = BootstrapSnapshot( workspace_files={"USER.md": "main"}, report=report, ) runner._bootstrap_snapshots[("other", "session-a", "full")] = BootstrapSnapshot( workspace_files={"USER.md": "other"}, report=report, ) runner._memory_snapshots[("main", "session-a")] = object() # type: ignore[assignment] runner._handle_bootstrap_source_write("main", "USER.md") assert ("main", "session-a", "full") not in runner._bootstrap_snapshots assert ("main", "session-b", "minimal") not in runner._bootstrap_snapshots assert ("other", "session-a", "full") in runner._bootstrap_snapshots assert ("main", "session-a") in runner._memory_snapshots assert "USER.md" not in filter_workspace_filenames_for_session(None, "subagent:worker") def test_runtime_write_callbacks_are_composed() -> None: runner = TurnRunner(provider_selector=None) memory_calls: list[tuple[str, str]] = [] bootstrap_calls: list[tuple[str, str]] = [] ctx = ToolContext( agent_id="main", on_memory_source_write=lambda agent_id, path: memory_calls.append((agent_id, path)), on_bootstrap_source_write=lambda agent_id, path: bootstrap_calls.append((agent_id, path)), ) runner._bootstrap_snapshots[("main", "session-a", "full")] = BootstrapSnapshot( workspace_files={"USER.md": "main"}, report=[], ) updated = runner._with_runtime_write_callbacks(ctx, "main") assert updated.on_memory_source_write is not None assert updated.on_bootstrap_source_write is not None updated.on_memory_source_write("main", "MEMORY.md") updated.on_bootstrap_source_write("main", "USER.md") assert memory_calls == [("main", "MEMORY.md")] assert bootstrap_calls == [("main", "USER.md")] assert ("main", "session-a", "full") not in runner._bootstrap_snapshots def test_unattended_bootstrap_context_skips_only_bootstrap_md(tmp_path) -> None: for filename in ( "AGENTS.md", "SOUL.md", "IDENTITY.md", "TOOLS.md", "USER.md", "BOOTSTRAP.md", ): (tmp_path / filename).write_text(f"{filename} body\n", encoding="utf-8") (tmp_path / "IDENTITY.md").write_text("name: Test Agent\n", encoding="utf-8") runner = TurnRunner( provider_selector=None, config=SimpleNamespace( workspace_dir=str(tmp_path), memory=SimpleNamespace(source="workspace"), tools=SimpleNamespace(profile=None), ), ) metadata: dict[str, object] = {} runner._assemble_prompt( "main", [], session_key="agent:main:auto", prompt_metadata=metadata, bootstrap_context_mode="unattended", ) report = metadata["bootstrap_files"] filenames = {item.filename for item in report} # type: ignore[attr-defined] assert "BOOTSTRAP.md" not in filenames assert {"AGENTS.md", "SOUL.md", "IDENTITY.md", "TOOLS.md", "USER.md"} <= filenames def test_prompt_metadata_uses_effective_memory_retrieval_metadata(tmp_path) -> None: (tmp_path / "AGENTS.md").write_text("agents\n", encoding="utf-8") class _Retriever: def effective_retrieval_metadata(self) -> dict[str, str]: return { "configured_retrieval_mode": "hybrid", "retrieval_mode": "hybrid", "embedding_requested_provider": "auto", "embedding_effective_provider": "local", "embedding_model": "BAAI/bge-small-zh-v1.5", "vector_weight": "0.7", "text_weight": "0.3", } runner = TurnRunner( provider_selector=None, config=SimpleNamespace( workspace_dir=str(tmp_path), memory=SimpleNamespace(source="workspace", retrieval_mode="hybrid"), tools=SimpleNamespace(profile=None), ), memory_retrievers={"main": _Retriever()}, ) metadata: dict[str, object] = {} runner._assemble_prompt( "main", [], session_key="agent:main:auto", prompt_metadata=metadata, ) assert metadata["retrieval_mode"] == "hybrid" assert metadata["embedding_effective_provider"] == "local" assert metadata["embedding_model"] == "BAAI/bge-small-zh-v1.5" assert metadata["memory_retrieval_vector_weight"] == "0.7" assert metadata["memory_retrieval_text_weight"] == "0.3" def test_fresh_user_session_omits_live_daily_notes_from_dynamic_context(tmp_path) -> None: (tmp_path / "AGENTS.md").write_text("agents\n", encoding="utf-8") runner = TurnRunner( provider_selector=None, config=SimpleNamespace( workspace_dir=str(tmp_path), memory=SimpleNamespace(source="workspace"), tools=SimpleNamespace(profile=None), ), ) runner._load_daily_notes = lambda _workspace_dir: {"2026-05-31.md": "stale Labubu context"} metadata: dict[str, object] = {} assembled = runner._assemble_prompt( "main", [], session_key="agent:main:fresh", prompt_metadata=metadata, fresh_user_session=True, ) dynamic = assembled[1] if isinstance(assembled, tuple) else "" assert "## Recent Notes" not in dynamic assert "stale Labubu context" not in dynamic assert metadata["daily_notes_fresh_session_omitted"] is True assert metadata["daily_notes_count_before_omit"] == 1 def test_fresh_user_session_omits_snapshot_daily_notes_from_dynamic_context(tmp_path) -> None: (tmp_path / "AGENTS.md").write_text("agents\n", encoding="utf-8") runner = TurnRunner( provider_selector=None, config=SimpleNamespace( workspace_dir=str(tmp_path), memory=SimpleNamespace(source="workspace"), tools=SimpleNamespace(profile=None), ), ) runner._memory_snapshots[("main", "agent:main:fresh")] = MemorySnapshot( memory_md="stable memory", daily_notes={"2026-05-31.md": "snapshot daily context"}, ) metadata: dict[str, object] = {} assembled = runner._assemble_prompt( "main", [], session_key="agent:main:fresh", prompt_metadata=metadata, fresh_user_session=True, ) dynamic = assembled[1] if isinstance(assembled, tuple) else "" full_prompt = "\n".join(assembled) if isinstance(assembled, tuple) else assembled assert "snapshot daily context" not in dynamic assert "stable memory" in full_prompt assert metadata["daily_notes_fresh_session_omitted"] is True assert metadata["daily_notes_count_before_omit"] == 1 def test_non_fresh_user_session_omits_live_daily_notes_from_dynamic_context(tmp_path) -> None: (tmp_path / "AGENTS.md").write_text("agents\n", encoding="utf-8") runner = TurnRunner( provider_selector=None, config=SimpleNamespace( workspace_dir=str(tmp_path), memory=SimpleNamespace(source="workspace"), tools=SimpleNamespace(profile=None), ), ) runner._load_daily_notes = lambda _workspace_dir: {"2026-05-31.md": "daily context"} metadata: dict[str, object] = {} assembled = runner._assemble_prompt( "main", [], session_key="agent:main:existing", prompt_metadata=metadata, ) dynamic = assembled[1] if isinstance(assembled, tuple) else "" assert "## Recent Notes" not in dynamic assert "daily context" not in dynamic assert metadata["daily_notes_omitted"] is True assert metadata["daily_notes_policy_reason"] == "auto_injection_disabled" assert metadata["daily_notes_count_before_omit"] == 1 def test_daily_notes_auto_omit_preserves_memory_md(tmp_path) -> None: (tmp_path / "AGENTS.md").write_text("agents\n", encoding="utf-8") (tmp_path / "MEMORY.md").write_text("stable curated memory\n", encoding="utf-8") runner = TurnRunner( provider_selector=None, config=SimpleNamespace( workspace_dir=str(tmp_path), memory=SimpleNamespace(source="workspace"), tools=SimpleNamespace(profile=None), ), ) runner._load_daily_notes = lambda _workspace_dir: {"2026-05-31.md": "daily context"} metadata: dict[str, object] = {} assembled = runner._assemble_prompt( "main", [], session_key="agent:main:existing", prompt_metadata=metadata, ) full_prompt = "\n".join(assembled) if isinstance(assembled, tuple) else assembled dynamic = assembled[1] if isinstance(assembled, tuple) else "" assert "stable curated memory" in full_prompt assert "daily context" not in dynamic assert metadata["memory_md_present"] is True assert metadata["daily_notes_omitted"] is True def test_stateless_bootstrap_context_skips_persona_memory_and_bootstrap(tmp_path) -> None: for filename in ( "AGENTS.md", "SOUL.md", "IDENTITY.md", "TOOLS.md", "USER.md", "MEMORY.md", "HEARTBEAT.md", "BOOTSTRAP.md", ): (tmp_path / filename).write_text(f"{filename} body\n", encoding="utf-8") runner = TurnRunner( provider_selector=None, config=SimpleNamespace( workspace_dir=str(tmp_path), memory=SimpleNamespace(source="workspace"), tools=SimpleNamespace(profile=None), ), ) metadata: dict[str, object] = {} runner._assemble_prompt( "main", [], session_key="agent:main:auto", prompt_metadata=metadata, bootstrap_context_mode="stateless", ) report = metadata["bootstrap_files"] filenames = {item.filename for item in report} # type: ignore[attr-defined] assert filenames == {"TOOLS.md"} assert metadata["memory_md_present"] is False def test_stateless_keep_project_rules_preserves_only_agents_md(tmp_path) -> None: for filename in ( "AGENTS.md", "SOUL.md", "IDENTITY.md", "TOOLS.md", "USER.md", "MEMORY.md", "HEARTBEAT.md", "BOOTSTRAP.md", ): (tmp_path / filename).write_text(f"{filename} body\n", encoding="utf-8") runner = TurnRunner( provider_selector=None, config=SimpleNamespace( workspace_dir=str(tmp_path), memory=SimpleNamespace(source="workspace"), tools=SimpleNamespace(profile=None), ), ) metadata: dict[str, object] = {} runner._assemble_prompt( "main", [], session_key="agent:main:auto", prompt_metadata=metadata, bootstrap_context_mode="stateless_keep_project_rules", ) report = metadata["bootstrap_files"] filenames = {item.filename for item in report} # type: ignore[attr-defined] assert filenames == {"AGENTS.md", "TOOLS.md"} assert metadata["memory_md_present"] is False def test_full_and_unattended_bootstrap_snapshots_use_distinct_keys(tmp_path) -> None: (tmp_path / "AGENTS.md").write_text("agents\n", encoding="utf-8") (tmp_path / "BOOTSTRAP.md").write_text("bootstrap\n", encoding="utf-8") runner = TurnRunner( provider_selector=None, config=SimpleNamespace( workspace_dir=str(tmp_path), memory=SimpleNamespace(source="workspace"), tools=SimpleNamespace(profile=None), ), ) session_key = "agent:main:auto" runner._assemble_prompt("main", [], session_key=session_key) runner._assemble_prompt( "main", [], session_key=session_key, bootstrap_context_mode="unattended", ) assert ("main", session_key, "full") in runner._bootstrap_snapshots assert ("main", session_key, "unattended") in runner._bootstrap_snapshots full_snapshot = runner._bootstrap_snapshots[("main", session_key, "full")] unattended_snapshot = runner._bootstrap_snapshots[("main", session_key, "unattended")] assert "BOOTSTRAP.md" in full_snapshot.workspace_files assert ( "BOOTSTRAP.md" not in unattended_snapshot.workspace_files )