| 1 | import os |
| 2 | from typing import NotRequired, TypedDict, TYPE_CHECKING, cast |
| 3 | |
| 4 | from helpers import files, dirty_json, persist_chat, file_tree, extension |
| 5 | from helpers.print_style import PrintStyle |
| 6 | |
| 7 | |
| 8 | if TYPE_CHECKING: |
| 9 | from agent import AgentContext |
| 10 | |
| 11 | PROJECTS_PARENT_DIR = "usr/projects" |
| 12 | PROJECT_META_DIR = ".a0proj" |
| 13 | PROJECT_INSTRUCTIONS_DIR = "instructions" |
| 14 | PROJECT_KNOWLEDGE_DIR = "knowledge" |
| 15 | PROJECT_SKILLS_DIR = "skills" |
| 16 | PROJECT_HEADER_FILE = "project.json" |
| 17 | PROJECT_MCP_SERVERS_FILE = "mcp_servers.json" |
| 18 | PROJECT_AGENTS_MD_FILES = ( |
| 19 | "AGENTS.override.md", |
| 20 | "AGENTS.Override.md", |
| 21 | "AGENTS.md", |
| 22 | "Agents.md", |
| 23 | "agents.md", |
| 24 | ) |
| 25 | DEFAULT_MCP_SERVERS_CONFIG = '{\n "mcpServers": {}\n}' |
| 26 | |
| 27 | CONTEXT_DATA_KEY_PROJECT = "project" |
| 28 | |
| 29 | |
| 30 | class FileStructureInjectionSettings(TypedDict): |
| 31 | enabled: bool |
| 32 | max_depth: int |
| 33 | max_files: int |
| 34 | max_folders: int |
| 35 | max_lines: int |
| 36 | gitignore: str |
| 37 | |
| 38 | class SubAgentSettings(TypedDict): |
| 39 | enabled: bool |
| 40 | |
| 41 | class BasicProjectData(TypedDict): |
| 42 | title: str |
| 43 | description: str |
| 44 | instructions: str |
| 45 | include_agents_md: NotRequired[bool] |
| 46 | mcp_servers: NotRequired[str] |
| 47 | color: str |
| 48 | git_url: str |
| 49 | file_structure: FileStructureInjectionSettings |
| 50 | |
| 51 | class GitStatusData(TypedDict, total=False): |
| 52 | is_git_repo: bool |
| 53 | remote_url: str |
| 54 | current_branch: str |
| 55 | is_dirty: bool |
| 56 | untracked_count: int |
| 57 | last_commit: dict |
| 58 | error: str |
| 59 | |
| 60 | class EditProjectData(BasicProjectData): |
| 61 | name: str |
| 62 | instruction_files_count: int |
| 63 | knowledge_files_count: int |
| 64 | variables: str |
| 65 | secrets: str |
| 66 | mcp_servers: str |
| 67 | git_status: GitStatusData |
| 68 | |
| 69 | |
| 70 | ProjectExtendedData = dict[str, object] |
| 71 | _PROJECT_CORE_EDIT_KEYS = frozenset(BasicProjectData.__annotations__) | frozenset( |
| 72 | EditProjectData.__annotations__ |
| 73 | ) |
| 74 | _PROJECT_TRANSIENT_INPUT_KEYS = frozenset({"git_token", "subagents"}) |
| 75 | |
| 76 | |
| 77 | def get_projects_parent_folder(): |
| 78 | return files.get_abs_path(PROJECTS_PARENT_DIR) |
| 79 | |
| 80 | |
| 81 | def get_project_folder(name: str): |
| 82 | return files.get_abs_path(get_projects_parent_folder(), name) |
| 83 | |
| 84 | |
| 85 | def get_project_meta(name: str, *sub_dirs: str): |
| 86 | return files.get_abs_path(get_project_folder(name), PROJECT_META_DIR, *sub_dirs) |
| 87 | |
| 88 | |
| 89 | def validate_project_name(name: str | None) -> str: |
| 90 | candidate = str(name or "").strip() |
| 91 | if ( |
| 92 | not candidate |
| 93 | or candidate in {".", ".."} |
| 94 | or os.path.basename(candidate) != candidate |
| 95 | ): |
| 96 | raise ValueError("Invalid project name") |
| 97 | return candidate |
| 98 | |
| 99 | |
| 100 | def delete_project(name: str): |
| 101 | abs_path = files.get_abs_path(PROJECTS_PARENT_DIR, name) |
| 102 | files.delete_dir(abs_path) |
| 103 | deactivate_project_in_chats(name) |
| 104 | return name |
| 105 | |
| 106 | |
| 107 | def create_project(name: str, data: BasicProjectData): |
| 108 | extended_data = _project_extended_data_for_save(data) |
| 109 | mcp_servers = data.get("mcp_servers") if isinstance(data, dict) else None |
| 110 | abs_path = files.create_dir_safe( |
| 111 | files.get_abs_path(PROJECTS_PARENT_DIR, name), rename_format="{name}_{number}" |
| 112 | ) |
| 113 | create_project_meta_folders(name) |
| 114 | data = _normalizeBasicData(data) |
| 115 | save_project_header(name, data) |
| 116 | save_project_mcp_servers(name, mcp_servers or DEFAULT_MCP_SERVERS_CONFIG) |
| 117 | save_project_extended_data(name, extended_data) |
| 118 | return name |
| 119 | |
| 120 | |
| 121 | def clone_git_project(name: str, git_url: str, git_token: str, data: BasicProjectData): |
| 122 | """Clone a git repository as a new A0 project. Token is used only for cloning via http header.""" |
| 123 | from helpers import git |
| 124 | |
| 125 | extended_data = _project_extended_data_for_save(data) |
| 126 | mcp_servers = data.get("mcp_servers") if isinstance(data, dict) else None |
| 127 | |
| 128 | abs_path = files.create_dir_safe( |
| 129 | files.get_abs_path(PROJECTS_PARENT_DIR, name), rename_format="{name}_{number}" |
| 130 | ) |
| 131 | actual_name = files.basename(abs_path) |
| 132 | |
| 133 | try: |
| 134 | # Clone with token via http.extraHeader (token never in URL or git config) |
| 135 | git.clone_repo(git_url, abs_path, token=git_token) |
| 136 | clean_url = git.strip_auth_from_url(git_url) |
| 137 | |
| 138 | # Check if cloned repo already has .a0proj |
| 139 | meta_path = os.path.join(abs_path, PROJECT_META_DIR, PROJECT_HEADER_FILE) |
| 140 | if os.path.exists(meta_path): |
| 141 | # Merge: keep cloned content, override only user-specified fields |
| 142 | cloned_header: BasicProjectData = dirty_json.parse(files.read_file(meta_path)) # type: ignore |
| 143 | cloned_header["title"] = data.get("title") or cloned_header.get("title", "") |
| 144 | cloned_header["color"] = data.get("color") or cloned_header.get("color", "") |
| 145 | cloned_header["git_url"] = clean_url |
| 146 | save_project_header(actual_name, cloned_header) |
| 147 | else: |
| 148 | # New project: create meta folders and save header |
| 149 | create_project_meta_folders(actual_name) |
| 150 | data = _normalizeBasicData(data) |
| 151 | data["git_url"] = clean_url |
| 152 | save_project_header(actual_name, data) |
| 153 | |
| 154 | if mcp_servers: |
| 155 | save_project_mcp_servers(actual_name, mcp_servers) |
| 156 | save_project_extended_data(actual_name, extended_data) |
| 157 | |
| 158 | return actual_name |
| 159 | except Exception as e: |
| 160 | try: |
| 161 | files.delete_dir(abs_path) |
| 162 | except Exception: |
| 163 | pass |
| 164 | raise e |
| 165 | |
| 166 | |
| 167 | def load_project_header(name: str): |
| 168 | abs_path = files.get_abs_path( |
| 169 | PROJECTS_PARENT_DIR, name, PROJECT_META_DIR, PROJECT_HEADER_FILE |
| 170 | ) |
| 171 | header: dict = dirty_json.parse(files.read_file(abs_path)) # type: ignore |
| 172 | header["name"] = name |
| 173 | return header |
| 174 | |
| 175 | |
| 176 | def _default_file_structure_settings(): |
| 177 | try: |
| 178 | gitignore = files.read_file("conf/projects.default.gitignore") |
| 179 | except Exception: |
| 180 | gitignore = "" |
| 181 | return FileStructureInjectionSettings( |
| 182 | enabled=True, |
| 183 | max_depth=5, |
| 184 | max_files=20, |
| 185 | max_folders=20, |
| 186 | max_lines=250, |
| 187 | gitignore=gitignore, |
| 188 | ) |
| 189 | |
| 190 | |
| 191 | def _normalizeBasicData(data: BasicProjectData) -> BasicProjectData: |
| 192 | return { |
| 193 | "title": data.get("title", ""), |
| 194 | "description": data.get("description", ""), |
| 195 | "instructions": data.get("instructions", ""), |
| 196 | "include_agents_md": _normalize_include_agents_md( |
| 197 | data.get("include_agents_md", True) |
| 198 | ), |
| 199 | "color": data.get("color", ""), |
| 200 | "git_url": data.get("git_url", ""), |
| 201 | "file_structure": data.get( |
| 202 | "file_structure", |
| 203 | _default_file_structure_settings(), |
| 204 | ), |
| 205 | } |
| 206 | |
| 207 | |
| 208 | def _normalizeEditData(data: EditProjectData) -> EditProjectData: |
| 209 | normalized: EditProjectData = { |
| 210 | "name": data.get("name", ""), |
| 211 | "title": data.get("title", ""), |
| 212 | "description": data.get("description", ""), |
| 213 | "instructions": data.get("instructions", ""), |
| 214 | "include_agents_md": _normalize_include_agents_md( |
| 215 | data.get("include_agents_md", True) |
| 216 | ), |
| 217 | "variables": data.get("variables", ""), |
| 218 | "mcp_servers": data.get("mcp_servers", DEFAULT_MCP_SERVERS_CONFIG), |
| 219 | "color": data.get("color", ""), |
| 220 | "git_url": data.get("git_url", ""), |
| 221 | "git_status": data.get("git_status", {"is_git_repo": False}), |
| 222 | "instruction_files_count": data.get("instruction_files_count", 0), |
| 223 | "knowledge_files_count": data.get("knowledge_files_count", 0), |
| 224 | "secrets": data.get("secrets", ""), |
| 225 | "file_structure": data.get( |
| 226 | "file_structure", |
| 227 | _default_file_structure_settings(), |
| 228 | ), |
| 229 | } |
| 230 | return normalized |
| 231 | |
| 232 | |
| 233 | def _edit_data_to_basic_data(data: EditProjectData): |
| 234 | return _normalizeBasicData(data) |
| 235 | |
| 236 | |
| 237 | def _basic_data_to_edit_data(data: BasicProjectData) -> EditProjectData: |
| 238 | base: EditProjectData = cast( |
| 239 | EditProjectData, |
| 240 | { |
| 241 | **data, |
| 242 | "name": "", |
| 243 | "instruction_files_count": 0, |
| 244 | "knowledge_files_count": 0, |
| 245 | "variables": "", |
| 246 | "secrets": "", |
| 247 | "git_status": {"is_git_repo": False}, |
| 248 | }, |
| 249 | ) |
| 250 | return _normalizeEditData(base) |
| 251 | |
| 252 | |
| 253 | def update_project(name: str, data: EditProjectData): |
| 254 | extended_data = _project_extended_data_for_save(data) |
| 255 | |
| 256 | # merge with current state |
| 257 | current = load_edit_project_data(name) |
| 258 | current.update(data) |
| 259 | current = _normalizeEditData(current) |
| 260 | |
| 261 | # save header data |
| 262 | header = _edit_data_to_basic_data(current) |
| 263 | save_project_header(name, header) |
| 264 | |
| 265 | # save secrets |
| 266 | save_project_variables(name, current["variables"]) |
| 267 | save_project_secrets(name, current["secrets"]) |
| 268 | save_project_mcp_servers(name, current["mcp_servers"]) |
| 269 | save_project_extended_data(name, extended_data) |
| 270 | |
| 271 | reactivate_project_in_chats(name) |
| 272 | return name |
| 273 | |
| 274 | |
| 275 | def load_basic_project_data(name: str) -> BasicProjectData: |
| 276 | data = cast(BasicProjectData, load_project_header(name)) |
| 277 | normalized = _normalizeBasicData(data) |
| 278 | return normalized |
| 279 | |
| 280 | |
| 281 | def load_edit_project_data(name: str) -> EditProjectData: |
| 282 | from helpers import git |
| 283 | |
| 284 | data = load_basic_project_data(name) |
| 285 | create_project_meta_folders(name) |
| 286 | additional_instructions = get_additional_instructions_files(name) |
| 287 | variables = load_project_variables(name) |
| 288 | mcp_servers = load_project_mcp_servers(name) |
| 289 | secrets = load_project_secrets_masked(name) |
| 290 | knowledge_files_count = get_knowledge_files_count(name) |
| 291 | git_status = cast(GitStatusData, git.get_repo_status(get_project_folder(name))) |
| 292 | |
| 293 | data = cast( |
| 294 | EditProjectData, |
| 295 | { |
| 296 | **data, |
| 297 | "name": name, |
| 298 | "instruction_files_count": len(additional_instructions), |
| 299 | "knowledge_files_count": knowledge_files_count, |
| 300 | "variables": variables, |
| 301 | "mcp_servers": mcp_servers, |
| 302 | "secrets": secrets, |
| 303 | "git_status": git_status, |
| 304 | }, |
| 305 | ) |
| 306 | data = _normalizeEditData(data) |
| 307 | _merge_project_extended_data(data, load_project_extended_data(name)) |
| 308 | return data |
| 309 | |
| 310 | |
| 311 | def save_project_header(name: str, data: BasicProjectData): |
| 312 | # save project header file |
| 313 | header = dirty_json.stringify(_project_header_for_save(data)) |
| 314 | abs_path = files.get_abs_path( |
| 315 | PROJECTS_PARENT_DIR, name, PROJECT_META_DIR, PROJECT_HEADER_FILE |
| 316 | ) |
| 317 | |
| 318 | files.write_file(abs_path, header) |
| 319 | |
| 320 | |
| 321 | @extension.extensible |
| 322 | def load_project_extended_data(name: str) -> ProjectExtendedData: |
| 323 | return {} |
| 324 | |
| 325 | |
| 326 | @extension.extensible |
| 327 | def save_project_extended_data(name: str, project_data: ProjectExtendedData): |
| 328 | return None |
| 329 | |
| 330 | |
| 331 | def _project_extended_data_for_save(data: object) -> ProjectExtendedData: |
| 332 | if not isinstance(data, dict): |
| 333 | return {} |
| 334 | return { |
| 335 | str(key): value |
| 336 | for key, value in data.items() |
| 337 | if str(key) not in _PROJECT_CORE_EDIT_KEYS |
| 338 | and str(key) not in _PROJECT_TRANSIENT_INPUT_KEYS |
| 339 | } |
| 340 | |
| 341 | |
| 342 | def _merge_project_extended_data( |
| 343 | data: EditProjectData, |
| 344 | extended_data: object, |
| 345 | ) -> None: |
| 346 | if not isinstance(extended_data, dict): |
| 347 | return |
| 348 | |
| 349 | conflicts = sorted(str(key) for key in extended_data if key in _PROJECT_CORE_EDIT_KEYS) |
| 350 | if conflicts: |
| 351 | raise ValueError( |
| 352 | "Project extension data cannot overwrite core project fields: " |
| 353 | + ", ".join(conflicts) |
| 354 | ) |
| 355 | |
| 356 | data.update(extended_data) # type: ignore[typeddict-item] |
| 357 | |
| 358 | |
| 359 | def load_project_mcp_servers(name: str) -> str: |
| 360 | project_name = validate_project_name(name) |
| 361 | try: |
| 362 | return files.read_file(get_project_meta(project_name, PROJECT_MCP_SERVERS_FILE)) |
| 363 | except Exception: |
| 364 | return DEFAULT_MCP_SERVERS_CONFIG |
| 365 | |
| 366 | |
| 367 | def save_project_mcp_servers(name: str, mcp_servers: str): |
| 368 | project_name = validate_project_name(name) |
| 369 | content = mcp_servers if isinstance(mcp_servers, str) else DEFAULT_MCP_SERVERS_CONFIG |
| 370 | files.write_file(get_project_meta(project_name, PROJECT_MCP_SERVERS_FILE), content) |
| 371 | |
| 372 | |
| 373 | def get_active_projects_list(): |
| 374 | return _get_projects_list(get_projects_parent_folder()) |
| 375 | |
| 376 | |
| 377 | def _get_projects_list(parent_dir): |
| 378 | projects = [] |
| 379 | |
| 380 | # folders in project directory |
| 381 | for name in os.listdir(parent_dir): |
| 382 | try: |
| 383 | abs_path = os.path.join(parent_dir, name) |
| 384 | if os.path.isdir(abs_path): |
| 385 | project_data = load_basic_project_data(name) |
| 386 | projects.append( |
| 387 | { |
| 388 | "name": name, |
| 389 | "title": project_data.get("title", ""), |
| 390 | "description": project_data.get("description", ""), |
| 391 | "color": project_data.get("color", ""), |
| 392 | } |
| 393 | ) |
| 394 | except Exception as e: |
| 395 | PrintStyle.error(f"Error loading project {name}: {str(e)}") |
| 396 | |
| 397 | # sort projects by name |
| 398 | projects.sort(key=lambda x: x["name"]) |
| 399 | return projects |
| 400 | |
| 401 | |
| 402 | def reconcile_agent_profile( |
| 403 | context: "AgentContext", project_name: str | None, available: dict | None = None |
| 404 | ) -> bool: |
| 405 | from helpers import subagents |
| 406 | from initialize import initialize_agent |
| 407 | |
| 408 | if available is None: |
| 409 | available = subagents.get_available_agents_dict(project_name) |
| 410 | if getattr(context.config, "profile", "") in available: |
| 411 | return False |
| 412 | |
| 413 | config = initialize_agent() |
| 414 | if config.profile not in available: |
| 415 | fallback = "agent0" if "agent0" in available else next(iter(available), "agent0") |
| 416 | config = initialize_agent(override_settings={"agent_profile": fallback}) |
| 417 | context.config = config |
| 418 | context.agent0.config = config |
| 419 | return True |
| 420 | |
| 421 | |
| 422 | def reconcile_agent_profiles( |
| 423 | project_name: str | None, *, all_scopes: bool = False |
| 424 | ) -> None: |
| 425 | from agent import AgentContext |
| 426 | from helpers import subagents |
| 427 | from helpers.state_monitor_integration import mark_dirty_for_context |
| 428 | |
| 429 | available_by_project: dict[str | None, dict] = {} |
| 430 | for context in AgentContext.all(): |
| 431 | context_project = get_context_project_name(context) |
| 432 | if not all_scopes and context_project != project_name: |
| 433 | continue |
| 434 | if context_project not in available_by_project: |
| 435 | available_by_project[context_project] = ( |
| 436 | subagents.get_available_agents_dict(context_project) |
| 437 | ) |
| 438 | if not reconcile_agent_profile( |
| 439 | context, context_project, available_by_project[context_project] |
| 440 | ): |
| 441 | continue |
| 442 | persist_chat.save_tmp_chat(context) |
| 443 | mark_dirty_for_context( |
| 444 | context.id, reason="projects.reconcile_agent_profiles" |
| 445 | ) |
| 446 | |
| 447 | |
| 448 | def activate_project(context_id: str, name: str, *, mark_dirty: bool = True): |
| 449 | from agent import AgentContext |
| 450 | |
| 451 | data = load_edit_project_data(name) |
| 452 | context = AgentContext.get(context_id) |
| 453 | if context is None: |
| 454 | raise Exception("Context not found") |
| 455 | display_name = str(data.get("title", name)) |
| 456 | display_name = display_name[:22] + "..." if len(display_name) > 25 else display_name |
| 457 | context.set_data(CONTEXT_DATA_KEY_PROJECT, name) |
| 458 | context.set_output_data( |
| 459 | CONTEXT_DATA_KEY_PROJECT, |
| 460 | {"name": name, "title": display_name, "color": data.get("color", "")}, |
| 461 | ) |
| 462 | reconcile_agent_profile(context, name) |
| 463 | |
| 464 | # persist |
| 465 | persist_chat.save_tmp_chat(context) |
| 466 | |
| 467 | if mark_dirty: |
| 468 | from helpers.state_monitor_integration import mark_dirty_all |
| 469 | mark_dirty_all(reason="projects.activate_project") |
| 470 | |
| 471 | |
| 472 | def deactivate_project(context_id: str, *, mark_dirty: bool = True): |
| 473 | from agent import AgentContext |
| 474 | |
| 475 | context = AgentContext.get(context_id) |
| 476 | if context is None: |
| 477 | raise Exception("Context not found") |
| 478 | context.set_data(CONTEXT_DATA_KEY_PROJECT, None) |
| 479 | context.set_output_data(CONTEXT_DATA_KEY_PROJECT, None) |
| 480 | reconcile_agent_profile(context, None) |
| 481 | |
| 482 | # persist |
| 483 | persist_chat.save_tmp_chat(context) |
| 484 | |
| 485 | if mark_dirty: |
| 486 | from helpers.state_monitor_integration import mark_dirty_all |
| 487 | mark_dirty_all(reason="projects.deactivate_project") |
| 488 | |
| 489 | |
| 490 | def reactivate_project_in_chats(name: str): |
| 491 | from agent import AgentContext |
| 492 | |
| 493 | for context in AgentContext.all(): |
| 494 | if context.get_data(CONTEXT_DATA_KEY_PROJECT) == name: |
| 495 | activate_project(context.id, name, mark_dirty=False) |
| 496 | |
| 497 | from helpers.state_monitor_integration import mark_dirty_all |
| 498 | mark_dirty_all(reason="projects.reactivate_project_in_chats") |
| 499 | |
| 500 | |
| 501 | def deactivate_project_in_chats(name: str): |
| 502 | from agent import AgentContext |
| 503 | |
| 504 | for context in AgentContext.all(): |
| 505 | if context.get_data(CONTEXT_DATA_KEY_PROJECT) == name: |
| 506 | deactivate_project(context.id, mark_dirty=False) |
| 507 | |
| 508 | from helpers.state_monitor_integration import mark_dirty_all |
| 509 | mark_dirty_all(reason="projects.deactivate_project_in_chats") |
| 510 | |
| 511 | |
| 512 | def build_system_prompt_vars(name: str): |
| 513 | project_data = load_basic_project_data(name) |
| 514 | main_instructions = project_data.get("instructions", "") or "" |
| 515 | include_agents_md = project_data.get("include_agents_md", True) |
| 516 | instruction_files = get_project_instruction_files( |
| 517 | name, |
| 518 | include_agents_md=include_agents_md, |
| 519 | ) |
| 520 | instruction_parts = [ |
| 521 | main_instructions, |
| 522 | _format_project_instruction_files(instruction_files), |
| 523 | ] |
| 524 | complete_instructions = "\n\n".join( |
| 525 | part.strip() for part in instruction_parts if part.strip() |
| 526 | ).strip() |
| 527 | return { |
| 528 | "project_name": project_data.get("title", ""), |
| 529 | "project_description": project_data.get("description", ""), |
| 530 | "project_instructions": complete_instructions or "", |
| 531 | "include_agents_md": include_agents_md, |
| 532 | "project_path": files.normalize_a0_path(get_project_folder(name)), |
| 533 | "project_git_url": project_data.get("git_url", ""), |
| 534 | } |
| 535 | |
| 536 | |
| 537 | def get_agents_md_chain(root: str, target: str) -> list[tuple[str, str]]: |
| 538 | root_real = os.path.realpath(files.fix_dev_path(root)) |
| 539 | target_real = os.path.realpath(files.fix_dev_path(target)) |
| 540 | if os.path.isfile(target_real): |
| 541 | target_real = os.path.dirname(target_real) |
| 542 | |
| 543 | if files.is_in_dir(target_real, root_real): |
| 544 | dirs = [] |
| 545 | cursor = target_real |
| 546 | while True: |
| 547 | dirs.append(cursor) |
| 548 | if cursor == root_real: |
| 549 | break |
| 550 | parent = os.path.dirname(cursor) |
| 551 | if parent == cursor: |
| 552 | break |
| 553 | cursor = parent |
| 554 | dirs.reverse() |
| 555 | else: |
| 556 | dirs = [root_real] |
| 557 | |
| 558 | chain = [] |
| 559 | for dir_path in dirs: |
| 560 | for filename in PROJECT_AGENTS_MD_FILES: |
| 561 | matches = files.read_text_files_in_dir(dir_path, pattern=filename) |
| 562 | if filename not in matches: |
| 563 | continue |
| 564 | chain.append((files.get_abs_path(dir_path, filename), matches[filename])) |
| 565 | break |
| 566 | return chain |
| 567 | |
| 568 | |
| 569 | def build_agents_md_protocol(name: str, target: str | None = None) -> str: |
| 570 | project_folder = get_project_folder(name) |
| 571 | project_agents_md = get_project_agents_md_instruction_file(name) |
| 572 | project_agents_md_path = ( |
| 573 | os.path.realpath(files.fix_dev_path(project_agents_md[0])) |
| 574 | if project_agents_md |
| 575 | else "" |
| 576 | ) |
| 577 | entries = [ |
| 578 | (path, content) |
| 579 | for path, content in get_agents_md_chain( |
| 580 | files.get_abs_path(""), |
| 581 | target or project_folder, |
| 582 | ) |
| 583 | if os.path.realpath(path) != project_agents_md_path |
| 584 | ] |
| 585 | if not entries: |
| 586 | return "" |
| 587 | |
| 588 | instructions = [] |
| 589 | for path, content in entries: |
| 590 | instructions.append( |
| 591 | f"### path: {files.normalize_a0_path(path)}\n\n{content.strip()}" |
| 592 | ) |
| 593 | return files.read_prompt_file( |
| 594 | "agent.protocol.projects.agents_md.md", |
| 595 | _directories=["prompts"], |
| 596 | agents_md_instructions="\n\n".join(instructions), |
| 597 | ).strip() |
| 598 | |
| 599 | |
| 600 | def get_additional_instructions_files(name: str): |
| 601 | instructions_folder = files.get_abs_path( |
| 602 | get_project_folder(name), PROJECT_META_DIR, PROJECT_INSTRUCTIONS_DIR |
| 603 | ) |
| 604 | return files.read_text_files_in_dir(instructions_folder) |
| 605 | |
| 606 | |
| 607 | def get_project_instruction_files( |
| 608 | name: str, |
| 609 | include_agents_md: bool = True, |
| 610 | ) -> list[tuple[str, str]]: |
| 611 | project_folder = get_project_folder(name) |
| 612 | result: list[tuple[str, str]] = [] |
| 613 | |
| 614 | if include_agents_md: |
| 615 | agents_md = get_project_agents_md_instruction_file(name) |
| 616 | if agents_md: |
| 617 | result.append(agents_md) |
| 618 | |
| 619 | additional_instructions = get_additional_instructions_files(name) |
| 620 | for filename in sorted(additional_instructions): |
| 621 | path = files.get_abs_path( |
| 622 | project_folder, |
| 623 | PROJECT_META_DIR, |
| 624 | PROJECT_INSTRUCTIONS_DIR, |
| 625 | filename, |
| 626 | ) |
| 627 | result.append( |
| 628 | (files.normalize_a0_path(path), additional_instructions[filename]) |
| 629 | ) |
| 630 | |
| 631 | return result |
| 632 | |
| 633 | |
| 634 | def get_project_agents_md_instruction_file(name: str) -> tuple[str, str] | None: |
| 635 | project_folder = get_project_folder(name) |
| 636 | for path, content in get_agents_md_chain(project_folder, project_folder): |
| 637 | return (files.normalize_a0_path(path), content) |
| 638 | return None |
| 639 | |
| 640 | |
| 641 | def _format_project_instruction_files(instruction_files: list[tuple[str, str]]) -> str: |
| 642 | if not instruction_files: |
| 643 | return "" |
| 644 | |
| 645 | parts = ["## project instruction files"] |
| 646 | for path, content in instruction_files: |
| 647 | parts.append(f"### path: {path}\n\n{content}") |
| 648 | return "\n\n".join(parts) |
| 649 | |
| 650 | |
| 651 | def _normalize_include_agents_md(value: object) -> bool: |
| 652 | if value is None: |
| 653 | return True |
| 654 | if isinstance(value, bool): |
| 655 | return value |
| 656 | if isinstance(value, str): |
| 657 | return value.strip().lower() not in {"0", "false", "no", "off"} |
| 658 | return bool(value) |
| 659 | |
| 660 | |
| 661 | def _project_header_for_save(data: BasicProjectData) -> dict: |
| 662 | header = dict(data) |
| 663 | header["include_agents_md"] = _normalize_include_agents_md( |
| 664 | header.get("include_agents_md", True) |
| 665 | ) |
| 666 | return header |
| 667 | |
| 668 | |
| 669 | def get_context_project_name(context: "AgentContext") -> str | None: |
| 670 | return context.get_data(CONTEXT_DATA_KEY_PROJECT) |
| 671 | |
| 672 | |
| 673 | def load_project_variables(name: str): |
| 674 | try: |
| 675 | abs_path = files.get_abs_path(get_project_meta(name), "variables.env") |
| 676 | return files.read_file(abs_path) |
| 677 | except Exception: |
| 678 | return "" |
| 679 | |
| 680 | |
| 681 | def save_project_variables(name: str, variables: str): |
| 682 | abs_path = files.get_abs_path(get_project_meta(name), "variables.env") |
| 683 | files.write_file(abs_path, variables) |
| 684 | |
| 685 | |
| 686 | def load_project_subagents(name: str) -> dict[str, SubAgentSettings]: |
| 687 | try: |
| 688 | abs_path = files.get_abs_path(get_project_meta(name), "agents.json") |
| 689 | data = dirty_json.parse(files.read_file(abs_path)) |
| 690 | if isinstance(data, dict): |
| 691 | return _normalize_subagents(data, name) # type: ignore[arg-type,return-value] |
| 692 | return {} |
| 693 | except Exception: |
| 694 | return {} |
| 695 | |
| 696 | |
| 697 | def save_project_subagents(name: str, subagents_data: dict[str, SubAgentSettings]): |
| 698 | abs_path = files.get_abs_path(get_project_meta(name), "agents.json") |
| 699 | normalized = _normalize_subagents(subagents_data, name) |
| 700 | content = dirty_json.stringify(normalized) |
| 701 | files.write_file(abs_path, content) |
| 702 | |
| 703 | |
| 704 | def set_project_subagent_enabled(name: str, profile_id: str, enabled: bool) -> None: |
| 705 | from helpers import subagents |
| 706 | |
| 707 | name = validate_project_name(name) |
| 708 | if not os.path.isdir(get_project_folder(name)): |
| 709 | raise ValueError("Project not found.") |
| 710 | if not isinstance(enabled, bool): |
| 711 | raise ValueError("Agent availability must be true or false.") |
| 712 | agent = subagents.get_agents_dict(name).get(profile_id) |
| 713 | if not agent: |
| 714 | raise ValueError(f'Agent profile "{profile_id}" does not exist.') |
| 715 | |
| 716 | path = get_project_meta(name, "agents.json") |
| 717 | try: |
| 718 | settings = dirty_json.parse(files.read_file(path)) |
| 719 | except FileNotFoundError: |
| 720 | settings = {} |
| 721 | except Exception as exc: |
| 722 | raise ValueError("Project agent availability is invalid.") from exc |
| 723 | if not isinstance(settings, dict) or any( |
| 724 | not isinstance(key, str) |
| 725 | or not isinstance(value, dict) |
| 726 | or not isinstance(value.get("enabled"), bool) |
| 727 | for key, value in settings.items() |
| 728 | ): |
| 729 | raise ValueError("Project agent availability is invalid.") |
| 730 | |
| 731 | if agent.enabled == enabled: |
| 732 | settings.pop(profile_id, None) |
| 733 | else: |
| 734 | settings[profile_id] = {"enabled": enabled} |
| 735 | save_project_subagents(name, settings) |
| 736 | |
| 737 | |
| 738 | def _normalize_subagents( |
| 739 | subagents_data: dict[str, SubAgentSettings], project_name: str = "" |
| 740 | ) -> dict[str, SubAgentSettings]: |
| 741 | from helpers import subagents |
| 742 | |
| 743 | scoped_agents = subagents.get_agents_dict(project_name or None) |
| 744 | |
| 745 | normalized: dict[str, SubAgentSettings] = {} |
| 746 | for key, value in subagents_data.items(): |
| 747 | agent = scoped_agents.get(key) |
| 748 | if not agent: |
| 749 | continue |
| 750 | |
| 751 | enabled = bool(value["enabled"]) |
| 752 | if agent.enabled == enabled: |
| 753 | continue |
| 754 | |
| 755 | normalized[key] = {"enabled": enabled} |
| 756 | |
| 757 | return normalized |
| 758 | |
| 759 | |
| 760 | def load_project_secrets_masked(name: str, merge_with_global=False): |
| 761 | from helpers import secrets |
| 762 | |
| 763 | mgr = secrets.get_project_secrets_manager(name, merge_with_global) |
| 764 | return mgr.get_masked_secrets() |
| 765 | |
| 766 | |
| 767 | def save_project_secrets(name: str, secrets: str): |
| 768 | from helpers.secrets import get_project_secrets_manager |
| 769 | |
| 770 | secrets_manager = get_project_secrets_manager(name) |
| 771 | secrets_manager.save_secrets_with_merge(secrets) |
| 772 | |
| 773 | |
| 774 | def create_project_meta_folders(name: str): |
| 775 | # create instructions folder |
| 776 | files.create_dir(get_project_meta(name, PROJECT_INSTRUCTIONS_DIR)) |
| 777 | |
| 778 | # create knowledge folders (plugins create their own subdirs lazily) |
| 779 | files.create_dir(get_project_meta(name, PROJECT_KNOWLEDGE_DIR)) |
| 780 | |
| 781 | # create project skills folder for Project Settings > Skills > Open Folder |
| 782 | files.create_dir(get_project_meta(name, PROJECT_SKILLS_DIR)) |
| 783 | |
| 784 | |
| 785 | def get_knowledge_files_count(name: str): |
| 786 | knowledge_folder = files.get_abs_path( |
| 787 | get_project_meta(name, PROJECT_KNOWLEDGE_DIR) |
| 788 | ) |
| 789 | return len(files.list_files_in_dir_recursively(knowledge_folder)) |
| 790 | |
| 791 | def get_file_structure(name: str, basic_data: BasicProjectData|None=None) -> str: |
| 792 | project_folder = get_project_folder(name) |
| 793 | if basic_data is None: |
| 794 | basic_data = load_basic_project_data(name) |
| 795 | |
| 796 | tree = str(file_tree.file_tree( |
| 797 | project_folder, |
| 798 | max_depth=basic_data["file_structure"]["max_depth"], |
| 799 | max_files=basic_data["file_structure"]["max_files"], |
| 800 | max_folders=basic_data["file_structure"]["max_folders"], |
| 801 | max_lines=basic_data["file_structure"]["max_lines"], |
| 802 | ignore=basic_data["file_structure"]["gitignore"], |
| 803 | output_mode=file_tree.OUTPUT_MODE_STRING |
| 804 | )) |
| 805 | |
| 806 | # empty? |
| 807 | if "\n" not in tree: |
| 808 | tree += "\n # Empty" |
| 809 | |
| 810 | return tree |