agent-zero/helpers/projects.py
Alessandro a24d9a73a0 Fix project skills folder initialization
Create the .a0proj/skills directory during project metadata setup so Project Settings can open the skills folder for new projects. Repair the folder when loading existing project settings, and cover both creation and repair with regression tests.
2026-06-17 02:37:28 +02:00

673 lines
22 KiB
Python

import os
from typing import Literal, NotRequired, TypedDict, TYPE_CHECKING, cast
from helpers import files, dirty_json, persist_chat, file_tree
from helpers.print_style import PrintStyle
if TYPE_CHECKING:
from agent import AgentContext
PROJECTS_PARENT_DIR = "usr/projects"
PROJECT_META_DIR = ".a0proj"
PROJECT_INSTRUCTIONS_DIR = "instructions"
PROJECT_KNOWLEDGE_DIR = "knowledge"
PROJECT_SKILLS_DIR = "skills"
PROJECT_HEADER_FILE = "project.json"
PROJECT_MCP_SERVERS_FILE = "mcp_servers.json"
PROJECT_AGENTS_MD_FILES = ("AGENTS.md", "Agents.md", "agents.md")
DEFAULT_MCP_SERVERS_CONFIG = '{\n "mcpServers": {}\n}'
CONTEXT_DATA_KEY_PROJECT = "project"
class FileStructureInjectionSettings(TypedDict):
enabled: bool
max_depth: int
max_files: int
max_folders: int
max_lines: int
gitignore: str
class SubAgentSettings(TypedDict):
enabled: bool
class BasicProjectData(TypedDict):
title: str
description: str
instructions: str
include_agents_md: NotRequired[bool]
mcp_servers: NotRequired[str]
color: str
git_url: str
file_structure: FileStructureInjectionSettings
class GitStatusData(TypedDict, total=False):
is_git_repo: bool
remote_url: str
current_branch: str
is_dirty: bool
untracked_count: int
last_commit: dict
error: str
class EditProjectData(BasicProjectData):
name: str
instruction_files_count: int
knowledge_files_count: int
variables: str
secrets: str
mcp_servers: str
subagents: dict[str, SubAgentSettings]
git_status: GitStatusData
def get_projects_parent_folder():
return files.get_abs_path(PROJECTS_PARENT_DIR)
def get_project_folder(name: str):
return files.get_abs_path(get_projects_parent_folder(), name)
def get_project_meta(name: str, *sub_dirs: str):
return files.get_abs_path(get_project_folder(name), PROJECT_META_DIR, *sub_dirs)
def validate_project_name(name: str | None) -> str:
candidate = str(name or "").strip()
if (
not candidate
or candidate in {".", ".."}
or os.path.basename(candidate) != candidate
):
raise ValueError("Invalid project name")
return candidate
def delete_project(name: str):
abs_path = files.get_abs_path(PROJECTS_PARENT_DIR, name)
files.delete_dir(abs_path)
deactivate_project_in_chats(name)
return name
def create_project(name: str, data: BasicProjectData):
llm_data = data.get("llm") if isinstance(data, dict) else None
mcp_servers = data.get("mcp_servers") if isinstance(data, dict) else None
abs_path = files.create_dir_safe(
files.get_abs_path(PROJECTS_PARENT_DIR, name), rename_format="{name}_{number}"
)
create_project_meta_folders(name)
data = _normalizeBasicData(data)
save_project_header(name, data)
save_project_mcp_servers(name, mcp_servers or DEFAULT_MCP_SERVERS_CONFIG)
save_project_llm_settings(name, llm_data)
return name
def clone_git_project(name: str, git_url: str, git_token: str, data: BasicProjectData):
"""Clone a git repository as a new A0 project. Token is used only for cloning via http header."""
from helpers import git
llm_data = data.get("llm") if isinstance(data, dict) else None
mcp_servers = data.get("mcp_servers") if isinstance(data, dict) else None
abs_path = files.create_dir_safe(
files.get_abs_path(PROJECTS_PARENT_DIR, name), rename_format="{name}_{number}"
)
actual_name = files.basename(abs_path)
try:
# Clone with token via http.extraHeader (token never in URL or git config)
git.clone_repo(git_url, abs_path, token=git_token)
clean_url = git.strip_auth_from_url(git_url)
# Check if cloned repo already has .a0proj
meta_path = os.path.join(abs_path, PROJECT_META_DIR, PROJECT_HEADER_FILE)
if os.path.exists(meta_path):
# Merge: keep cloned content, override only user-specified fields
cloned_header: BasicProjectData = dirty_json.parse(files.read_file(meta_path)) # type: ignore
cloned_header["title"] = data.get("title") or cloned_header.get("title", "")
cloned_header["color"] = data.get("color") or cloned_header.get("color", "")
cloned_header["git_url"] = clean_url
save_project_header(actual_name, cloned_header)
else:
# New project: create meta folders and save header
create_project_meta_folders(actual_name)
data = _normalizeBasicData(data)
data["git_url"] = clean_url
save_project_header(actual_name, data)
if mcp_servers:
save_project_mcp_servers(actual_name, mcp_servers)
save_project_llm_settings(actual_name, llm_data)
return actual_name
except Exception as e:
try:
files.delete_dir(abs_path)
except Exception:
pass
raise e
def load_project_header(name: str):
abs_path = files.get_abs_path(
PROJECTS_PARENT_DIR, name, PROJECT_META_DIR, PROJECT_HEADER_FILE
)
header: dict = dirty_json.parse(files.read_file(abs_path)) # type: ignore
header["name"] = name
return header
def _default_file_structure_settings():
try:
gitignore = files.read_file("conf/projects.default.gitignore")
except Exception:
gitignore = ""
return FileStructureInjectionSettings(
enabled=True,
max_depth=5,
max_files=20,
max_folders=20,
max_lines=250,
gitignore=gitignore,
)
def _normalizeBasicData(data: BasicProjectData) -> BasicProjectData:
return {
"title": data.get("title", ""),
"description": data.get("description", ""),
"instructions": data.get("instructions", ""),
"include_agents_md": _normalize_include_agents_md(
data.get("include_agents_md", True)
),
"color": data.get("color", ""),
"git_url": data.get("git_url", ""),
"file_structure": data.get(
"file_structure",
_default_file_structure_settings(),
),
}
def _normalizeEditData(data: EditProjectData) -> EditProjectData:
normalized: EditProjectData = {
"name": data.get("name", ""),
"title": data.get("title", ""),
"description": data.get("description", ""),
"instructions": data.get("instructions", ""),
"include_agents_md": _normalize_include_agents_md(
data.get("include_agents_md", True)
),
"variables": data.get("variables", ""),
"mcp_servers": data.get("mcp_servers", DEFAULT_MCP_SERVERS_CONFIG),
"color": data.get("color", ""),
"git_url": data.get("git_url", ""),
"git_status": data.get("git_status", {"is_git_repo": False}),
"instruction_files_count": data.get("instruction_files_count", 0),
"knowledge_files_count": data.get("knowledge_files_count", 0),
"secrets": data.get("secrets", ""),
"file_structure": data.get(
"file_structure",
_default_file_structure_settings(),
),
"subagents": data.get("subagents", {}),
}
return normalized
def _edit_data_to_basic_data(data: EditProjectData):
return _normalizeBasicData(data)
def _basic_data_to_edit_data(data: BasicProjectData) -> EditProjectData:
base: EditProjectData = cast(
EditProjectData,
{
**data,
"name": "",
"instruction_files_count": 0,
"knowledge_files_count": 0,
"variables": "",
"secrets": "",
"subagents": {},
"git_status": {"is_git_repo": False},
},
)
return _normalizeEditData(base)
def update_project(name: str, data: EditProjectData):
llm_data = data.get("llm") if isinstance(data, dict) else None
# merge with current state
current = load_edit_project_data(name)
current.update(data)
current = _normalizeEditData(current)
# save header data
header = _edit_data_to_basic_data(current)
save_project_header(name, header)
# save secrets
save_project_variables(name, current["variables"])
save_project_secrets(name, current["secrets"])
save_project_mcp_servers(name, current["mcp_servers"])
save_project_subagents(name, current["subagents"])
save_project_llm_settings(name, llm_data)
reactivate_project_in_chats(name)
return name
def load_basic_project_data(name: str) -> BasicProjectData:
data = cast(BasicProjectData, load_project_header(name))
normalized = _normalizeBasicData(data)
return normalized
def load_edit_project_data(name: str) -> EditProjectData:
from helpers import git
data = load_basic_project_data(name)
create_project_meta_folders(name)
additional_instructions = get_additional_instructions_files(name)
variables = load_project_variables(name)
mcp_servers = load_project_mcp_servers(name)
secrets = load_project_secrets_masked(name)
subagents = load_project_subagents(name)
knowledge_files_count = get_knowledge_files_count(name)
git_status = cast(GitStatusData, git.get_repo_status(get_project_folder(name)))
data = cast(
EditProjectData,
{
**data,
"name": name,
"instruction_files_count": len(additional_instructions),
"knowledge_files_count": knowledge_files_count,
"variables": variables,
"mcp_servers": mcp_servers,
"secrets": secrets,
"subagents": subagents,
"git_status": git_status,
},
)
data = _normalizeEditData(data)
data["llm"] = load_project_llm_data(name) # type: ignore[typeddict-unknown-key]
return data
def save_project_header(name: str, data: BasicProjectData):
# save project header file
header = dirty_json.stringify(_project_header_for_save(data))
abs_path = files.get_abs_path(
PROJECTS_PARENT_DIR, name, PROJECT_META_DIR, PROJECT_HEADER_FILE
)
files.write_file(abs_path, header)
def load_project_llm_data(name: str) -> dict:
from plugins._model_config.helpers import model_config
config = model_config.normalize_config_for_save(
model_config.get_config(project_name=name) or {}
)
return {
"config": config,
"selected_preset": {
"scope": "current",
"project_name": name,
"name": "Current config",
},
"presets": model_config.get_combined_presets(name),
"global_presets": model_config.get_presets(),
"project_presets": model_config.get_project_presets(name),
}
def save_project_llm_settings(name: str, llm_data: object):
if not isinstance(llm_data, dict):
return
from helpers import plugins
from plugins._model_config.helpers import model_config
project_presets = llm_data.get("project_presets")
if isinstance(project_presets, list):
model_config.save_presets(project_presets, project_name=name)
config_to_save = None
selected_preset = llm_data.get("selected_preset")
if isinstance(selected_preset, dict) and selected_preset.get("scope") in {
"global",
"project",
}:
preset = model_config.resolve_preset_selection(selected_preset, project_name=name)
if preset:
base_config = model_config.get_config(project_name=name) or {}
config_to_save = model_config.build_config_from_preset(preset, base_config)
config = llm_data.get("config")
if config_to_save is None and isinstance(config, dict):
config_to_save = model_config.normalize_config_for_save(config)
if config_to_save is not None:
plugins.save_plugin_config("_model_config", name, "", config_to_save)
def load_project_mcp_servers(name: str) -> str:
project_name = validate_project_name(name)
try:
return files.read_file(get_project_meta(project_name, PROJECT_MCP_SERVERS_FILE))
except Exception:
return DEFAULT_MCP_SERVERS_CONFIG
def save_project_mcp_servers(name: str, mcp_servers: str):
project_name = validate_project_name(name)
content = mcp_servers if isinstance(mcp_servers, str) else DEFAULT_MCP_SERVERS_CONFIG
files.write_file(get_project_meta(project_name, PROJECT_MCP_SERVERS_FILE), content)
def get_active_projects_list():
return _get_projects_list(get_projects_parent_folder())
def _get_projects_list(parent_dir):
projects = []
# folders in project directory
for name in os.listdir(parent_dir):
try:
abs_path = os.path.join(parent_dir, name)
if os.path.isdir(abs_path):
project_data = load_basic_project_data(name)
projects.append(
{
"name": name,
"title": project_data.get("title", ""),
"description": project_data.get("description", ""),
"color": project_data.get("color", ""),
}
)
except Exception as e:
PrintStyle.error(f"Error loading project {name}: {str(e)}")
# sort projects by name
projects.sort(key=lambda x: x["name"])
return projects
def activate_project(context_id: str, name: str, *, mark_dirty: bool = True):
from agent import AgentContext
data = load_edit_project_data(name)
context = AgentContext.get(context_id)
if context is None:
raise Exception("Context not found")
display_name = str(data.get("title", name))
display_name = display_name[:22] + "..." if len(display_name) > 25 else display_name
context.set_data(CONTEXT_DATA_KEY_PROJECT, name)
context.set_output_data(
CONTEXT_DATA_KEY_PROJECT,
{"name": name, "title": display_name, "color": data.get("color", "")},
)
# persist
persist_chat.save_tmp_chat(context)
if mark_dirty:
from helpers.state_monitor_integration import mark_dirty_all
mark_dirty_all(reason="projects.activate_project")
def deactivate_project(context_id: str, *, mark_dirty: bool = True):
from agent import AgentContext
context = AgentContext.get(context_id)
if context is None:
raise Exception("Context not found")
context.set_data(CONTEXT_DATA_KEY_PROJECT, None)
context.set_output_data(CONTEXT_DATA_KEY_PROJECT, None)
# persist
persist_chat.save_tmp_chat(context)
if mark_dirty:
from helpers.state_monitor_integration import mark_dirty_all
mark_dirty_all(reason="projects.deactivate_project")
def reactivate_project_in_chats(name: str):
from agent import AgentContext
for context in AgentContext.all():
if context.get_data(CONTEXT_DATA_KEY_PROJECT) == name:
activate_project(context.id, name, mark_dirty=False)
persist_chat.save_tmp_chat(context)
from helpers.state_monitor_integration import mark_dirty_all
mark_dirty_all(reason="projects.reactivate_project_in_chats")
def deactivate_project_in_chats(name: str):
from agent import AgentContext
for context in AgentContext.all():
if context.get_data(CONTEXT_DATA_KEY_PROJECT) == name:
deactivate_project(context.id, mark_dirty=False)
persist_chat.save_tmp_chat(context)
from helpers.state_monitor_integration import mark_dirty_all
mark_dirty_all(reason="projects.deactivate_project_in_chats")
def build_system_prompt_vars(name: str):
project_data = load_basic_project_data(name)
main_instructions = project_data.get("instructions", "") or ""
instruction_files = get_project_instruction_files(
name,
include_agents_md=project_data.get("include_agents_md", True),
)
instruction_parts = [
main_instructions,
_format_project_instruction_files(instruction_files),
]
complete_instructions = "\n\n".join(
part.strip() for part in instruction_parts if part.strip()
).strip()
return {
"project_name": project_data.get("title", ""),
"project_description": project_data.get("description", ""),
"project_instructions": complete_instructions or "",
"project_path": files.normalize_a0_path(get_project_folder(name)),
"project_git_url": project_data.get("git_url", ""),
}
def get_additional_instructions_files(name: str):
instructions_folder = files.get_abs_path(
get_project_folder(name), PROJECT_META_DIR, PROJECT_INSTRUCTIONS_DIR
)
return files.read_text_files_in_dir(instructions_folder)
def get_project_instruction_files(
name: str,
include_agents_md: bool = True,
) -> list[tuple[str, str]]:
project_folder = get_project_folder(name)
result: list[tuple[str, str]] = []
if include_agents_md:
agents_md = get_project_agents_md_instruction_file(name)
if agents_md:
result.append(agents_md)
additional_instructions = get_additional_instructions_files(name)
for filename in sorted(additional_instructions):
path = files.get_abs_path(
project_folder,
PROJECT_META_DIR,
PROJECT_INSTRUCTIONS_DIR,
filename,
)
result.append(
(files.normalize_a0_path(path), additional_instructions[filename])
)
return result
def get_project_agents_md_instruction_file(name: str) -> tuple[str, str] | None:
project_folder = get_project_folder(name)
for filename in PROJECT_AGENTS_MD_FILES:
matches = files.read_text_files_in_dir(project_folder, pattern=filename)
if filename in matches:
path = files.get_abs_path(project_folder, filename)
return (files.normalize_a0_path(path), matches[filename])
return None
def _format_project_instruction_files(instruction_files: list[tuple[str, str]]) -> str:
if not instruction_files:
return ""
parts = ["## project instruction files"]
for path, content in instruction_files:
parts.append(f"### path: {path}\n\n{content}")
return "\n\n".join(parts)
def _normalize_include_agents_md(value: object) -> bool:
if value is None:
return True
if isinstance(value, bool):
return value
if isinstance(value, str):
return value.strip().lower() not in {"0", "false", "no", "off"}
return bool(value)
def _project_header_for_save(data: BasicProjectData) -> dict:
header = dict(data)
header["include_agents_md"] = _normalize_include_agents_md(
header.get("include_agents_md", True)
)
return header
def get_context_project_name(context: "AgentContext") -> str | None:
return context.get_data(CONTEXT_DATA_KEY_PROJECT)
def load_project_variables(name: str):
try:
abs_path = files.get_abs_path(get_project_meta(name), "variables.env")
return files.read_file(abs_path)
except Exception:
return ""
def save_project_variables(name: str, variables: str):
abs_path = files.get_abs_path(get_project_meta(name), "variables.env")
files.write_file(abs_path, variables)
def load_project_subagents(name: str) -> dict[str, SubAgentSettings]:
try:
abs_path = files.get_abs_path(get_project_meta(name), "agents.json")
data = dirty_json.parse(files.read_file(abs_path))
if isinstance(data, dict):
return _normalize_subagents(data) # type: ignore[arg-type,return-value]
return {}
except Exception:
return {}
def save_project_subagents(name: str, subagents_data: dict[str, SubAgentSettings]):
abs_path = files.get_abs_path(get_project_meta(name), "agents.json")
normalized = _normalize_subagents(subagents_data)
content = dirty_json.stringify(normalized)
files.write_file(abs_path, content)
def _normalize_subagents(
subagents_data: dict[str, SubAgentSettings]
) -> dict[str, SubAgentSettings]:
from helpers import subagents
agents_dict = subagents.get_agents_dict()
normalized: dict[str, SubAgentSettings] = {}
for key, value in subagents_data.items():
agent = agents_dict.get(key)
if not agent:
continue
enabled = bool(value["enabled"])
if agent.enabled == enabled:
continue
normalized[key] = {"enabled": enabled}
return normalized
def load_project_secrets_masked(name: str, merge_with_global=False):
from helpers import secrets
mgr = secrets.get_project_secrets_manager(name, merge_with_global)
return mgr.get_masked_secrets()
def save_project_secrets(name: str, secrets: str):
from helpers.secrets import get_project_secrets_manager
secrets_manager = get_project_secrets_manager(name)
secrets_manager.save_secrets_with_merge(secrets)
def create_project_meta_folders(name: str):
# create instructions folder
files.create_dir(get_project_meta(name, PROJECT_INSTRUCTIONS_DIR))
# create knowledge folders (plugins create their own subdirs lazily)
files.create_dir(get_project_meta(name, PROJECT_KNOWLEDGE_DIR))
# create project skills folder for Project Settings > Skills > Open Folder
files.create_dir(get_project_meta(name, PROJECT_SKILLS_DIR))
def get_knowledge_files_count(name: str):
knowledge_folder = files.get_abs_path(
get_project_meta(name, PROJECT_KNOWLEDGE_DIR)
)
return len(files.list_files_in_dir_recursively(knowledge_folder))
def get_file_structure(name: str, basic_data: BasicProjectData|None=None) -> str:
project_folder = get_project_folder(name)
if basic_data is None:
basic_data = load_basic_project_data(name)
tree = str(file_tree.file_tree(
project_folder,
max_depth=basic_data["file_structure"]["max_depth"],
max_files=basic_data["file_structure"]["max_files"],
max_folders=basic_data["file_structure"]["max_folders"],
max_lines=basic_data["file_structure"]["max_lines"],
ignore=basic_data["file_structure"]["gitignore"],
output_mode=file_tree.OUTPUT_MODE_STRING
))
# empty?
if "\n" not in tree:
tree += "\n # Empty"
return tree