"""Focused tests for the Agent Soul-backed config service.""" from __future__ import annotations import io import zipfile from collections.abc import Callable from datetime import datetime from types import SimpleNamespace from unittest.mock import patch import pytest from sqlalchemy.orm import Session, sessionmaker from extensions.storage.storage_type import StorageType from models.agent import ( Agent, AgentConfigDraft, AgentConfigDraftType, AgentConfigSnapshot, AgentScope, AgentSource, ) from models.agent_config_entities import ( AgentConfigFileRefConfig, AgentConfigSkillRefConfig, AgentEnvVariableConfig, AgentFileRefConfig, AgentSoulConfig, ) from models.enums import CreatorUserRole from models.model import UploadFile from models.tools import ToolFile from services.agent.skill_package_service import SkillPackageError from services.agent_config_service import ( AgentConfigService, AgentConfigServiceError, AgentConfigTarget, AgentConfigVersionKind, ConfigDownloadRequest, ConfigPushPayload, ConfigPushSkillItem, ) MODULE = "services.agent_config_service" TENANT = "11111111-1111-1111-1111-111111111111" OTHER_TENANT = "22222222-2222-2222-2222-222222222222" AGENT = "33333333-3333-3333-3333-333333333333" USER = "44444444-4444-4444-4444-444444444444" END_USER = "55555555-5555-5555-5555-555555555555" SNAPSHOT = "66666666-6666-6666-6666-666666666666" DRAFT = "77777777-7777-7777-7777-777777777777" BUILD_DRAFT = "88888888-8888-8888-8888-888888888888" TOOL_FILE = "99999999-9999-9999-9999-999999999999" SKILL_FILE = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa" UPLOAD_FILE = "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb" NORMALIZED_SKILL_FILE = "cccccccc-cccc-cccc-cccc-cccccccccccc" AGENT_CONFIG_TABLES = (Agent, AgentConfigDraft, AgentConfigSnapshot) def _soul(**updates) -> AgentSoulConfig: payload = AgentSoulConfig().model_dump(mode="json") payload.update(updates) return AgentSoulConfig.model_validate(payload) def _agent(*, tenant_id: str = TENANT) -> Agent: return Agent( id=AGENT, tenant_id=tenant_id, name="Config Agent", scope=AgentScope.ROSTER, source=AgentSource.ROSTER, ) def _draft( *, version_id: str = DRAFT, draft_type: AgentConfigDraftType = AgentConfigDraftType.DRAFT, account_id: str | None = None, soul: AgentSoulConfig | None = None, ) -> AgentConfigDraft: return AgentConfigDraft( id=version_id, tenant_id=TENANT, agent_id=AGENT, draft_type=draft_type, account_id=account_id, draft_owner_key=account_id or "", config_snapshot=soul or _soul(), ) def _snapshot(*, soul: AgentSoulConfig | None = None) -> AgentConfigSnapshot: return AgentConfigSnapshot( id=SNAPSHOT, tenant_id=TENANT, agent_id=AGENT, version=1, config_snapshot=soul or _soul(), ) def _service(sqlite_session: Session) -> AgentConfigService: """Bind service-owned sessions to the current test's isolated SQLite engine.""" return AgentConfigService( session_factory=sessionmaker(bind=sqlite_session.get_bind(), expire_on_commit=False), ) def _persist_target( sqlite_session: Session, version: AgentConfigDraft | AgentConfigSnapshot, *, tenant_id: str = TENANT, ) -> None: sqlite_session.add_all([_agent(tenant_id=tenant_id), version]) sqlite_session.commit() def _version(*, version_id: str = "version-1", snapshot: AgentSoulConfig | None = None) -> SimpleNamespace: agent_soul = snapshot or _soul() return SimpleNamespace( id=version_id, config_snapshot_dict=agent_soul.model_dump(mode="json"), config_snapshot=agent_soul, ) def _target( *, kind: AgentConfigVersionKind, writable: bool, version_id: str = "version-1", soul: AgentSoulConfig | None = None, ) -> AgentConfigTarget: agent_soul = soul or _soul() return AgentConfigTarget( tenant_id=TENANT, agent_id=AGENT, version_id=version_id, kind=kind, writable=writable, version=_version(version_id=version_id, snapshot=agent_soul), agent_soul=agent_soul, ) def _zip_bytes(members: dict[str, bytes]) -> bytes: buffer = io.BytesIO() with zipfile.ZipFile(buffer, "w") as archive: for name, payload in members.items(): zip_info = zipfile.ZipInfo(filename=name) zip_info.date_time = (1980, 1, 1, 0, 0, 0) archive.writestr(zip_info, payload) return buffer.getvalue() @pytest.mark.parametrize( ("kind", "user_id", "version_id", "expected_writable"), [ (AgentConfigVersionKind.SNAPSHOT, None, SNAPSHOT, False), (AgentConfigVersionKind.DRAFT, USER, DRAFT, False), (AgentConfigVersionKind.BUILD_DRAFT, USER, BUILD_DRAFT, True), ], ) @pytest.mark.parametrize("sqlite_session", [AGENT_CONFIG_TABLES], indirect=True) def test_resolve_target_supports_snapshot_draft_and_build_draft( kind: AgentConfigVersionKind, user_id: str | None, version_id: str, expected_writable: bool, sqlite_session: Session, ) -> None: if kind == AgentConfigVersionKind.SNAPSHOT: version = _snapshot() else: version = _draft( version_id=version_id, draft_type=( AgentConfigDraftType.DEBUG_BUILD if kind == AgentConfigVersionKind.BUILD_DRAFT else AgentConfigDraftType.DRAFT ), account_id=USER if kind == AgentConfigVersionKind.BUILD_DRAFT else None, ) _persist_target(sqlite_session, version) target = _service(sqlite_session).resolve_target( tenant_id=TENANT, agent_id=AGENT, config_version_id=version_id, config_version_kind=kind, user_id=user_id, ) assert target.agent_id == AGENT assert target.version_id == version_id assert target.kind == kind assert target.writable is expected_writable assert target.agent_soul == _soul() @pytest.mark.parametrize("sqlite_session", [AGENT_CONFIG_TABLES], indirect=True) def test_resolve_target_requires_user_for_build_draft(sqlite_session: Session) -> None: _persist_target( sqlite_session, _draft( version_id=BUILD_DRAFT, draft_type=AgentConfigDraftType.DEBUG_BUILD, account_id=USER, ), ) with pytest.raises(AgentConfigServiceError, match="user_id is required") as exc_info: _service(sqlite_session).resolve_target( tenant_id=TENANT, agent_id=AGENT, config_version_id=BUILD_DRAFT, config_version_kind=AgentConfigVersionKind.BUILD_DRAFT, ) assert exc_info.value.code == "missing_user_id" @pytest.mark.parametrize("sqlite_session", [AGENT_CONFIG_TABLES], indirect=True) def test_resolve_target_hides_build_draft_from_another_user(sqlite_session: Session) -> None: _persist_target( sqlite_session, _draft( version_id=BUILD_DRAFT, draft_type=AgentConfigDraftType.DEBUG_BUILD, account_id=USER, ), ) with pytest.raises(AgentConfigServiceError) as exc_info: _service(sqlite_session).resolve_target( tenant_id=TENANT, agent_id=AGENT, config_version_id=BUILD_DRAFT, config_version_kind=AgentConfigVersionKind.BUILD_DRAFT, user_id=END_USER, ) assert (exc_info.value.code, exc_info.value.status_code) == ("config_version_not_found", 404) @pytest.mark.parametrize( ("agent_tenant_id", "expected_code"), [ (OTHER_TENANT, "agent_not_found"), (TENANT, "config_version_not_found"), ], ) @pytest.mark.parametrize("sqlite_session", [AGENT_CONFIG_TABLES], indirect=True) def test_resolve_target_maps_missing_agent_and_version( agent_tenant_id: str, expected_code: str, sqlite_session: Session, ) -> None: sqlite_session.add(_agent(tenant_id=agent_tenant_id)) sqlite_session.commit() with pytest.raises(AgentConfigServiceError) as exc_info: _service(sqlite_session).resolve_target( tenant_id=TENANT, agent_id=AGENT, config_version_id=SNAPSHOT, config_version_kind=AgentConfigVersionKind.SNAPSHOT, user_id=USER, ) assert exc_info.value.code == expected_code @pytest.mark.parametrize("sqlite_session", [AGENT_CONFIG_TABLES], indirect=True) def test_push_rejects_non_build_draft_writes(sqlite_session: Session) -> None: _persist_target(sqlite_session, _draft(soul=_soul(config_note="before"))) with pytest.raises(AgentConfigServiceError, match="build drafts") as exc_info: _service(sqlite_session).push( tenant_id=TENANT, agent_id=AGENT, user_id=USER, config_version_id=DRAFT, config_version_kind=AgentConfigVersionKind.DRAFT, payload=ConfigPushPayload(note="ignored"), ) assert exc_info.value.code == "config_not_writable" sqlite_session.expire_all() persisted = sqlite_session.get(AgentConfigDraft, DRAFT) assert persisted is not None assert persisted.config_snapshot.config_note == "before" @pytest.mark.parametrize("sqlite_session", [AGENT_CONFIG_TABLES], indirect=True) def test_push_for_console_allows_shared_draft_mutations(sqlite_session: Session) -> None: _persist_target(sqlite_session, _draft(soul=_soul(config_note="before"))) manifest = _service(sqlite_session).push_for_console( tenant_id=TENANT, agent_id=AGENT, user_id=USER, config_version_id=DRAFT, config_version_kind=AgentConfigVersionKind.DRAFT, payload=ConfigPushPayload(note="after"), ) assert manifest["note"] == "after" sqlite_session.expire_all() persisted = sqlite_session.get(AgentConfigDraft, DRAFT) assert persisted is not None assert persisted.config_snapshot.config_note == "after" @pytest.mark.parametrize("sqlite_session", [(*AGENT_CONFIG_TABLES, ToolFile)], indirect=True) def test_push_accepts_tenant_scoped_tool_file_sources_from_different_upload_owner( sqlite_session: Session, ) -> None: _persist_target( sqlite_session, _draft( version_id=BUILD_DRAFT, draft_type=AgentConfigDraftType.DEBUG_BUILD, account_id=USER, ), ) file_source = ToolFile( tenant_id=TENANT, user_id=END_USER, conversation_id=None, size=7, mimetype="text/plain", file_key="file-key", name="guide.txt", ) file_source.id = TOOL_FILE skill_source = ToolFile( tenant_id=TENANT, user_id=END_USER, conversation_id=None, size=123, mimetype="application/zip", file_key="skill-key", name="alpha.zip", ) skill_source.id = SKILL_FILE sqlite_session.add_all([file_source, skill_source]) sqlite_session.commit() skill_ref = AgentConfigSkillRefConfig( name="alpha", description="Alpha skill", file_id=NORMALIZED_SKILL_FILE, size=321, mime_type="application/zip", ) service = _service(sqlite_session) with ( patch(f"{MODULE}.storage.load_once", return_value=b"skill-archive"), patch.object( service._skill_normalizer, "normalize", return_value=(skill_ref, object()), ), ): manifest = service.push( tenant_id=TENANT, agent_id=AGENT, user_id=USER, config_version_id=BUILD_DRAFT, config_version_kind=AgentConfigVersionKind.BUILD_DRAFT, payload=ConfigPushPayload.model_validate( { "files": [{"name": "guide.txt", "file_ref": {"kind": "tool_file", "id": TOOL_FILE}}], "skills": [{"name": "alpha", "file_ref": {"kind": "tool_file", "id": SKILL_FILE}}], } ), ) files = manifest["files"] skills = manifest["skills"] assert isinstance(files, dict) assert isinstance(skills, dict) assert files["items"][0]["file_id"] == TOOL_FILE assert skills["items"][0]["file_id"] == NORMALIZED_SKILL_FILE sqlite_session.expire_all() persisted = sqlite_session.get(AgentConfigDraft, BUILD_DRAFT) assert persisted is not None assert persisted.config_snapshot.config_files[0].file_id == TOOL_FILE assert persisted.config_snapshot.config_skills[0].file_id == NORMALIZED_SKILL_FILE persisted_source = sqlite_session.get(ToolFile, TOOL_FILE) assert persisted_source is not None assert persisted_source.user_id == END_USER @pytest.mark.parametrize("sqlite_session", [(*AGENT_CONFIG_TABLES, ToolFile)], indirect=True) def test_request_download_signs_config_tool_files_without_rechecking_end_user_owner( sqlite_session: Session, ) -> None: soul = _soul( config_files=[AgentConfigFileRefConfig(name="guide.txt", file_kind="tool_file", file_id=TOOL_FILE)], config_skills=[AgentConfigSkillRefConfig(name="alpha", file_id=SKILL_FILE)], ) _persist_target( sqlite_session, _draft( version_id=BUILD_DRAFT, draft_type=AgentConfigDraftType.DEBUG_BUILD, account_id=USER, soul=soul, ), ) guide = ToolFile( tenant_id=TENANT, user_id=END_USER, conversation_id=None, size=7, mimetype="text/plain", file_key="tool-files/guide.txt", name="guide.txt", ) guide.id = TOOL_FILE skill = ToolFile( tenant_id=TENANT, user_id=END_USER, conversation_id=None, size=123, mimetype="application/zip", file_key="tool-files/alpha.zip", name="alpha.zip", ) skill.id = SKILL_FILE sqlite_session.add_all([guide, skill]) sqlite_session.commit() service = _service(sqlite_session) file_download = service.request_download( tenant_id=TENANT, agent_id=AGENT, config_version_id=BUILD_DRAFT, config_version_kind=AgentConfigVersionKind.BUILD_DRAFT, kind="file", name="guide.txt", user_id=USER, ) skill_download = service.request_download( tenant_id=TENANT, agent_id=AGENT, config_version_id=BUILD_DRAFT, config_version_kind=AgentConfigVersionKind.BUILD_DRAFT, kind="skill", name="alpha", user_id=USER, ) assert (file_download.filename, file_download.mime_type, file_download.size) == ("guide.txt", "text/plain", 7) assert file_download.download_uri.startswith(f"/files/tools/{TOOL_FILE}.txt?") assert "as_attachment=true" in file_download.download_uri assert (skill_download.filename, skill_download.mime_type, skill_download.size) == ( "alpha.zip", "application/zip", 123, ) assert skill_download.download_uri.startswith(f"/files/tools/{SKILL_FILE}.zip?") @pytest.mark.parametrize("sqlite_session", [(*AGENT_CONFIG_TABLES, UploadFile)], indirect=True) def test_request_download_signs_tenant_scoped_config_upload_file(sqlite_session: Session) -> None: soul = _soul( config_files=[AgentConfigFileRefConfig(name="guide.txt", file_kind="upload_file", file_id=UPLOAD_FILE)] ) _persist_target(sqlite_session, _draft(soul=soul)) upload_file = UploadFile( tenant_id=TENANT, storage_type=StorageType.LOCAL, key="uploads/guide.txt", name="source-name.txt", size=7, extension="txt", mime_type="text/plain", created_by_role=CreatorUserRole.END_USER, created_by=END_USER, created_at=datetime(2025, 1, 1), used=False, ) upload_file.id = UPLOAD_FILE sqlite_session.add(upload_file) sqlite_session.commit() result = _service(sqlite_session).request_download( tenant_id=TENANT, agent_id=AGENT, config_version_id=DRAFT, config_version_kind=AgentConfigVersionKind.DRAFT, kind="file", name="guide.txt", user_id=USER, ) assert (result.filename, result.mime_type, result.size) == ("guide.txt", "text/plain", 7) assert result.download_uri.startswith(f"/files/{UPLOAD_FILE}/file-preview?") assert "as_attachment=true" in result.download_uri @pytest.mark.parametrize("sqlite_session", [(*AGENT_CONFIG_TABLES, ToolFile)], indirect=True) def test_request_download_rejects_config_source_from_another_tenant(sqlite_session: Session) -> None: soul = _soul(config_files=[AgentConfigFileRefConfig(name="guide.txt", file_kind="tool_file", file_id=TOOL_FILE)]) _persist_target(sqlite_session, _draft(soul=soul)) source = ToolFile( tenant_id=OTHER_TENANT, user_id=END_USER, conversation_id=None, size=7, mimetype="text/plain", file_key="tool-files/guide.txt", name="guide.txt", ) source.id = TOOL_FILE sqlite_session.add(source) sqlite_session.commit() with pytest.raises(AgentConfigServiceError) as exc_info: _service(sqlite_session).request_download( tenant_id=TENANT, agent_id=AGENT, config_version_id=DRAFT, config_version_kind=AgentConfigVersionKind.DRAFT, kind="file", name="guide.txt", user_id=USER, ) assert (exc_info.value.code, exc_info.value.status_code) == ("config_file_not_found", 404) @pytest.mark.parametrize("sqlite_session", [AGENT_CONFIG_TABLES], indirect=True) def test_push_file_for_console_rejects_snapshot_writes(sqlite_session: Session) -> None: _persist_target(sqlite_session, _snapshot()) with pytest.raises(AgentConfigServiceError, match="editable drafts") as exc_info: _service(sqlite_session).push_file_for_console( tenant_id=TENANT, agent_id=AGENT, user_id=USER, config_version_id=SNAPSHOT, config_version_kind=AgentConfigVersionKind.SNAPSHOT, upload_file_id=UPLOAD_FILE, ) assert exc_info.value.code == "config_not_writable" sqlite_session.expire_all() persisted = sqlite_session.get(AgentConfigSnapshot, SNAPSHOT) assert persisted is not None assert persisted.config_snapshot.config_files == [] @pytest.mark.parametrize("sqlite_session", [(*AGENT_CONFIG_TABLES, UploadFile)], indirect=True) def test_push_file_for_console_uses_service_owned_upload_lookup_and_naming(sqlite_session: Session) -> None: _persist_target(sqlite_session, _draft()) upload_file = UploadFile( tenant_id=TENANT, storage_type=StorageType.LOCAL, key="uploads/guide.txt", name="guide.txt", size=7, extension="txt", mime_type="text/plain", created_by_role=CreatorUserRole.ACCOUNT, created_by=USER, created_at=datetime(2025, 1, 1), used=False, hash="sha256:abc", ) upload_file.id = UPLOAD_FILE sqlite_session.add(upload_file) sqlite_session.commit() response = _service(sqlite_session).push_file_for_console( tenant_id=TENANT, agent_id=AGENT, user_id=USER, config_version_id=DRAFT, config_version_kind=AgentConfigVersionKind.DRAFT, upload_file_id=UPLOAD_FILE, ) assert response == { "file": { "id": "guide.txt", "name": "guide.txt", "file_id": UPLOAD_FILE, "is_missing": False, "size": 7, "hash": "sha256:abc", "mime_type": "text/plain", }, "config_version": { "id": DRAFT, "kind": "draft", "writable": True, }, } sqlite_session.expire_all() persisted = sqlite_session.get(AgentConfigDraft, DRAFT) assert persisted is not None assert persisted.config_snapshot.config_files[0].file_id == UPLOAD_FILE @pytest.mark.parametrize("sqlite_session", [AGENT_CONFIG_TABLES], indirect=True) def test_upload_skill_for_console_maps_package_validation_failures(sqlite_session: Session) -> None: _persist_target(sqlite_session, _draft()) service = _service(sqlite_session) message = "skill package must contain exactly one skill; multiple skill folders in one archive are not supported" with patch.object( service._skill_normalizer, "normalize", side_effect=SkillPackageError("files_outside_skill_root", message, status_code=400), ): with pytest.raises(AgentConfigServiceError, match="exactly one skill") as exc_info: service.upload_skill_for_console( tenant_id=TENANT, agent_id=AGENT, user_id=USER, config_version_id=DRAFT, config_version_kind=AgentConfigVersionKind.DRAFT, content=b"bad-archive", filename="skills.zip", ) assert exc_info.value.code == "files_outside_skill_root" assert exc_info.value.message == message assert exc_info.value.status_code == 400 sqlite_session.expire_all() persisted = sqlite_session.get(AgentConfigDraft, DRAFT) assert persisted is not None assert persisted.config_snapshot.config_skills == [] @pytest.mark.parametrize("sqlite_session", [()], indirect=True) def test_apply_skill_updates_rejects_non_tool_file_refs(sqlite_session: Session) -> None: service = AgentConfigService() with pytest.raises(AgentConfigServiceError, match="tool files") as exc_info: service._apply_skill_updates( sqlite_session, tenant_id=TENANT, user_id=USER, current=[], updates=[ ConfigPushSkillItem.model_validate( {"name": "alpha", "file_ref": {"kind": "upload_file", "id": "upload-1"}} ) ], ) assert exc_info.value.code == "invalid_skill_file_ref" @pytest.mark.parametrize( ("error_code", "message"), [ ("skill_name_mismatch", "skill name does not match requested config key"), ("invalid_archive", "stored tool file is not a valid skill archive"), ], ) @pytest.mark.parametrize("sqlite_session", [()], indirect=True) def test_apply_skill_updates_maps_normalizer_failures( error_code: str, message: str, sqlite_session: Session, ) -> None: service = AgentConfigService() tool_file = ToolFile( user_id=USER, tenant_id=TENANT, conversation_id=None, file_key="tool-files/alpha.zip", mimetype="application/zip", name="alpha.zip", size=1, ) with ( patch.object(service, "_require_tool_file_source", return_value=tool_file), patch(f"{MODULE}.storage.load_once", return_value=b"bad-archive"), patch.object( service._skill_normalizer, "normalize", side_effect=SkillPackageError(error_code, message, status_code=400), ), ): with pytest.raises(AgentConfigServiceError, match=message) as exc_info: service._apply_skill_updates( sqlite_session, tenant_id=TENANT, user_id=USER, current=[], updates=[ ConfigPushSkillItem.model_validate( {"name": "alpha", "file_ref": {"kind": "tool_file", "id": "tool-file-1"}} ) ], ) assert exc_info.value.code == error_code def test_apply_env_text_supports_delete_comments_export_and_keeps_unmentioned_values() -> None: current = [ AgentEnvVariableConfig(key="KEEP", name="KEEP", value="old"), AgentEnvVariableConfig(key="REMOVE", name="REMOVE", value="gone"), AgentEnvVariableConfig(key="UNTOUCHED", name="UNTOUCHED", value="still-here"), ] updated = AgentConfigService._apply_env_text( current, "# comment\nexport KEEP=new-value\nREMOVE=\nNEW='two words'\n", ) values = {item.key: item.value for item in updated} assert values == { "KEEP": "new-value", "UNTOUCHED": "still-here", "NEW": "two words", } @pytest.mark.parametrize( "archive_bytes", [ pytest.param(b"not-a-zip-archive", id="not-a-zip-archive"), pytest.param(_zip_bytes({"README.md": b"missing skill md"}), id="missing-skill-md"), ], ) def test_inspect_skill_maps_invalid_archives_to_service_errors(archive_bytes: bytes) -> None: service = AgentConfigService() target = _target( kind=AgentConfigVersionKind.BUILD_DRAFT, writable=True, soul=_soul(config_skills=[AgentConfigSkillRefConfig(name="alpha", file_id="tool-file-1")]), ) with ( patch.object(service, "resolve_target", return_value=target), patch.object(service, "_load_tool_file_bytes", return_value=(archive_bytes, "application/zip")), ): with pytest.raises(AgentConfigServiceError, match="stored config skill archive is invalid") as exc_info: service.inspect_skill( tenant_id=TENANT, agent_id=AGENT, config_version_id="build-draft-1", config_version_kind=AgentConfigVersionKind.BUILD_DRAFT, name="alpha", user_id=USER, ) assert exc_info.value.code == "skill_archive_invalid" assert exc_info.value.status_code == 500 def test_request_download_falls_back_to_workspace_runtime_skill() -> None: service = AgentConfigService() target = _target(kind=AgentConfigVersionKind.DRAFT, writable=False, soul=_soul(config_skills=[])) expected = SimpleNamespace( filename="workspace-skill.zip", mime_type="application/zip", size=123, download_uri="/files/tools/workspace.zip?signature=1", ) with ( patch.object(service, "resolve_target", return_value=target), patch(f"{MODULE}.SkillManagementService") as skill_management_service, patch.object(service, "_resolve_download_request", return_value=expected) as resolve_download_request, ): skill_management_service.return_value.list_runtime_agent_skills.return_value = [ { "name": "workspace-skill", "file_id": "workspace-archive-id", } ] download = service.request_download( tenant_id=TENANT, agent_id=AGENT, config_version_id="draft-1", config_version_kind=AgentConfigVersionKind.DRAFT, kind="skill", name="workspace-skill", user_id=USER, ) assert download is expected resolve_download_request.assert_called_once_with( tenant_id=TENANT, file_kind="tool_file", file_id="workspace-archive-id", filename="workspace-skill.zip", default_mime_type="application/zip", missing_code="config_skill_not_found", missing_message="config skill payload is missing", ) def test_inspect_skill_falls_back_to_workspace_runtime_skill() -> None: service = AgentConfigService() target = _target(kind=AgentConfigVersionKind.DRAFT, writable=False, soul=_soul(config_skills=[])) archive = _zip_bytes( { "SKILL.md": b"---\nname: workspace-skill\ndescription: Workspace skill.\n---\n# Workspace", "references/policy.md": b"Policy", } ) with ( patch.object(service, "resolve_target", return_value=target), patch(f"{MODULE}.SkillManagementService") as skill_management_service, ): skill_management_service.return_value.pull_runtime_agent_skill.return_value = SimpleNamespace(payload=archive) skill_management_service.return_value.list_runtime_agent_skills.return_value = [ { "id": "skill-1", "name": "workspace-skill", "description": "Workspace skill.", "size": len(archive), "hash": "hash", "mime_type": "application/zip", } ] result = service.inspect_skill( tenant_id=TENANT, agent_id=AGENT, config_version_id="draft-1", config_version_kind=AgentConfigVersionKind.DRAFT, name="workspace-skill", user_id=USER, ) assert result["id"] == "skill-1" assert result["source"] == "config_skill_zip" assert result["skill_md"]["text"] == "---\nname: workspace-skill\ndescription: Workspace skill.\n---\n# Workspace" assert [item["path"] for item in result["files"]] == ["SKILL.md", "references", "references/policy.md"] def test_manifest_uses_items_shape_without_download_urls() -> None: target = _target( kind=AgentConfigVersionKind.DRAFT, writable=False, soul=_soul( config_skills=[AgentConfigSkillRefConfig(name="alpha", description="Alpha skill", file_id="tool-file-1")], config_files=[AgentConfigFileRefConfig(name="guide.txt", file_kind="upload_file", file_id="upload-file-1")], config_note="Use the guide.", ), ) with patch(f"{MODULE}.SkillManagementService") as skill_management_service: skill_management_service.return_value.list_runtime_agent_skills.return_value = [] manifest = AgentConfigService._manifest_for_target(target) assert manifest == { "agent_id": AGENT, "config_version": { "id": "version-1", "kind": "draft", "writable": True, }, "skills": { "items": [ { "id": "alpha", "name": "alpha", "file_id": "tool-file-1", "is_missing": False, "description": "Alpha skill", "size": None, "hash": None, "mime_type": "application/zip", } ] }, "files": { "items": [ { "id": "guide.txt", "name": "guide.txt", "file_id": "upload-file-1", "is_missing": False, "size": None, "hash": None, "mime_type": None, } ] }, "env_keys": [], "note": "Use the guide.", } @pytest.mark.parametrize("sqlite_session", [AGENT_CONFIG_TABLES], indirect=True) def test_manifest_preserves_missing_config_assets_and_download_rejects_them(sqlite_session: Session) -> None: soul = _soul( config_skills=[{"name": "alpha", "file_id": "", "is_missing": True}], config_files=[{"name": "guide.txt", "file_kind": "upload_file", "file_id": "", "is_missing": True}], ) _persist_target(sqlite_session, _draft(soul=soul)) service = _service(sqlite_session) target = service.resolve_target( tenant_id=TENANT, agent_id=AGENT, config_version_id=DRAFT, config_version_kind=AgentConfigVersionKind.DRAFT, user_id=USER, ) with patch(f"{MODULE}.SkillManagementService") as skill_management_service: skill_management_service.return_value.list_runtime_agent_skills.return_value = [] manifest = service._manifest_for_target(target) assert manifest["skills"]["items"][0]["is_missing"] is True # type: ignore[index] assert manifest["files"]["items"][0]["is_missing"] is True # type: ignore[index] with pytest.raises(AgentConfigServiceError) as skill_error: service.request_download( tenant_id=TENANT, agent_id=AGENT, config_version_id=DRAFT, config_version_kind=AgentConfigVersionKind.DRAFT, kind="skill", name="alpha", user_id=USER, ) with pytest.raises(AgentConfigServiceError) as file_error: service.request_download( tenant_id=TENANT, agent_id=AGENT, config_version_id=DRAFT, config_version_kind=AgentConfigVersionKind.DRAFT, kind="file", name="guide.txt", user_id=USER, ) assert (skill_error.value.code, skill_error.value.status_code) == ("config_skill_missing", 409) assert (file_error.value.code, file_error.value.status_code) == ("config_file_missing", 409) def test_config_asset_refs_require_file_id_unless_marked_missing() -> None: assert AgentFileRefConfig().file_id is None assert ( AgentConfigFileRefConfig( name="guide.txt", file_kind="upload_file", is_missing=True, ).file_id == "" ) with pytest.raises(ValueError, match="file_id is required"): AgentConfigSkillRefConfig(name="alpha") with pytest.raises(ValueError, match="must not retain"): AgentConfigFileRefConfig( name="guide.txt", file_kind="upload_file", file_id="workspace-file-id", is_missing=True, ) def test_manifest_appends_published_workspace_skills() -> None: target = _target( kind=AgentConfigVersionKind.DRAFT, writable=False, soul=_soul( config_skills=[AgentConfigSkillRefConfig(name="alpha", description="Alpha skill", file_id="tool-file-1")] ), ) with patch(f"{MODULE}.SkillManagementService") as skill_management_service: skill_management_service.return_value.list_runtime_agent_skills.return_value = [ { "id": "workspace-skill-id", "name": "beta", "file_id": "tool-file-2", "description": "Beta workspace skill", "size": 123, "hash": "sha256:beta", "mime_type": "application/zip", }, { "id": "duplicate", "name": "alpha", "file_id": "tool-file-ignored", "description": "Duplicate workspace skill", "size": 456, "hash": "sha256:ignored", "mime_type": "application/zip", }, ] manifest = AgentConfigService._manifest_for_target(target) assert [item["name"] for item in manifest["skills"]["items"]] == ["alpha", "beta"] assert manifest["skills"]["items"][1]["file_id"] == "tool-file-2" def test_list_skills_excludes_workspace_skill_bindings() -> None: target = _target( kind=AgentConfigVersionKind.DRAFT, writable=False, soul=_soul( config_skills=[AgentConfigSkillRefConfig(name="alpha", description="Alpha skill", file_id="tool-file-1")] ), ) service = AgentConfigService() with ( patch.object(service, "resolve_target", return_value=target), patch(f"{MODULE}.SkillManagementService") as skill_management_service, ): skill_management_service.return_value.list_runtime_agent_skills.return_value = [ { "id": "workspace-skill-id", "name": "beta", "file_id": "tool-file-2", "description": "Beta workspace skill", "size": 123, "hash": "sha256:beta", "mime_type": "application/zip", } ] result = service.list_skills( tenant_id=target.tenant_id, agent_id=target.agent_id, config_version_id=target.version_id, config_version_kind=target.kind, user_id=None, ) assert [item["name"] for item in result["items"]] == ["alpha"] skill_management_service.return_value.list_runtime_agent_skills.assert_not_called() def test_preview_skill_file_returns_text_preview() -> None: service = AgentConfigService() target = _target( kind=AgentConfigVersionKind.BUILD_DRAFT, writable=True, soul=_soul(config_skills=[AgentConfigSkillRefConfig(name="alpha", file_id="tool-file-1")]), ) archive_bytes = _zip_bytes( { "SKILL.md": b"# Alpha\n", "references/guide.md": b"hello world", } ) with ( patch.object(service, "resolve_target", return_value=target), patch.object(service, "_load_tool_file_bytes", return_value=(archive_bytes, "application/zip")), ): preview = service.preview_skill_file( tenant_id=TENANT, agent_id=AGENT, config_version_id="build-draft-1", config_version_kind=AgentConfigVersionKind.BUILD_DRAFT, name="alpha", path="references/guide.md", user_id=USER, ) assert preview == { "path": "references/guide.md", "size": 11, "truncated": False, "binary": False, "text": "hello world", } def test_preview_skill_file_marks_binary_and_truncated_payloads() -> None: service = AgentConfigService() target = _target( kind=AgentConfigVersionKind.BUILD_DRAFT, writable=True, soul=_soul(config_skills=[AgentConfigSkillRefConfig(name="alpha", file_id="tool-file-1")]), ) archive_bytes = _zip_bytes( { "SKILL.md": b"# Alpha\n", "bin/data.bin": b"\x00" + (b"x" * (AgentConfigService.PREVIEW_MAX_BYTES + 10)), } ) with ( patch.object(service, "resolve_target", return_value=target), patch.object(service, "_load_tool_file_bytes", return_value=(archive_bytes, "application/zip")), ): preview = service.preview_skill_file( tenant_id=TENANT, agent_id=AGENT, config_version_id="build-draft-1", config_version_kind=AgentConfigVersionKind.BUILD_DRAFT, name="alpha", path="bin/data.bin", user_id=USER, ) assert preview == { "path": "bin/data.bin", "size": AgentConfigService.PREVIEW_MAX_BYTES + 11, "truncated": True, "binary": True, "text": None, } def test_resolve_skill_file_member_path_requires_existing_member() -> None: service = AgentConfigService() target = _target( kind=AgentConfigVersionKind.BUILD_DRAFT, writable=True, soul=_soul(config_skills=[AgentConfigSkillRefConfig(name="alpha", file_id="tool-file-1")]), ) archive_bytes = _zip_bytes( { "SKILL.md": b"# Alpha\n", "references/guide.md": b"hello world", } ) with ( patch.object(service, "resolve_target", return_value=target), patch.object(service, "_load_tool_file_bytes", return_value=(archive_bytes, "application/zip")), ): assert ( service.resolve_skill_file_member_path( tenant_id=TENANT, agent_id=AGENT, config_version_id="build-draft-1", config_version_kind=AgentConfigVersionKind.BUILD_DRAFT, name="alpha", path="references/guide.md", user_id=USER, ) == "references/guide.md" ) with pytest.raises(AgentConfigServiceError, match="config skill file not found") as exc_info: service.resolve_skill_file_member_path( tenant_id=TENANT, agent_id=AGENT, config_version_id="build-draft-1", config_version_kind=AgentConfigVersionKind.BUILD_DRAFT, name="alpha", path="references/missing.md", user_id=USER, ) assert exc_info.value.code == "config_skill_file_not_found" assert exc_info.value.status_code == 404 def test_download_url_helpers_bind_shared_download_request_to_console_origin( config_overrides: Callable[..., None], ) -> None: config_overrides(FILES_URL="https://example.com") service = AgentConfigService() with ( patch.object( service, "request_download", side_effect=[ ConfigDownloadRequest("alpha.zip", "application/zip", 10, "/files/alpha.zip?sign=1"), ConfigDownloadRequest("guide.txt", "text/plain", 20, "/files/guide.txt?sign=2"), ], ), ): assert ( service.download_skill_url( tenant_id=TENANT, agent_id=AGENT, config_version_id="build-draft-1", config_version_kind=AgentConfigVersionKind.BUILD_DRAFT, name="alpha", user_id=USER, ) == "https://example.com/files/alpha.zip?sign=1" ) assert ( service.download_file_url( tenant_id=TENANT, agent_id=AGENT, config_version_id="build-draft-1", config_version_kind=AgentConfigVersionKind.BUILD_DRAFT, name="guide.txt", user_id=USER, ) == "https://example.com/files/guide.txt?sign=2" )