Runtime state for Atlas.
Created once per server session (MCP) or once per CLI invocation.
Holds lazy-loaded references to manifest, registry, config, and router.
All heavy I/O is deferred until the first access.
Source code in src/atlas/runtime.py
| def __init__(self, project_dir: str | None = None) -> None:
self.project_dir: str = os.path.abspath(project_dir or os.getcwd())
self.atlas_dir: str = os.path.join(self.project_dir, ".atlas")
self.warehouse_dir: str = self._find_warehouse()
# Lazy backing fields
self._manifest: dict | None = None
self._config: AtlasConfig | None = None
self._registry: dict | None = None
self._router: CategoryRouter | None = None
self._notes: dict | None = None
self._context: dict | None = None
|
property
True when the .atlas/ directory exists.
Clear all cached state.
Call after any operation that modifies .atlas/ so the next access
reloads fresh data from disk.
Source code in src/atlas/runtime.py
| def invalidate(self) -> None:
"""Clear all cached state.
Call after any operation that modifies .atlas/ so the next access
reloads fresh data from disk.
"""
self._manifest = None
self._config = None
self._registry = None
self._router = None
self._notes = None
self._context = None
|
Persist the in-memory manifest to .atlas/manifest.json.
Source code in src/atlas/runtime.py
| def save_manifest(self) -> None:
"""Persist the in-memory manifest to .atlas/manifest.json."""
if self._manifest is not None:
self._write_json(
os.path.join(self.atlas_dir, "manifest.json"), self._manifest
)
|
Persist the in-memory notes to .atlas/notes.json.
Source code in src/atlas/runtime.py
| def save_notes(self) -> None:
"""Persist the in-memory notes to .atlas/notes.json."""
if self._notes is not None:
self._write_json(
os.path.join(self.atlas_dir, "notes.json"), self._notes
)
|
save_config(data: dict) -> None
Persist data to .atlas/config.json.
Source code in src/atlas/runtime.py
| def save_config(self, data: dict) -> None:
"""Persist *data* to .atlas/config.json."""
self._write_json(os.path.join(self.atlas_dir, "config.json"), data)
|
query(
contexts: list[list[str]], message: str | None = None
) -> str
Retrieve pre-built markdown for one or more context groups.
Each group is [module_name, *filter_words]. Results are
concatenated with a blank line separator.
Source code in src/atlas/runtime.py
| def query(
self,
contexts: list[list[str]],
message: str | None = None,
) -> str:
"""Retrieve pre-built markdown for one or more context groups.
Each group is ``[module_name, *filter_words]``. Results are
concatenated with a blank line separator.
"""
parts: list[str] = []
retrieve_dir = os.path.join(self.atlas_dir, "retrieve")
installed = self.manifest.get("installed_modules", {})
for group in contexts:
if not group:
continue
module_name = group[0]
filters = group[1:]
# "status" is a virtual module — build live with dynamic data
if module_name == "status":
if not self.is_initialized:
continue
active_task = self.context.get("active")
history = self._read_recent_history(limit=5)
git_status = ""
if self.router.has_category_installed("vcs"):
git_status = self._quick_git_status()
content = build_status_file(
self.manifest,
installed,
active_task=active_task,
recent_activity=history if history else None,
git_status=git_status,
)
else:
# Read the pre-built file
md_path = os.path.join(retrieve_dir, f"{module_name}.md")
if os.path.isfile(md_path):
try:
with open(md_path) as f:
content = f.read()
except OSError:
content = ""
else:
# Fall back to building on-the-fly if not pre-built
content = build_retrieve_file(
module_name,
self.atlas_dir,
self.registry,
self.warehouse_dir,
installed,
)
if filters:
content = filter_sections(content, filters)
# Append module notes
module_notes = self.notes.get(module_name, [])
if module_notes:
note_lines = "\n".join(
f" • {n['text']}" for n in module_notes
)
content += f"\n\n⚠️ Project Notes:\n{note_lines}"
if content:
parts.append(content)
result = "\n\n".join(parts)
if message:
result = f"{result}\n\n---\n{message}" if result else message
return result
|
add_modules(names: list[str]) -> dict
Install one or more modules from the warehouse.
Source code in src/atlas/runtime.py
| def add_modules(self, names: list[str]) -> dict:
"""Install one or more modules from the warehouse."""
if not self.is_initialized:
return error_result("NOT_INITIALIZED", "run atlas init first")
installed: list[str] = []
failed: list[dict] = []
pkg_mgr = self.manifest.get("detected", {}).get("package_manager", "")
for name in names:
result = install_module(
name,
self.registry,
self.warehouse_dir,
self.atlas_dir,
self.manifest,
package_manager=pkg_mgr,
)
if result["ok"]:
installed.append(name)
else:
failed.append({"name": name, "error": result.get("error", "")})
if installed:
self.save_manifest()
retrieve_dir = os.path.join(self.atlas_dir, "retrieve")
os.makedirs(retrieve_dir, exist_ok=True)
for name in installed:
content = build_retrieve_file(
name,
self.atlas_dir,
self.registry,
self.warehouse_dir,
self.manifest.get("installed_modules", {}),
)
if content:
with open(os.path.join(retrieve_dir, f"{name}.md"), "w") as f:
f.write(content)
self.invalidate()
self._append_history(f"add {', '.join(installed)}")
return ok_result(installed=installed, failed=failed)
|
remove_module(name: str) -> dict
Uninstall a module from the project.
Source code in src/atlas/runtime.py
| def remove_module(self, name: str) -> dict:
"""Uninstall a module from the project."""
if not self.is_initialized:
return error_result("NOT_INITIALIZED", "run atlas init first")
result = remove_module(name, self.registry, self.atlas_dir, self.manifest)
if result["ok"]:
self.save_manifest()
self.invalidate()
self._append_history(f"remove {name}")
return result
|
just(
task_name: str, extra_args: list[str] | None = None
) -> dict
Execute a named task from installed module commands.
Source code in src/atlas/runtime.py
| def just(self, task_name: str, extra_args: list[str] | None = None) -> dict:
"""Execute a named task from installed module commands."""
if not self.is_initialized:
return error_result("NOT_INITIALIZED", "run atlas init first")
if not task_name:
return error_result("INVALID_ARGUMENT", "task name required")
installed_mods = self.manifest.get("installed_modules", {})
command: str | None = None
for mod_name in installed_mods:
mod_json = self._load_json(
os.path.join(self.atlas_dir, "modules", f"{mod_name}.json"), {}
)
cmds = mod_json.get("commands", {})
if task_name in cmds:
command = cmds[task_name]
break
if command is None:
return error_result(
"INVALID_ARGUMENT",
f"Task '{task_name}' not found in any installed module",
)
result = run_task(task_name, command, self.project_dir)
# Augment output on failure with error code hints from installed modules
if result.get("ok") and result.get("returncode", 0) != 0:
merged_codes: dict[str, str] = {}
for mod_name in installed_mods:
mod_json = self._load_json(
os.path.join(self.atlas_dir, "modules", f"{mod_name}.json"), {}
)
merged_codes.update(mod_json.get("error_codes", {}))
if merged_codes:
result["output"] = augment_errors(result["output"], merged_codes)
if result.get("ok"):
self._append_history(f"just {task_name}")
return result
|
add_note(module_name: str, text: str) -> dict
Append a note to module_name.
Source code in src/atlas/runtime.py
| def add_note(self, module_name: str, text: str) -> dict:
"""Append a note to *module_name*."""
if not self.is_initialized:
return error_result("NOT_INITIALIZED", "run atlas init first")
notes_list = self.notes.setdefault(module_name, [])
notes_list.append({"text": text})
self.save_notes()
self._append_history(f"note add {module_name}")
return ok_result(module=module_name, note=text, index=len(notes_list) - 1)
|
remove_note(module_name: str, index: int | str) -> dict
Remove a note by index (or all notes) from module_name.
Source code in src/atlas/runtime.py
| def remove_note(self, module_name: str, index: int | str) -> dict:
"""Remove a note by index (or all notes) from *module_name*."""
if not self.is_initialized:
return error_result("NOT_INITIALIZED", "run atlas init first")
notes_list = self.notes.get(module_name, [])
if not notes_list:
return error_result("INVALID_ARGUMENT", f"No notes for module '{module_name}'")
if index == "all":
self.notes[module_name] = []
else:
try:
idx = int(index)
notes_list.pop(idx)
except (ValueError, IndexError):
return error_result(
"INVALID_ARGUMENT",
f"Invalid note index '{index}' for module '{module_name}'",
)
self.save_notes()
self._append_history(f"note remove {module_name}")
return ok_result(module=module_name, removed=index)
|
build_session_brief() -> str
Build the auto-brief text for MCP prompt injection.
Source code in src/atlas/runtime.py
| def build_session_brief(self) -> str:
"""Build the auto-brief text for MCP prompt injection."""
parts: list[str] = []
detected = self.manifest.get("detected", {})
parts.append(f"# Atlas — {detected.get('project_name', 'project')}")
parts.append(f"Installed: {', '.join(self.installed_modules)}")
if self.context.get("active"):
task = self.context["active"]
parts.append(
f"\n## Active Task\n→ {task['type']} #{task['id']}: {task['title']}"
)
# Recent activity (from history.jsonl — last 3-5 entries)
history = self._read_recent_history(limit=5)
if history:
parts.append("\n## Recent Activity")
for entry in history:
parts.append(f" {entry.get('ago', '?')}: {entry.get('summary', '')}")
# Git status (quick subprocess if git installed)
if self.router.has_category_installed("vcs"):
git_status = self._quick_git_status()
if git_status:
parts.append(f"\n## Git Status\n{git_status}")
all_notes: list[str] = []
for mod, note_list in self.notes.items():
for note in note_list:
all_notes.append(f" ⚠️ {mod}: {note['text']}")
if all_notes:
parts.append("\n## Notes\n" + "\n".join(all_notes))
parts.append("\n## Atlas Tool")
parts.append(" One tool: atlas. Spaces filter, commas combine, -- separates.")
parts.append(f" Retrieve: {', '.join(self.installed_modules[:5])}")
parts.append(" Help: atlas list")
return "\n".join(parts)
|