diff --git a/backend/app/services/agent_seeder.py b/backend/app/services/agent_seeder.py index 92d2d42d6..790dc07ff 100644 --- a/backend/app/services/agent_seeder.py +++ b/backend/app/services/agent_seeder.py @@ -4,7 +4,7 @@ from loguru import logger -from sqlalchemy import select +from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from sqlalchemy.exc import IntegrityError @@ -13,6 +13,7 @@ from app.models.agent import Agent, AgentPermission from app.models.org import AgentAgentRelationship from app.models.skill import Skill +from app.models.tenant_setting import TenantSetting from app.models.tool import Tool, AgentTool from app.models.trigger import AgentTrigger from app.models.user import User @@ -23,6 +24,8 @@ settings = get_settings() SEED_MARKER_KEY = "_bootstrap/.seeded" +DEFAULT_AGENT_SEED_SETTING_KEY = "bootstrap:default_agents:v1" +DEFAULT_AGENT_NAMES = {"morty": "Morty", "meeseeks": "Meeseeks"} async def _read_seed_marker() -> str: @@ -42,6 +45,147 @@ async def _append_seed_marker(line: str) -> None: await storage.write_text(SEED_MARKER_KEY, updated, encoding="utf-8") +def _parse_default_agent_ids(value: object) -> dict[str, uuid.UUID | None]: + """Read stable default-Agent IDs from a tenant setting value.""" + raw_agents = value.get("agents") if isinstance(value, dict) else None + raw_agents = raw_agents if isinstance(raw_agents, dict) else {} + parsed: dict[str, uuid.UUID | None] = {} + for key in DEFAULT_AGENT_NAMES: + raw_id = raw_agents.get(key) + try: + parsed[key] = uuid.UUID(str(raw_id)) if raw_id else None + except (TypeError, ValueError, AttributeError): + parsed[key] = None + return parsed + + +def _parse_legacy_default_agent_ids(marker: str) -> dict[str, uuid.UUID | None]: + """Parse the last valid ID for each default Agent from the legacy marker.""" + parsed: dict[str, uuid.UUID | None] = {key: None for key in DEFAULT_AGENT_NAMES} + for line in marker.splitlines(): + key, separator, raw_id = line.partition("=") + if not separator or key not in DEFAULT_AGENT_NAMES: + continue + try: + parsed[key] = uuid.UUID(raw_id.strip()) + except ValueError: + continue + return parsed + + +def _default_agent_setting_value( + agent_ids: dict[str, uuid.UUID | None], + *, + source: str, +) -> dict: + return { + "initialized": True, + "agents": { + key: str(agent_ids.get(key)) if agent_ids.get(key) else None + for key in DEFAULT_AGENT_NAMES + }, + "source": source, + } + + +async def _lock_default_agent_seed(db: AsyncSession, tenant_id: uuid.UUID) -> None: + """Serialize first-seed and compatibility backfill for one tenant.""" + scope = f"default-agent-bootstrap:{tenant_id}" + await db.execute( + select(func.pg_advisory_xact_lock(func.hashtextextended(scope, 0))) + ) + + +async def _load_default_agents_by_ids( + db: AsyncSession, + tenant_id: uuid.UUID, + agent_ids: dict[str, uuid.UUID | None], +) -> dict[str, Agent | None]: + wanted_ids = {agent_id for agent_id in agent_ids.values() if agent_id is not None} + if not wanted_ids: + return {key: None for key in DEFAULT_AGENT_NAMES} + result = await db.execute( + select(Agent).where( + Agent.tenant_id == tenant_id, + Agent.id.in_(wanted_ids), + Agent.agent_type == "native", + ) + ) + agents_by_id = {agent.id: agent for agent in result.scalars().all()} + return { + key: agents_by_id.get(agent_id) if agent_id else None + for key, agent_id in agent_ids.items() + } + + +async def _load_historical_default_agents( + db: AsyncSession, + tenant_id: uuid.UUID, +) -> dict[str, Agent | None]: + """Find canonical-name history, including stopped and logically deleted rows.""" + result = await db.execute( + select(Agent) + .where( + Agent.tenant_id == tenant_id, + Agent.name.in_(DEFAULT_AGENT_NAMES.values()), + Agent.agent_type == "native", + ) + .order_by(Agent.created_at.asc()) + ) + historical: dict[str, Agent | None] = {key: None for key in DEFAULT_AGENT_NAMES} + key_by_name = {name: key for key, name in DEFAULT_AGENT_NAMES.items()} + for agent in result.scalars().all(): + key = key_by_name.get(agent.name) + if key and historical[key] is None: + historical[key] = agent + return historical + + +async def _repair_seeded_default_agents( + db: AsyncSession, + agents: dict[str, Agent | None], + *, + created_keys: set[str] | None = None, +) -> None: + """Repair storage only for default Agents that still exist and are not deleted.""" + repairable = { + key: agent + for key, agent in agents.items() + if agent is not None and agent.deleted_at is None + } + if not repairable: + return + + all_skills_result = await db.execute( + select(Skill).options(selectinload(Skill.files)) + ) + all_skills = {skill.folder_name: skill for skill in all_skills_result.scalars().all()} + repair_specs = { + "morty": (MORTY_SOUL, MORTY_SKILLS), + "meeseeks": (MEESEEKS_SOUL, MEESEEKS_SKILLS), + } + for key, agent in repairable.items(): + soul_content, skill_folders = repair_specs[key] + await _repair_default_agent_storage( + db, + agent, + soul_content=soul_content, + skill_folders=skill_folders, + all_skills=all_skills, + overwrite_skill_files=key in (created_keys or set()), + ) + + +async def _append_default_agent_seed_marker( + agent_ids: dict[str, uuid.UUID | None], +) -> None: + """Preserve other bootstrap entries while recording default-Agent IDs.""" + await _append_seed_marker("seeded") + for key, agent_id in agent_ids.items(): + if agent_id: + await _append_seed_marker(f"{key}={agent_id}") + + async def _repair_default_agent_storage( db: AsyncSession, agent: Agent, @@ -262,14 +406,9 @@ async def _repair_default_agent_storage( async def seed_default_agents(): - """Create missing default agents and repair missing storage for existing ones. - - Database rows are the duplicate-creation guard. The storage marker is only - an operational hint because deployments can switch or lose storage while - preserving the database. - """ + """Initialize default Agents once, then only repair surviving Agent storage.""" + marker_ids_to_write: dict[str, uuid.UUID | None] | None = None async with async_session() as db: - # Get platform admin as creator admin_result = await db.execute( select(User).where(User.role == "platform_admin").limit(1) @@ -279,27 +418,78 @@ async def seed_default_agents(): logger.warning("[AgentSeeder] No platform admin found, skipping default agents") return - # DB-backed idempotency is the source of truth. The storage marker can - # disappear when deployments switch volumes/backends, so it is only a - # fast-path hint and must never be the only duplicate guard. - existing_result = await db.execute( - select(Agent) - .where( - Agent.tenant_id == admin.tenant_id, - Agent.name.in_(["Morty", "Meeseeks"]), - Agent.agent_type == "native", - Agent.status != "stopped", + await _lock_default_agent_seed(db, admin.tenant_id) + + setting_result = await db.execute( + select(TenantSetting).where( + TenantSetting.tenant_id == admin.tenant_id, + TenantSetting.key == DEFAULT_AGENT_SEED_SETTING_KEY, ) - .order_by(Agent.created_at.asc()) ) - existing_by_name: dict[str, Agent] = {} - for agent in existing_result.scalars().all(): - existing_by_name.setdefault(agent.name, agent) + seed_setting = setting_result.scalar_one_or_none() + + if seed_setting is not None: + seed_value = seed_setting.value if isinstance(seed_setting.value, dict) else {} + if seed_value.get("initialized") is not True: + logger.warning( + "[AgentSeeder] Default-Agent initialization setting is malformed; " + "skipping creation conservatively" + ) + agent_ids = _parse_default_agent_ids(seed_setting.value) + seeded_agents = await _load_default_agents_by_ids( + db, + admin.tenant_id, + agent_ids, + ) + await _repair_seeded_default_agents(db, seeded_agents) + await db.commit() + logger.info( + "[AgentSeeder] Default Agents already initialized; " + "creation skipped and surviving storage checked" + ) + return - created_agents: list[Agent] = [] - created_names: set[str] = set() + # Existing deployments predate the DB setting. Recover stable IDs from + # the shared legacy marker first, then fall back to canonical-name DB + # history including stopped and logically deleted rows. + try: + legacy_ids = _parse_legacy_default_agent_ids(await _read_seed_marker()) + except Exception as exc: + logger.warning(f"[AgentSeeder] Legacy seed marker unavailable: {exc}") + legacy_ids = {key: None for key in DEFAULT_AGENT_NAMES} - if "Morty" not in existing_by_name: + seeded_agents = await _load_default_agents_by_ids( + db, + admin.tenant_id, + legacy_ids, + ) + if any(agent is not None for agent in seeded_agents.values()): + source = "legacy_marker" + else: + seeded_agents = await _load_historical_default_agents(db, admin.tenant_id) + source = "database_history" + + if any(agent is not None for agent in seeded_agents.values()): + agent_ids = { + key: agent.id if agent is not None else None + for key, agent in seeded_agents.items() + } + db.add( + TenantSetting( + tenant_id=admin.tenant_id, + key=DEFAULT_AGENT_SEED_SETTING_KEY, + value=_default_agent_setting_value(agent_ids, source=source), + ) + ) + await _repair_seeded_default_agents(db, seeded_agents) + await db.commit() + marker_ids_to_write = agent_ids + logger.info( + "[AgentSeeder] Backfilled default-Agent initialization state: " + f"tenant={admin.tenant_id} source={source}" + ) + else: + # No durable initialization evidence: this is a fresh tenant. morty = Agent( name="Morty", role_description="Research analyst & knowledge assistant — curious, thorough, great at finding and synthesizing information", @@ -309,13 +499,6 @@ async def seed_default_agents(): tenant_id=admin.tenant_id, status="idle", ) - db.add(morty) - created_agents.append(morty) - created_names.add("Morty") - else: - morty = existing_by_name["Morty"] - - if "Meeseeks" not in existing_by_name: meeseeks = Agent( name="Meeseeks", role_description="Task executor & project manager — goal-oriented, systematic planner, strong at breaking down and completing complex tasks", @@ -325,100 +508,86 @@ async def seed_default_agents(): tenant_id=admin.tenant_id, status="idle", ) + db.add(morty) db.add(meeseeks) - created_agents.append(meeseeks) - created_names.add("Meeseeks") - else: - meeseeks = existing_by_name["Meeseeks"] - - await db.flush() # get IDs - - # ── Participant identities ── - from app.models.participant import Participant - for agent in created_agents: - db.add(Participant(type="agent", ref_id=agent.id, display_name=agent.name, avatar_url=agent.avatar_url)) - await db.flush() - - # ── Permissions (company-wide, manage) ── - for agent in created_agents: - db.add(AgentPermission(agent_id=agent.id, scope_type="company", access_level="manage")) - - # ── Assign skills ── - all_skills_result = await db.execute( - select(Skill).options(selectinload(Skill.files)) - ) - all_skills = {s.folder_name: s for s in all_skills_result.scalars().all()} - - await _repair_default_agent_storage( - db, - morty, - soul_content=MORTY_SOUL, - skill_folders=MORTY_SKILLS, - all_skills=all_skills, - overwrite_skill_files=morty.name in created_names, - ) - await _repair_default_agent_storage( - db, - meeseeks, - soul_content=MEESEEKS_SOUL, - skill_folders=MEESEEKS_SKILLS, - all_skills=all_skills, - overwrite_skill_files=meeseeks.name in created_names, - ) + await db.flush() - # ── Assign all default tools ── - default_tools_result = await db.execute( - select(Tool).where(Tool.is_default) - ) - default_tools = default_tools_result.scalars().all() + created_agents = {"morty": morty, "meeseeks": meeseeks} + agent_ids = {key: agent.id for key, agent in created_agents.items()} + db.add( + TenantSetting( + tenant_id=admin.tenant_id, + key=DEFAULT_AGENT_SEED_SETTING_KEY, + value=_default_agent_setting_value(agent_ids, source="created"), + ) + ) - for agent in created_agents: - for tool in default_tools: - db.add(AgentTool(agent_id=agent.id, tool_id=tool.id, enabled=True)) + from app.models.participant import Participant - # ── Mutual relationships ── - relationship_specs = [ - ( - morty.id, - meeseeks.id, - "Expert task executor who breaks down complex tasks into structured plans and executes them systematically. Delegate multi-step tasks to him.", - ), - ( - meeseeks.id, - morty.id, - "Research expert with strong learning ability. Ask him for information retrieval, web research, data analysis, and knowledge synthesis.", - ), - ] - for agent_id, target_agent_id, description in relationship_specs: - rel_result = await db.execute( - select(AgentAgentRelationship).where( - AgentAgentRelationship.agent_id == agent_id, - AgentAgentRelationship.target_agent_id == target_agent_id, + for agent in created_agents.values(): + db.add( + Participant( + type="agent", + ref_id=agent.id, + display_name=agent.name, + avatar_url=agent.avatar_url, + ) ) + db.add( + AgentPermission( + agent_id=agent.id, + scope_type="company", + access_level="manage", + ) + ) + await db.flush() + + await _repair_seeded_default_agents( + db, + created_agents, + created_keys=set(created_agents), ) - if not rel_result.scalar_one_or_none(): - db.add(AgentAgentRelationship( - agent_id=agent_id, - target_agent_id=target_agent_id, - relation="collaborator", - description=description, - )) + default_tools_result = await db.execute(select(Tool).where(Tool.is_default)) + default_tools = default_tools_result.scalars().all() + for agent in created_agents.values(): + for tool in default_tools: + db.add(AgentTool(agent_id=agent.id, tool_id=tool.id, enabled=True)) + relationship_specs = [ + ( + morty.id, + meeseeks.id, + "Expert task executor who breaks down complex tasks into structured plans and executes them systematically. Delegate multi-step tasks to him.", + ), + ( + meeseeks.id, + morty.id, + "Research expert with strong learning ability. Ask him for information retrieval, web research, data analysis, and knowledge synthesis.", + ), + ] + for agent_id, target_agent_id, description in relationship_specs: + db.add( + AgentAgentRelationship( + agent_id=agent_id, + target_agent_id=target_agent_id, + relation="collaborator", + description=description, + ) + ) - await db.commit() - logger.info( - "[AgentSeeder] Default agent seeding complete: " - f"Morty ({morty.id}), Meeseeks ({meeseeks.id}), created={len(created_agents)}" - ) + await db.commit() + marker_ids_to_write = agent_ids + logger.info( + "[AgentSeeder] Default Agent initialization complete: " + f"Morty ({morty.id}), Meeseeks ({meeseeks.id})" + ) - # Write seed marker AFTER a successful commit so a failed seed can be retried - await get_storage_backend().write_text( - SEED_MARKER_KEY, - f"seeded\nmorty={morty.id}\nmeeseeks={meeseeks.id}\n", - encoding="utf-8", - ) - logger.info(f"[AgentSeeder] Wrote seed marker to {SEED_MARKER_KEY}") + if marker_ids_to_write: + try: + await _append_default_agent_seed_marker(marker_ids_to_write) + except Exception as exc: + logger.warning(f"[AgentSeeder] Failed to update legacy seed marker: {exc}") async def seed_okr_agent(): diff --git a/backend/tests/test_agent_seeder_storage_repair.py b/backend/tests/test_agent_seeder_storage_repair.py index 52f9a6aa7..94c791b79 100644 --- a/backend/tests/test_agent_seeder_storage_repair.py +++ b/backend/tests/test_agent_seeder_storage_repair.py @@ -1,3 +1,4 @@ +from datetime import datetime, timezone from types import SimpleNamespace from unittest.mock import AsyncMock import uuid @@ -30,8 +31,18 @@ async def __aexit__(self, exc_type, exc, traceback): return False -def _agent(name: str = "Morty") -> SimpleNamespace: - return SimpleNamespace(id=uuid.uuid4(), name=name) +def _agent( + name: str = "Morty", + *, + status: str = "idle", + deleted_at=None, +) -> SimpleNamespace: + return SimpleNamespace( + id=uuid.uuid4(), + name=name, + status=status, + deleted_at=deleted_at, + ) def _skill(folder_name: str = "skill-creator", *, is_default: bool = True) -> SimpleNamespace: @@ -141,23 +152,30 @@ async def test_seed_existing_default_agents_still_runs_storage_repair(monkeypatc admin = SimpleNamespace(id=uuid.uuid4(), tenant_id=uuid.uuid4()) morty = _agent("Morty") meeseeks = _agent("Meeseeks") + added = [] + + async def execute(statement): + sql = str(statement) + if "FROM users" in sql: + return _Result(scalar=admin) + if "pg_advisory_xact_lock" in sql: + return _Result() + if "FROM tenant_settings" in sql: + return _Result(scalar=None) + if "FROM agents" in sql and "agents.id IN" in sql: + return _Result(scalars=[]) + if "FROM agents" in sql: + return _Result(scalars=[morty, meeseeks]) + return _Result(scalars=[]) + session = SimpleNamespace( - execute=AsyncMock( - side_effect=[ - _Result(scalar=admin), - _Result(scalars=[morty, meeseeks]), - _Result(scalars=[]), - _Result(scalars=[]), - _Result(scalar=None), - _Result(scalar=None), - ] - ), + execute=AsyncMock(side_effect=execute), flush=AsyncMock(), commit=AsyncMock(), - add=lambda value: None, + add=added.append, ) repair = AsyncMock(return_value=False) - storage = SimpleNamespace(write_text=AsyncMock()) + storage = _empty_storage() monkeypatch.setattr(agent_seeder, "async_session", lambda: _SessionContext(session)) monkeypatch.setattr(agent_seeder, "_repair_default_agent_storage", repair) monkeypatch.setattr(agent_seeder, "get_storage_backend", lambda: storage) @@ -166,5 +184,267 @@ async def test_seed_existing_default_agents_still_runs_storage_repair(monkeypatc assert repair.await_count == 2 assert [call.args[1].name for call in repair.await_args_list] == ["Morty", "Meeseeks"] + assert any(value.__class__.__name__ == "TenantSetting" for value in added) + session.commit.assert_awaited_once() + assert storage.write_text.await_count >= 1 + + +def _empty_storage(*, marker: str = "") -> SimpleNamespace: + return SimpleNamespace( + exists=AsyncMock(return_value=bool(marker)), + read_text=AsyncMock(return_value=marker), + write_text=AsyncMock(), + ) + + +@pytest.mark.asyncio +async def test_append_default_agent_marker_preserves_other_seed_entries(monkeypatch): + content = "seeded\nokr_agent=existing\n" + + async def read_text(*_args, **_kwargs): + return storage.content + + async def write_text(_key, value, **_kwargs): + storage.content = value + + storage = SimpleNamespace( + content=content, + exists=AsyncMock(return_value=True), + read_text=AsyncMock(side_effect=read_text), + write_text=AsyncMock(side_effect=write_text), + ) + monkeypatch.setattr(agent_seeder, "get_storage_backend", lambda: storage) + morty_id = uuid.uuid4() + meeseeks_id = uuid.uuid4() + + await agent_seeder._append_default_agent_seed_marker( + {"morty": morty_id, "meeseeks": meeseeks_id} + ) + + assert "okr_agent=existing\n" in storage.content + assert f"morty={morty_id}\n" in storage.content + assert f"meeseeks={meeseeks_id}\n" in storage.content + + +@pytest.mark.asyncio +async def test_seed_deleted_default_agents_backfills_without_recreating(monkeypatch): + admin = SimpleNamespace(id=uuid.uuid4(), tenant_id=uuid.uuid4()) + deleted_at = datetime.now(timezone.utc) + deleted_agents = [ + _agent("Morty", status="stopped", deleted_at=deleted_at), + _agent("Meeseeks", status="stopped", deleted_at=deleted_at), + ] + added = [] + + async def execute(statement): + sql = str(statement) + if "FROM users" in sql: + return _Result(scalar=admin) + if "pg_advisory_xact_lock" in sql: + return _Result() + if "FROM tenant_settings" in sql: + return _Result(scalar=None) + if "FROM agents" in sql: + if "agents.status !=" in sql: + return _Result(scalars=[]) + return _Result(scalars=deleted_agents) + return _Result(scalars=[]) + + session = SimpleNamespace( + execute=AsyncMock(side_effect=execute), + flush=AsyncMock(), + commit=AsyncMock(), + add=added.append, + ) + repair = AsyncMock(return_value=False) + storage = _empty_storage() + monkeypatch.setattr(agent_seeder, "async_session", lambda: _SessionContext(session)) + monkeypatch.setattr(agent_seeder, "_repair_default_agent_storage", repair) + monkeypatch.setattr(agent_seeder, "get_storage_backend", lambda: storage) + + await agent_seeder.seed_default_agents() + + assert not any(isinstance(value, agent_seeder.Agent) for value in added) + assert any(value.__class__.__name__ == "TenantSetting" for value in added) + repair.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_seed_legacy_marker_backfills_renamed_agents_without_recreating(monkeypatch): + admin = SimpleNamespace(id=uuid.uuid4(), tenant_id=uuid.uuid4()) + renamed_morty = _agent("Researcher") + renamed_meeseeks = _agent("Executor") + marker = ( + "seeded\n" + f"morty={renamed_morty.id}\n" + f"meeseeks={renamed_meeseeks.id}\n" + ) + added = [] + + async def execute(statement): + sql = str(statement) + if "FROM users" in sql: + return _Result(scalar=admin) + if "pg_advisory_xact_lock" in sql: + return _Result() + if "FROM tenant_settings" in sql: + return _Result(scalar=None) + if "FROM agents" in sql and "agents.id IN" in sql: + return _Result(scalars=[renamed_morty, renamed_meeseeks]) + if "FROM agents" in sql: + return _Result(scalars=[]) + return _Result(scalars=[]) + + session = SimpleNamespace( + execute=AsyncMock(side_effect=execute), + flush=AsyncMock(), + commit=AsyncMock(), + add=added.append, + ) + repair = AsyncMock(return_value=False) + storage = _empty_storage(marker=marker) + monkeypatch.setattr(agent_seeder, "async_session", lambda: _SessionContext(session)) + monkeypatch.setattr(agent_seeder, "_repair_default_agent_storage", repair) + monkeypatch.setattr(agent_seeder, "get_storage_backend", lambda: storage) + + await agent_seeder.seed_default_agents() + + assert not any(isinstance(value, agent_seeder.Agent) for value in added) + assert any(value.__class__.__name__ == "TenantSetting" for value in added) + assert [call.args[1].id for call in repair.await_args_list] == [ + renamed_morty.id, + renamed_meeseeks.id, + ] + + +@pytest.mark.asyncio +async def test_seed_database_marker_skips_deleted_and_repairs_stopped_survivor(monkeypatch): + admin = SimpleNamespace(id=uuid.uuid4(), tenant_id=uuid.uuid4()) + deleted_morty = _agent( + "Morty", + status="stopped", + deleted_at=datetime.now(timezone.utc), + ) + stopped_meeseeks = _agent("Meeseeks", status="stopped") + setting = SimpleNamespace( + value={ + "initialized": True, + "agents": { + "morty": str(deleted_morty.id), + "meeseeks": str(stopped_meeseeks.id), + }, + "source": "created", + } + ) + added = [] + + async def execute(statement): + sql = str(statement) + if "FROM users" in sql: + return _Result(scalar=admin) + if "pg_advisory_xact_lock" in sql: + return _Result() + if "FROM tenant_settings" in sql: + return _Result(scalar=setting) + if "FROM agents" in sql: + return _Result(scalars=[deleted_morty, stopped_meeseeks]) + return _Result(scalars=[]) + + session = SimpleNamespace( + execute=AsyncMock(side_effect=execute), + flush=AsyncMock(), + commit=AsyncMock(), + add=added.append, + ) + repair = AsyncMock(return_value=False) + storage = _empty_storage() + monkeypatch.setattr(agent_seeder, "async_session", lambda: _SessionContext(session)) + monkeypatch.setattr(agent_seeder, "_repair_default_agent_storage", repair) + monkeypatch.setattr(agent_seeder, "get_storage_backend", lambda: storage) + + await agent_seeder.seed_default_agents() + + assert not any(isinstance(value, agent_seeder.Agent) for value in added) + repair.assert_awaited_once() + assert repair.await_args.args[1].id == stopped_meeseeks.id + storage.write_text.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_seed_malformed_database_marker_never_recreates(monkeypatch): + admin = SimpleNamespace(id=uuid.uuid4(), tenant_id=uuid.uuid4()) + setting = SimpleNamespace(value={"unexpected": "value"}) + added = [] + + async def execute(statement): + sql = str(statement) + if "FROM users" in sql: + return _Result(scalar=admin) + if "pg_advisory_xact_lock" in sql: + return _Result() + if "FROM tenant_settings" in sql: + return _Result(scalar=setting) + return _Result(scalars=[]) + + session = SimpleNamespace( + execute=AsyncMock(side_effect=execute), + flush=AsyncMock(), + commit=AsyncMock(), + add=added.append, + ) + repair = AsyncMock(return_value=False) + monkeypatch.setattr(agent_seeder, "async_session", lambda: _SessionContext(session)) + monkeypatch.setattr(agent_seeder, "_repair_default_agent_storage", repair) + + await agent_seeder.seed_default_agents() + + assert not any(isinstance(value, agent_seeder.Agent) for value in added) + repair.assert_not_awaited() + session.commit.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_seed_fresh_tenant_creates_agents_and_database_marker(monkeypatch): + admin = SimpleNamespace(id=uuid.uuid4(), tenant_id=uuid.uuid4()) + added = [] + + async def execute(statement): + sql = str(statement) + if "FROM users" in sql: + return _Result(scalar=admin) + if "pg_advisory_xact_lock" in sql: + return _Result() + if "FROM tenant_settings" in sql: + return _Result(scalar=None) + return _Result(scalars=[]) + + async def flush(): + for value in added: + if isinstance(value, agent_seeder.Agent) and value.id is None: + value.id = uuid.uuid4() + + session = SimpleNamespace( + execute=AsyncMock(side_effect=execute), + flush=AsyncMock(side_effect=flush), + commit=AsyncMock(), + add=added.append, + ) + repair = AsyncMock(return_value=False) + storage = _empty_storage() + monkeypatch.setattr(agent_seeder, "async_session", lambda: _SessionContext(session)) + monkeypatch.setattr(agent_seeder, "_repair_default_agent_storage", repair) + monkeypatch.setattr(agent_seeder, "get_storage_backend", lambda: storage) + + await agent_seeder.seed_default_agents() + + created_agents = [value for value in added if isinstance(value, agent_seeder.Agent)] + settings = [value for value in added if value.__class__.__name__ == "TenantSetting"] + assert [agent.name for agent in created_agents] == ["Morty", "Meeseeks"] + assert len(settings) == 1 + assert settings[0].value["initialized"] is True + assert settings[0].value["source"] == "created" + assert all(settings[0].value["agents"].values()) + assert repair.await_count == 2 session.commit.assert_awaited_once() - storage.write_text.assert_awaited_once() + executed_sql = "\n".join(str(call.args[0]) for call in session.execute.await_args_list) + assert "pg_advisory_xact_lock" in executed_sql diff --git a/docs/prd/features/agent-directory/default-agent-seeding-technical-design.md b/docs/prd/features/agent-directory/default-agent-seeding-technical-design.md new file mode 100644 index 000000000..82da0b852 --- /dev/null +++ b/docs/prd/features/agent-directory/default-agent-seeding-technical-design.md @@ -0,0 +1,280 @@ +# 默认 Agent 一次性初始化技术方案 + +> 状态:待实现 +> +> 范围:Morty、Meeseeks 的首次创建、升级兼容和存储自愈 + +## 1. 业务语义 + +Morty 和 Meeseeks 是租户首次完成平台初始化时创建的默认 Agent。 + +初始化成功后,平台必须尊重用户对这两个 Agent 的生命周期操作: + +- 用户删除后,后续启动、重启和升级不得重新创建。 +- 用户重命名后,不得因为默认名称消失而创建同名副本。 +- 用户仅停止 Agent 时,不得创建副本。 +- 未删除的默认 Agent 如果 workspace 或 Skills 存储损坏,启动时仍可执行非覆盖式修复。 + +因此,“是否创建默认 Agent”与“是否修复默认 Agent 存储”必须是两项独立判断。 + +## 2. 当前实现与问题 + +### 2.1 当前调用链 + +`seed_default_agents()` 在两个入口运行: + +- 后端启动流程:`backend/app/main.py` +- 首个平台注册用户创建完成后:`backend/app/api/auth.py` + +重复调用本身是允许的,前提是 seeder 具备可靠的一次性语义。 + +### 2.2 当前创建判据 + +当前 seeder 按以下条件查找已有默认 Agent: + +```python +Agent.tenant_id == admin.tenant_id +Agent.name.in_(["Morty", "Meeseeks"]) +Agent.agent_type == "native" +Agent.status != "stopped" +``` + +如果对应名称不在查询结果中,就创建新的 Agent。 + +这个判据把可变运行状态当成了初始化事实: + +- 删除接口会保留 Agent 行,同时设置 `deleted_at` 和 `status="stopped"`。 +- 停止接口也会设置 `status="stopped"`。 +- 重命名会改变 `name`。 + +因此删除、停止和重命名都可能被错误解释为“从未初始化”。 + +### 2.3 现有 seed marker + +存储中已有 `_bootstrap/.seeded`,但默认 Agent seeder 当前只写入、不读取该标记。该文件不能作为新的唯一事实源:部署可能更换或丢失存储,而数据库仍然保留。 + +### 2.4 必须保留的存储自愈 + +现有 `_repair_default_agent_storage()` 会为仍存在的默认 Agent 修复缺失的根目录和 Skills 目录,并避免覆盖用户文件。这个能力必须保留,不能恢复成“发现 seed marker 后整段 seeder 直接返回”。 + +## 3. 技术目标 + +1. 使用租户级、持久、与名称和运行状态无关的初始化事实。 +2. 默认 Agent 每个租户最多自动初始化一次。 +3. 删除、停止、重命名均不触发重新创建。 +4. 对未删除的默认 Agent 保留存储自愈。 +5. 兼容没有数据库初始化标记的现有部署。 +6. 多实例同时启动时不得重复创建。 +7. 不增加依赖,优先复用现有表和数据库锁模式。 + +## 4. 数据事实源 + +### 4.1 新的规范事实 + +复用现有 `tenant_settings` 表,不新增表和 Alembic migration。 + +建议设置项: + +```text +key = "bootstrap:default_agents:v1" +``` + +建议 value: + +```json +{ + "initialized": true, + "agents": { + "morty": "", + "meeseeks": "" + }, + "source": "created|legacy_marker|database_history" +} +``` + +语义: + +- 设置项存在即表示该租户已经完成过默认 Agent 初始化;`initialized=true` 用于校验和诊断。即使 value 损坏,也必须保守地停止自动创建并记录告警。 +- Agent ID 是稳定身份,用于后续存储修复;不再通过名称反查身份。 +- ID 对应 Agent 已删除或物理不存在时,也不得重新创建。 +- `source` 仅用于诊断和升级审计,不参与业务判断。 + +### 4.2 删除事实 + +`Agent.deleted_at` 是 Agent 是否被用户逻辑删除的事实源。 + +- `deleted_at is None`:Agent 仍存在,可以检查和修复存储。 +- `deleted_at is not None`:Agent 已删除,跳过存储修复,也不得补建。 +- `status` 只描述运行状态,不参与初始化或删除判断。 + +### 4.3 legacy marker 的角色 + +`_bootstrap/.seeded` 只用于现有部署的兼容识别和运维诊断,不再作为长期唯一事实源。 + +后续如仍需写入 legacy marker,必须使用追加/合并方式,不能覆盖 `okr_agent` 等其他 seed 信息。 + +## 5. 核心流程 + +### 5.1 并发边界 + +进入租户默认 Agent 初始化流程后,先获取租户级 PostgreSQL transaction advisory lock。锁键建议包含租户 ID和固定命名空间: + +```text +default-agent-bootstrap: +``` + +锁内重新读取 `tenant_settings`,避免多个后端实例同时判断“未初始化”并重复创建。 + +### 5.2 已有数据库标记 + +如果 `bootstrap:default_agents:v1` 已存在: + +1. 不执行任何默认 Agent 创建。 +2. 按设置中保存的 Agent ID 查询数据库,查询必须包含 stopped 和逻辑删除行。 +3. 对 `deleted_at is None` 的 Agent 调用 `_repair_default_agent_storage()`。 +4. 对已删除或不存在的 Agent 直接跳过。 + +### 5.3 新租户首次初始化 + +如果数据库标记不存在,并且兼容识别没有发现历史初始化事实: + +1. 创建 Morty 和 Meeseeks。 +2. 创建 Participant、权限、默认工具和相互关系。 +3. 初始化 workspace 和 Skills。 +4. 在同一数据库事务中写入 `bootstrap:default_agents:v1`,保存两个 Agent ID。 +5. 提交事务。 +6. 数据库提交成功后,以追加方式更新 legacy marker;marker 写入失败只记录告警,不回滚已经成立的数据库事实。 + +数据库中的 Agent 和初始化设置必须一起提交,避免出现“Agent 已创建但初始化设置缺失”的中间状态。 + +## 6. 现有部署兼容 + +### 6.1 是否必须回填 + +如果采用 `tenant_settings` 作为新的规范事实,现有租户必须建立这个事实,否则“数据库标记不存在”仍可能被错误理解为全新租户。 + +但不需要: + +- 新增 Alembic 数据迁移; +- 单独执行离线回填脚本; +- 人工逐租户处理。 + +采用 seeder 首次运行时的懒回填即可。也就是说,兼容回填是逻辑上必须的,但不需要独立发布步骤。 + +### 6.2 懒回填顺序 + +数据库标记不存在时,按以下顺序识别历史初始化: + +1. 读取 legacy marker 中的 `morty`、`meeseeks` ID。 +2. 校验 marker 指向的 Agent 是否属于当前租户;查询包含已删除和 stopped 行。 +3. 如果 marker 无法使用,则查询当前租户所有历史 Agent 行,包括已删除和 stopped 行,查找曾存在的 canonical 名称 Morty/Meeseeks。 +4. 发现任一可信历史证据,就写入 `bootstrap:default_agents:v1`,`source` 分别记录为 `legacy_marker` 或 `database_history`,不创建缺失 Agent。 +5. 只有完全没有数据库标记、有效 legacy marker 和历史 Agent 证据时,才执行首次创建。 + +这里采用保守策略:有历史证据时宁可不自动创建,也不能覆盖用户删除意图。 + +### 6.3 无法完全恢复的历史状态 + +如果现有部署同时满足以下条件: + +- legacy marker 已丢失; +- 默认 Agent 已被重命名; +- 数据库中没有可识别的 canonical 名称历史; +- 数据库初始化标记尚未建立; + +系统无法只根据现有数据可靠证明该 Agent 曾由默认 seeder 创建。不得通过角色描述、Bio 或 workspace 内容做模糊猜测。 + +该极端状态只能通过运维确认后补写租户设置。修复上线后,新数据库标记会消除后续同类歧义。 + +### 6.4 已被旧逻辑重新创建的 Agent + +升级兼容过程不自动删除当前活跃 Agent。系统无法可靠判断用户是否已经开始使用旧逻辑重新创建出的对象。 + +用户可在修复上线后再次删除该 Agent;数据库初始化事实已经建立,后续不会再次创建。 + +## 7. 代码改动范围 + +### 7.1 `backend/app/services/agent_seeder.py` + +- 引入 `TenantSetting`。 +- 增加默认 Agent 设置 key 和 value 解析函数。 +- 增加 legacy marker 解析和懒回填函数。 +- 增加租户级 transaction advisory lock。 +- 将 `seed_default_agents()` 拆为: + - 初始化事实解析; + - 首次创建; + - 现存 Agent 存储修复。 +- 移除以 `name + status != stopped` 作为创建判据的逻辑。 +- 保留 `_repair_default_agent_storage()` 的非覆盖语义。 +- legacy marker 改为追加/合并写入,避免覆盖其他 seeder 条目。 + +### 7.2 `backend/tests/test_agent_seeder_storage_repair.py` + +扩展现有测试覆盖初始化状态、兼容回填和删除语义。 + +不需要修改前端、Agent 删除接口或数据库结构。 + +## 8. 测试设计 + +### 8.1 首次创建 + +- 没有设置、marker 和历史 Agent 时创建两个默认 Agent。 +- 创建与租户初始化设置在同一事务提交。 +- 初始化失败时不留下 `initialized=true`。 + +### 8.2 已初始化 + +- 两个 Agent 都存在:不创建,继续执行存储健康检查。 +- Morty 已删除:不创建 Morty,只检查未删除的 Meeseeks。 +- 两个都已删除:不创建,也不修复存储。 +- Agent 仅 stopped、未删除:不创建副本,仍允许存储修复。 +- Agent 已重命名:按 ID 识别,不创建 canonical 名称副本。 +- 设置中的 Agent ID 已不存在:不创建。 + +### 8.3 兼容回填 + +- legacy marker 有效:写入租户设置,不创建。 +- marker 指向已删除 Agent:仍视为已初始化,不创建。 +- marker 缺失但数据库存在历史 canonical Agent:写入租户设置,不创建。 +- marker 来自其他租户或格式损坏:忽略 marker,继续数据库历史判断。 +- 完全没有历史证据:执行首次创建。 + +### 8.4 并发 + +- 两个 seeder 并发进入时,只有锁内第一个流程可以创建。 +- 第二个流程取得锁后重新读取设置并进入已初始化分支。 + +### 8.5 回归验证 + +- 运行 `backend/tests/test_agent_seeder_storage_repair.py`。 +- 运行与 Agent 删除、列表可见性相关的 scoped tests。 +- 对修改文件运行 Ruff。 +- 验证现有存储漂移修复测试继续通过。 + +## 9. 验收标准 + +- 新租户仍自动获得 Morty 和 Meeseeks。 +- 删除任一默认 Agent 后,连续重启两次均不出现新副本。 +- 重命名任一默认 Agent 后,连续重启两次均不出现 canonical 名称副本。 +- stop 后重启不产生副本。 +- 未删除默认 Agent 的 workspace/Skills 丢失后仍能被修复。 +- 现有部署无需人工脚本即可自动建立数据库初始化事实。 +- 多实例同时启动不会重复创建默认 Agent。 + +## 10. 非目标与风险 + +- 本次不自动清理旧版本已经创建的重复 Agent。 +- 本次不改变普通 Agent 的删除、停止或重命名接口。 +- 本次不把 Morty/Meeseeks 改成不可删除的 system Agent。 +- 本次不以名称、Bio、角色描述等可变内容作为长期身份。 +- legacy marker 丢失且历史 Agent 已重命名的极端部署,需要运维确认;不做推测性自动修复。 + +## 11. 实施顺序 + +1. 先补删除、停止、重命名和 legacy 回填的失败测试。 +2. 增加租户初始化设置和兼容解析函数。 +3. 加入租户级并发锁。 +4. 拆分首次创建与存储修复路径。 +5. 运行 scoped tests 和 Ruff。 +6. 使用本地数据库验证首次初始化与删除后重启。 +7. 部署前检查目标环境当前 marker、历史默认 Agent 行和重复 Agent 状态,不自动清理数据。