mirror of
https://github.com/agent0ai/agent-zero.git
synced 2026-07-09 17:08:29 +00:00
bcf2634000
Load active-project AGENTS.md path-chain guidance into the protocol area without duplicating the project root instructions. Move the AGENTS.md protocol wording into a prompt template, reuse existing file/path helpers, and cover direct-path discovery plus prompt assembly with focused tests.
736 lines
23 KiB
Python
736 lines
23 KiB
Python
import os
|
|
from typing import NotRequired, TypedDict, TYPE_CHECKING, cast
|
|
|
|
from helpers import files, dirty_json, persist_chat, file_tree, extension
|
|
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.override.md",
|
|
"AGENTS.Override.md",
|
|
"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
|
|
|
|
|
|
ProjectExtendedData = dict[str, object]
|
|
_PROJECT_CORE_EDIT_KEYS = frozenset(BasicProjectData.__annotations__) | frozenset(
|
|
EditProjectData.__annotations__
|
|
)
|
|
_PROJECT_TRANSIENT_INPUT_KEYS = frozenset({"git_token"})
|
|
|
|
|
|
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):
|
|
extended_data = _project_extended_data_for_save(data)
|
|
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_extended_data(name, extended_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
|
|
|
|
extended_data = _project_extended_data_for_save(data)
|
|
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_extended_data(actual_name, extended_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):
|
|
extended_data = _project_extended_data_for_save(data)
|
|
|
|
# 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_extended_data(name, extended_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)
|
|
_merge_project_extended_data(data, load_project_extended_data(name))
|
|
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)
|
|
|
|
|
|
@extension.extensible
|
|
def load_project_extended_data(name: str) -> ProjectExtendedData:
|
|
return {}
|
|
|
|
|
|
@extension.extensible
|
|
def save_project_extended_data(name: str, project_data: ProjectExtendedData):
|
|
return None
|
|
|
|
|
|
def _project_extended_data_for_save(data: object) -> ProjectExtendedData:
|
|
if not isinstance(data, dict):
|
|
return {}
|
|
return {
|
|
str(key): value
|
|
for key, value in data.items()
|
|
if str(key) not in _PROJECT_CORE_EDIT_KEYS
|
|
and str(key) not in _PROJECT_TRANSIENT_INPUT_KEYS
|
|
}
|
|
|
|
|
|
def _merge_project_extended_data(
|
|
data: EditProjectData,
|
|
extended_data: object,
|
|
) -> None:
|
|
if not isinstance(extended_data, dict):
|
|
return
|
|
|
|
conflicts = sorted(str(key) for key in extended_data if key in _PROJECT_CORE_EDIT_KEYS)
|
|
if conflicts:
|
|
raise ValueError(
|
|
"Project extension data cannot overwrite core project fields: "
|
|
+ ", ".join(conflicts)
|
|
)
|
|
|
|
data.update(extended_data) # type: ignore[typeddict-item]
|
|
|
|
|
|
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 ""
|
|
include_agents_md = project_data.get("include_agents_md", True)
|
|
instruction_files = get_project_instruction_files(
|
|
name,
|
|
include_agents_md=include_agents_md,
|
|
)
|
|
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 "",
|
|
"include_agents_md": include_agents_md,
|
|
"project_path": files.normalize_a0_path(get_project_folder(name)),
|
|
"project_git_url": project_data.get("git_url", ""),
|
|
}
|
|
|
|
|
|
def get_agents_md_chain(root: str, target: str) -> list[tuple[str, str]]:
|
|
root_real = os.path.realpath(files.fix_dev_path(root))
|
|
target_real = os.path.realpath(files.fix_dev_path(target))
|
|
if os.path.isfile(target_real):
|
|
target_real = os.path.dirname(target_real)
|
|
|
|
if files.is_in_dir(target_real, root_real):
|
|
dirs = []
|
|
cursor = target_real
|
|
while True:
|
|
dirs.append(cursor)
|
|
if cursor == root_real:
|
|
break
|
|
parent = os.path.dirname(cursor)
|
|
if parent == cursor:
|
|
break
|
|
cursor = parent
|
|
dirs.reverse()
|
|
else:
|
|
dirs = [root_real]
|
|
|
|
chain = []
|
|
for dir_path in dirs:
|
|
for filename in PROJECT_AGENTS_MD_FILES:
|
|
matches = files.read_text_files_in_dir(dir_path, pattern=filename)
|
|
if filename not in matches:
|
|
continue
|
|
chain.append((files.get_abs_path(dir_path, filename), matches[filename]))
|
|
break
|
|
return chain
|
|
|
|
|
|
def build_agents_md_protocol(name: str, target: str | None = None) -> str:
|
|
project_folder = get_project_folder(name)
|
|
project_agents_md = get_project_agents_md_instruction_file(name)
|
|
project_agents_md_path = (
|
|
os.path.realpath(files.fix_dev_path(project_agents_md[0]))
|
|
if project_agents_md
|
|
else ""
|
|
)
|
|
entries = [
|
|
(path, content)
|
|
for path, content in get_agents_md_chain(
|
|
files.get_abs_path(""),
|
|
target or project_folder,
|
|
)
|
|
if os.path.realpath(path) != project_agents_md_path
|
|
]
|
|
if not entries:
|
|
return ""
|
|
|
|
instructions = []
|
|
for path, content in entries:
|
|
instructions.append(
|
|
f"### path: {files.normalize_a0_path(path)}\n\n{content.strip()}"
|
|
)
|
|
return files.read_prompt_file(
|
|
"agent.protocol.projects.agents_md.md",
|
|
_directories=["prompts"],
|
|
agents_md_instructions="\n\n".join(instructions),
|
|
).strip()
|
|
|
|
|
|
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 path, content in get_agents_md_chain(project_folder, project_folder):
|
|
return (files.normalize_a0_path(path), content)
|
|
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
|