Scott.Idem/Cortex-Inara
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: multi-user/multi-persona support with two-level home directory layout

Browse Source 
Restructures persona storage from a flat personas/{name}/ layout to
home/{username}/persona/{name}/, mirroring Linux home directories.

Changes:
- persona.py: two ContextVars (user + persona), Linux-style name validation,
  set_context(), get_user(), get_persona(), validate(), list_users(),
  list_user_personas(); persona_path() takes (username, name)
- config.py: replaces personas_dir with home_dir + home_root()
- git mv personas/inara → home/scott/persona/inara (history preserved)
- home/holly/persona/tina/: Holly's persona stub added
- cron_runner.py: all storage functions take (username, persona) params
- tools/cron.py: stamps user + persona on jobs; APScheduler IDs are
  {user}:{persona}:{job_id} to prevent collisions across users
- memory_distiller.py: distill_short/mid/long take (username, persona);
  added missing Path + settings imports
- scheduler.py: _load_user_crons() iterates home/*/persona/* (two-level)
- routers/chat.py, orchestrator.py: user field added; set_context() called
- tests/conftest.py: home_root fixture with two-level structure;
  patches home_dir instead of personas_dir
- tests/test_persona.py: fully rewritten for two-level API
- tests/test_api_files.py: updated fixture name and path
- .env.default: documents HOME_DIR setting; scrubs stale API key
- CLAUDE.md, README.md: directory maps updated for new layout

All 80 tests pass.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Scott Idem
2026-03-20 22:35:40 -04:00
parent 92a8f5d894
commit 77e770cdb2
51 changed files with 463 additions and 208 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
cortex/config.py
View File

@@ -27,8 +27,7 @@ class Settings(BaseSettings):
agent_name: str = "Inara"
user_name: str = "Scott"
personas_dir: Path = Path("../personas")
inara_dir: Path = Path("../personas/inara") # legacy — use personas_dir
home_dir: Path = Path("../home")
sessions_dir: Path = Path("./data/sessions")
default_model: str = "claude-sonnet-4-6"
default_tier: int = 2
@@ -74,15 +73,11 @@ class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")
def personas_root(self) -> Path:
"""Resolve personas_dir relative to this file's location if not absolute."""
if self.personas_dir.is_absolute():
return self.personas_dir
return (Path(__file__).parent / self.personas_dir).resolve()
def inara_path(self) -> Path:
"""Legacy helper — returns the inara persona directory. Prefer persona_path()."""
return self.personas_root() / "inara"
def home_root(self) -> Path:
"""Resolve home_dir relative to this file's location if not absolute."""
if self.home_dir.is_absolute():
return self.home_dir
return (Path(__file__).parent / self.home_dir).resolve()
def sessions_path(self) -> Path:
"""Resolve sessions_dir relative to this file's location if not absolute."""

22
cortex/cron_runner.py
View File

@@ -45,12 +45,12 @@ _DOW = {
# Storage
# ---------------------------------------------------------------------------
def crons_path(persona: str | None = None) -> Path:
return _persona_path(persona) / "CRONS.json"
def crons_path(username: str | None = None, persona: str | None = None) -> Path:
return _persona_path(username, persona) / "CRONS.json"
def load_crons(persona: str | None = None) -> list[dict]:
p = crons_path(persona)
def load_crons(username: str | None = None, persona: str | None = None) -> list[dict]:
p = crons_path(username, persona)
if not p.exists():
return []
try:
@@ -60,9 +60,11 @@ def load_crons(persona: str | None = None) -> list[dict]:
return []
def save_crons(crons: list[dict], persona: str | None = None) -> None:
def save_crons(crons: list[dict],
username: str | None = None,
persona: str | None = None) -> None:
import json
crons_path(persona).write_text(json.dumps(crons, indent=2) + "\n")
crons_path(username, persona).write_text(json.dumps(crons, indent=2) + "\n")
# ---------------------------------------------------------------------------
@@ -134,7 +136,7 @@ async def run_job(job: dict) -> None:
label = job.get("label", job.get("id", "cron"))
section = f"\n## {label} — {_now_label()}\n\n{payload}\n"
p_root = _persona_path(job.get("persona"))
p_root = _persona_path(job.get("user"), job.get("persona"))
if job_type == "remind":
p = p_root / "REMINDERS.md"
@@ -153,10 +155,10 @@ async def run_job(job: dict) -> None:
return
# Record last_run in the right persona's CRONS.json
persona = job.get("persona")
crons = load_crons(persona)
u, p = job.get("user"), job.get("persona")
crons = load_crons(u, p)
for c in crons:
if c["id"] == job["id"]:
c["last_run"] = datetime.now(timezone.utc).isoformat()
break
save_crons(crons, persona)
save_crons(crons, u, p)

14
cortex/memory_distiller.py
View File

@@ -7,6 +7,8 @@ Inara tiered memory distillation.
"""
import logging
from datetime import datetime
from pathlib import Path
from config import settings
from persona import persona_path as _persona_path
logger = logging.getLogger(__name__)
@@ -23,14 +25,14 @@ def _read(path: Path) -> str:
return path.read_text() if path.exists() else ""
def distill_short(persona: str | None = None) -> dict:
def distill_short(username: str | None = None, persona: str | None = None) -> dict:
"""
Roll the most recent session log files into MEMORY_SHORT.md.
No LLM involved — pure aggregation with budget truncation.
Files are included newest-first until the budget is reached,
then written in chronological order (oldest first).
"""
inara_dir = _persona_path(persona)
inara_dir = _persona_path(username, persona)
sessions_dir = inara_dir / "sessions"
budget = _budget_chars(settings.memory_budget_short)
@@ -72,13 +74,13 @@ def distill_short(persona: str | None = None) -> dict:
}
async def distill_mid(persona: str | None = None) -> dict:
async def distill_mid(username: str | None = None, persona: str | None = None) -> dict:
"""
Ask the LLM to summarize MEMORY_SHORT.md → MEMORY_MID.md.
"""
from llm_client import complete
inara_dir = _persona_path(persona)
inara_dir = _persona_path(username, persona)
short_content = _read(inara_dir / "MEMORY_SHORT.md")
if not short_content.strip() or "Not yet populated" in short_content:
@@ -116,13 +118,13 @@ async def distill_mid(persona: str | None = None) -> dict:
}
async def distill_long(persona: str | None = None) -> dict:
async def distill_long(username: str | None = None, persona: str | None = None) -> dict:
"""
Ask the LLM to integrate MEMORY_MID.md into MEMORY_LONG.md.
"""
from llm_client import complete
inara_dir = _persona_path(persona)
inara_dir = _persona_path(username, persona)
long_content = _read(inara_dir / "MEMORY_LONG.md")
mid_content = _read(inara_dir / "MEMORY_MID.md")

135
cortex/persona.py
View File

@@ -1,23 +1,28 @@
"""
Persona routing — per-request identity context.
Two-level identity context — user + persona, modelled on OS home directories.
Each HTTP request sets the active persona via set_persona() at entry.
Everything downstream (context_loader, tools, session_logger) calls
persona_path() to get the right working directory without needing to
pass the name through every function signature.
Layout on disk:
home/
scott/
persona/
inara/ ← IDENTITY.md, SOUL.md, sessions/, TASKS.json, …
abc/ ← a second persona for the same user
holly/
persona/
tina/
Directory layout:
personas/
inara/ ← Scott's agent
holly/ ← Second persona (add IDENTITY.md + SOUL.md + USER.md)
...
Each HTTP request sets both user and persona via set_context() at entry.
Everything downstream calls persona_path() to get the right directory.
Background tasks (cron, distiller) pass both names explicitly.
Background tasks (cron runner, memory distiller) don't have a request
context — they pass persona by name explicitly to persona_path(name).
Naming rules — same as Linux usernames:
^[a-z_][a-z0-9_-]{0,31}$
Lowercase, start with letter or underscore, max 32 chars.
Examples: scott, holly, whatever_name_asian-v3
Future Aether integration: persona name maps to account_id_random.
Replace the directory lookup in persona_path() with a DB lookup at
that point; the ContextVar contract stays the same.
Future Aether integration: (user, persona) maps to (account_id, persona_id).
Replace the directory lookups in persona_path() / validate() with DB lookups;
the ContextVar contract stays identical.
"""
import re
@@ -26,53 +31,103 @@ from pathlib import Path
from config import settings
_current: ContextVar[str] = ContextVar("persona", default="inara")
_user: ContextVar[str] = ContextVar("user", default="scott")
_persona: ContextVar[str] = ContextVar("persona", default="inara")
# Only alphanumeric + underscore + hyphen, 1–32 chars. Prevents path traversal.
_VALID = re.compile(r"^[a-zA-Z0-9_-]{1,32}$")
# Same rules as Linux usernames.
_VALID = re.compile(r"^[a-z_][a-z0-9_-]{0,31}$")
def set_persona(name: str) -> None:
"""Set the active persona for the current async task/coroutine."""
_current.set(name)
# ---------------------------------------------------------------------------
# Context setters / getters
# ---------------------------------------------------------------------------
def set_context(username: str, persona_name: str) -> None:
"""Set the active user + persona for the current async task/coroutine."""
_user.set(username)
_persona.set(persona_name)
def get_user() -> str:
return _user.get()
def get_persona() -> str:
"""Return the active persona name for the current task."""
return _current.get()
return _persona.get()
def persona_path(name: str | None = None) -> Path:
# ---------------------------------------------------------------------------
# Path resolution
# ---------------------------------------------------------------------------
def persona_path(username: str | None = None, name: str | None = None) -> Path:
"""
Return the filesystem path for a persona's data directory.
If name is omitted, uses the persona set for the current request.
Pass name explicitly for background tasks (cron, distiller).
home/{username}/persona/{name}/
If either arg is omitted, falls back to the ContextVar for the current request.
Pass both explicitly for background tasks (cron, distiller).
"""
return settings.personas_root() / (name or _current.get())
u = username or _user.get()
p = name or _persona.get()
return settings.home_root() / u / "persona" / p
def list_personas() -> list[str]:
"""Return all persona names that have an IDENTITY.md (i.e. are real personas)."""
root = settings.personas_root()
# ---------------------------------------------------------------------------
# Discovery
# ---------------------------------------------------------------------------
def list_users() -> list[str]:
"""Return all usernames that have at least one persona."""
root = settings.home_root()
if not root.exists():
return []
return sorted(
d.name for d in root.iterdir()
if d.is_dir() and (d / "persona").is_dir()
)
def list_user_personas(username: str) -> list[str]:
"""Return all persona names for a given user (must have IDENTITY.md)."""
persona_root = settings.home_root() / username / "persona"
if not persona_root.exists():
return []
return sorted(
d.name for d in persona_root.iterdir()
if d.is_dir() and (d / "IDENTITY.md").exists()
)
def validate(name: str) -> str:
"""
Validate a persona name from an untrusted source (e.g. HTTP request).
Returns the name if valid, raises ValueError otherwise.
"""
# ---------------------------------------------------------------------------
# Validation
# ---------------------------------------------------------------------------
def _check_name(name: str, label: str) -> None:
if not _VALID.match(name):
raise ValueError(
f"Invalid persona name {name!r}. "
f"Use letters, digits, underscores, or hyphens (max 32 chars)."
f"Invalid {label} {name!r}. "
f"Use lowercase letters, digits, underscores, or hyphens "
f"(must start with letter/underscore, max 32 chars)."
)
if not (settings.personas_root() / name / "IDENTITY.md").exists():
raise ValueError(f"Unknown persona: {name!r}")
return name
def validate(username: str, persona_name: str) -> tuple[str, str]:
"""
Validate a (username, persona_name) pair from an untrusted source.
Returns (username, persona_name) if valid, raises ValueError otherwise.
Checks format first (blocks path traversal), then existence.
"""
_check_name(username, "username")
_check_name(persona_name, "persona")
user_dir = settings.home_root() / username
if not user_dir.is_dir():
raise ValueError(f"Unknown user: {username!r}")
persona_dir = user_dir / "persona" / persona_name
if not (persona_dir / "IDENTITY.md").exists():
raise ValueError(f"Unknown persona {persona_name!r} for user {username!r}")
return username, persona_name

6
cortex/routers/chat.py
View File

@@ -8,7 +8,7 @@ from llm_client import complete
from session_logger import log_turn
from session_store import load as load_session, save as save_session, list_all, generate_session_id, delete as delete_session
from config import settings
from persona import set_persona, validate as validate_persona
from persona import set_context, validate as validate_persona
import event_bus
@@ -23,6 +23,7 @@ class ChatRequest(BaseModel):
include_long: bool = True
include_mid: bool = True
include_short: bool = True
user: str = "scott"
persona: str = "inara"
@@ -52,7 +53,8 @@ async def _stream_chat(req: ChatRequest):
data: {"type": "error", "message": "..."}
"""
try:
set_persona(validate_persona(req.persona))
user, persona = validate_persona(req.user, req.persona)
set_context(user, persona)
except ValueError as e:
yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n"
return

6
cortex/routers/orchestrator.py
View File

@@ -20,7 +20,7 @@ from pydantic import BaseModel
from config import settings
from context_loader import load_context
from persona import set_persona, validate as validate_persona
from persona import set_context, validate as validate_persona
import orchestrator_engine
logger = logging.getLogger(__name__)
@@ -47,6 +47,7 @@ class OrchestrateRequest(BaseModel):
include_long: bool = True
include_mid: bool = True
include_short: bool = True
user: str = "scott"
persona: str = "inara"
@@ -77,7 +78,8 @@ class JobStatusResponse(BaseModel):
async def orchestrate(req: OrchestrateRequest) -> OrchestrateResponse:
"""Submit a task to the orchestrator. Returns a job_id to poll."""
try:
set_persona(validate_persona(req.persona))
user, persona = validate_persona(req.user, req.persona)
set_context(user, persona)
except ValueError as e:
from fastapi import HTTPException
raise HTTPException(status_code=400, detail=str(e))

48
cortex/scheduler.py
View File

@@ -84,38 +84,42 @@ def start() -> None:
def _load_user_crons() -> None:
"""Register all enabled user-defined cron jobs across all personas."""
"""Register all enabled user-defined cron jobs across all users and personas."""
import asyncio
try:
from cron_runner import load_crons, parse_schedule, run_job
from persona import list_personas
from persona import list_users, list_user_personas
except ImportError as e:
logger.warning("could not import cron modules: %s", e)
return
total = 0
for persona_name in list_personas():
for job in load_crons(persona_name):
if not job.get("enabled", True):
continue
# Ensure persona is stamped on the job for run_job() to resolve paths
job.setdefault("persona", persona_name)
try:
kwargs = parse_schedule(job["schedule"])
sched_id = f"{persona_name}:{job['id']}"
_scheduler.add_job(
lambda j=job: asyncio.ensure_future(run_job(j)),
"cron",
id=sched_id,
replace_existing=True,
**kwargs,
)
total += 1
except Exception as e:
logger.warning("cron %s/%s skipped: %s", persona_name, job.get("id"), e)
persona_count = 0
for username in list_users():
for persona_name in list_user_personas(username):
persona_count += 1
for job in load_crons(username, persona_name):
if not job.get("enabled", True):
continue
# Ensure user + persona are stamped on the job for run_job() path resolution
job.setdefault("user", username)
job.setdefault("persona", persona_name)
try:
kwargs = parse_schedule(job["schedule"])
sched_id = f"{username}:{persona_name}:{job['id']}"
_scheduler.add_job(
lambda j=job: asyncio.ensure_future(run_job(j)),
"cron",
id=sched_id,
replace_existing=True,
**kwargs,
)
total += 1
except Exception as e:
logger.warning("cron %s/%s/%s skipped: %s", username, persona_name, job.get("id"), e)
if total:
logger.info("loaded %d user cron job(s) across %d persona(s)", total, len(list_personas()))
logger.info("loaded %d user cron job(s) across %d persona(s)", total, persona_count)
def stop() -> None:

36
cortex/tests/conftest.py
View File

@@ -2,10 +2,19 @@
Shared fixtures for Cortex test suite.
Key design choices:
- All file I/O goes to a tmp_path, never touching personas/inara/ or real sessions.
- All file I/O goes to a tmp_path, never touching home/ or real sessions.
- LLM calls are mocked by default — tests are fast and deterministic.
- The 'app' fixture patches settings before importing main, so all modules
- The 'client' fixture patches settings before importing main, so all modules
see the temp directory.
Home layout mirrors the two-level structure:
tmp/
scott/
persona/
inara/ ← the default test persona
holly/
persona/
tina/
"""
import json
@@ -19,19 +28,21 @@ from httpx import ASGITransport
# ---------------------------------------------------------------------------
# Temp persona directory
# Temp home directory
# ---------------------------------------------------------------------------
@pytest.fixture(scope="session")
def personas_root(tmp_path_factory) -> Path:
"""A temp personas/ dir with a minimal 'inara' persona for testing."""
root = tmp_path_factory.mktemp("personas")
_make_persona(root, "inara", "Inara", "Scott")
def home_root(tmp_path_factory) -> Path:
"""A temp home/ dir with minimal user/persona stubs for testing."""
root = tmp_path_factory.mktemp("home")
_make_persona(root, "scott", "inara", "Inara", "Scott")
_make_persona(root, "holly", "tina", "Tina", "Holly")
return root
def _make_persona(root: Path, name: str, agent: str, user: str) -> Path:
p = root / name
def _make_persona(root: Path, username: str, persona: str,
agent: str, user: str) -> Path:
p = root / username / "persona" / persona
p.mkdir(parents=True, exist_ok=True)
(p / "IDENTITY.md").write_text(f"# {agent}\nTest identity for {agent}.")
(p / "SOUL.md").write_text(f"# Soul\nTest soul for {agent}.")
@@ -54,7 +65,7 @@ def _make_persona(root: Path, name: str, agent: str, user: str) -> Path:
# ---------------------------------------------------------------------------
@pytest_asyncio.fixture
async def client(personas_root, tmp_path):
async def client(home_root, tmp_path):
"""HTTPX async test client against the live ASGI app with patched paths."""
import config
import persona as persona_mod
@@ -63,13 +74,12 @@ async def client(personas_root, tmp_path):
sessions_dir.mkdir()
with (
patch.object(config.settings, "personas_dir", personas_root),
patch.object(config.settings, "home_dir", home_root),
patch.object(config.settings, "sessions_dir", sessions_dir),
patch("scheduler.start"), # don't run APScheduler in tests
patch("scheduler.stop"),
):
# Force persona module to re-read patched settings
persona_mod._current.set("inara")
persona_mod.set_context("scott", "inara")
from main import app
async with httpx.AsyncClient(

4
cortex/tests/test_api_files.py
View File

@@ -50,10 +50,10 @@ async def test_files_put_not_allowed(client):
@pytest.mark.anyio
async def test_files_get_missing_but_allowed(client, personas_root):
async def test_files_get_missing_but_allowed(client, home_root):
"""An allowed file that doesn't exist yet returns 404."""
# Temporarily remove MEMORY_MID.md
f = personas_root / "inara" / "MEMORY_MID.md"
f = home_root / "scott" / "persona" / "inara" / "MEMORY_MID.md"
existed = f.exists()
if existed:
backup = f.read_text()

131
cortex/tests/test_persona.py
View File

@@ -1,5 +1,6 @@
"""
Unit tests for persona.py — validation, routing, path traversal.
Tests the two-level home/{username}/persona/{name}/ structure.
No HTTP involved.
"""
import pytest
@@ -7,88 +8,118 @@ from pathlib import Path
from unittest.mock import patch
def _make_temp_personas(tmp_path: Path) -> Path:
root = tmp_path / "personas"
for name in ("alice", "bob"):
p = root / name
def _make_home(tmp_path: Path) -> Path:
"""Create a minimal home/ tree with two users and some personas."""
root = tmp_path / "home"
for username, persona in [("scott", "inara"), ("holly", "tina"), ("scott", "alt")]:
p = root / username / "persona" / persona
p.mkdir(parents=True)
(p / "IDENTITY.md").write_text(f"# {name}")
# A directory WITHOUT IDENTITY.md — should not appear in list_personas()
(root / "incomplete").mkdir()
(p / "IDENTITY.md").write_text(f"# {persona}")
# A persona dir WITHOUT IDENTITY.md — should be invisible to list_user_personas()
incomplete = root / "scott" / "persona" / "broken"
incomplete.mkdir(parents=True)
return root
def test_validate_good(tmp_path):
root = _make_temp_personas(tmp_path)
root = _make_home(tmp_path)
import config, persona
with patch.object(config.settings, "personas_dir", root):
assert persona.validate("alice") == "alice"
assert persona.validate("bob") == "bob"
with patch.object(config.settings, "home_dir", root):
assert persona.validate("scott", "inara") == ("scott", "inara")
assert persona.validate("holly", "tina") == ("holly", "tina")
def test_validate_unknown(tmp_path):
root = _make_temp_personas(tmp_path)
def test_validate_unknown_user(tmp_path):
root = _make_home(tmp_path)
import config, persona
with patch.object(config.settings, "personas_dir", root):
with patch.object(config.settings, "home_dir", root):
with pytest.raises(ValueError, match="Unknown user"):
persona.validate("charlie", "inara")
def test_validate_unknown_persona(tmp_path):
root = _make_home(tmp_path)
import config, persona
with patch.object(config.settings, "home_dir", root):
with pytest.raises(ValueError, match="Unknown persona"):
persona.validate("charlie")
persona.validate("scott", "ghost")
def test_validate_path_traversal(tmp_path):
root = _make_temp_personas(tmp_path)
root = _make_home(tmp_path)
import config, persona
with patch.object(config.settings, "personas_dir", root):
with pytest.raises(ValueError, match="Invalid persona name"):
persona.validate("../../etc/passwd")
with pytest.raises(ValueError, match="Invalid persona name"):
persona.validate("../alice")
with pytest.raises(ValueError, match="Invalid persona name"):
persona.validate("alice/../../etc")
with patch.object(config.settings, "home_dir", root):
for bad in ("../../etc/passwd", "../scott", "scott/../../etc"):
with pytest.raises(ValueError, match="Invalid"):
persona.validate(bad, "inara")
with pytest.raises(ValueError, match="Invalid"):
persona.validate("scott", "../../etc/passwd")
def test_validate_special_chars(tmp_path):
root = _make_temp_personas(tmp_path)
root = _make_home(tmp_path)
import config, persona
with patch.object(config.settings, "personas_dir", root):
with patch.object(config.settings, "home_dir", root):
for bad in ("alice bob", "alice;bob", "alice\x00bob", "A" * 33, ""):
with pytest.raises(ValueError):
persona.validate(bad)
persona.validate(bad, "inara")
def test_validate_allows_hyphen_underscore(tmp_path):
root = _make_temp_personas(tmp_path)
# Create a persona with hyphen and underscore in name
p = root / "my_ai-agent"
root = _make_home(tmp_path)
p = root / "my_user" / "persona" / "my-agent"
p.mkdir(parents=True)
(p / "IDENTITY.md").write_text("# My Agent")
import config, persona
with patch.object(config.settings, "personas_dir", root):
assert persona.validate("my_ai-agent") == "my_ai-agent"
with patch.object(config.settings, "home_dir", root):
assert persona.validate("my_user", "my-agent") == ("my_user", "my-agent")
def test_list_personas(tmp_path):
root = _make_temp_personas(tmp_path)
def test_list_users(tmp_path):
root = _make_home(tmp_path)
import config, persona
with patch.object(config.settings, "personas_dir", root):
names = persona.list_personas()
assert "alice" in names
assert "bob" in names
assert "incomplete" not in names # no IDENTITY.md
with patch.object(config.settings, "home_dir", root):
users = persona.list_users()
assert "scott" in users
assert "holly" in users
def test_persona_path_uses_contextvar(tmp_path):
root = _make_temp_personas(tmp_path)
def test_list_user_personas(tmp_path):
root = _make_home(tmp_path)
import config, persona
with patch.object(config.settings, "personas_dir", root):
persona.set_persona("alice")
assert persona.persona_path() == root / "alice"
persona.set_persona("bob")
assert persona.persona_path() == root / "bob"
with patch.object(config.settings, "home_dir", root):
names = persona.list_user_personas("scott")
assert "inara" in names
assert "alt" in names
assert "broken" not in names # no IDENTITY.md
def test_persona_path_explicit_name(tmp_path):
root = _make_temp_personas(tmp_path)
def test_persona_path_uses_contextvars(tmp_path):
root = _make_home(tmp_path)
import config, persona
with patch.object(config.settings, "personas_dir", root):
persona.set_persona("alice")
assert persona.persona_path("bob") == root / "bob"
with patch.object(config.settings, "home_dir", root):
persona.set_context("scott", "inara")
assert persona.persona_path() == root / "scott" / "persona" / "inara"
persona.set_context("holly", "tina")
assert persona.persona_path() == root / "holly" / "persona" / "tina"
def test_persona_path_explicit_args(tmp_path):
root = _make_home(tmp_path)
import config, persona
with patch.object(config.settings, "home_dir", root):
persona.set_context("scott", "inara")
# Explicit args override the ContextVar
assert persona.persona_path("holly", "tina") == root / "holly" / "persona" / "tina"
# ContextVar unchanged
assert persona.persona_path() == root / "scott" / "persona" / "inara"
def test_get_user_and_persona(tmp_path):
import persona
persona.set_context("scott", "inara")
assert persona.get_user() == "scott"
assert persona.get_persona() == "inara"
persona.set_context("holly", "tina")
assert persona.get_user() == "holly"
assert persona.get_persona() == "tina"

28
cortex/tools/cron.py
View File

@@ -17,7 +17,7 @@ import secrets
from datetime import datetime, timezone
from pathlib import Path
from persona import persona_path, get_persona
from persona import persona_path, get_user, get_persona
from cron_runner import load_crons, save_crons, parse_schedule
@@ -34,7 +34,7 @@ def _short_id() -> str:
# ---------------------------------------------------------------------------
def _cron_list() -> str:
crons = load_crons()
crons = load_crons(get_user(), get_persona())
if not crons:
return "No crons scheduled."
@@ -60,10 +60,12 @@ def _cron_add(label: str, schedule: str, job_type: str, payload: str) -> str:
if job_type not in ("remind", "note"):
return "Bad type: must be 'remind' or 'note'."
crons = load_crons()
current_user = get_user()
current_persona = get_persona()
crons = load_crons(current_user, current_persona)
job = {
"id": _short_id(),
"user": current_user,
"persona": current_persona,
"label": label,
"schedule": schedule,
@@ -74,35 +76,37 @@ def _cron_add(label: str, schedule: str, job_type: str, payload: str) -> str:
"last_run": None,
}
crons.append(job)
save_crons(crons, current_persona)
save_crons(crons, current_user, current_persona)
# Register with the live scheduler
_scheduler_add(job, sched_kwargs)
return f"Created and scheduled: {job['id']} {schedule} [{job_type}] {label} (persona: {current_persona})"
return f"Created and scheduled: {job['id']} {schedule} [{job_type}] {label} (user: {current_user}, persona: {current_persona})"
def _cron_remove(cron_id: str) -> str:
user = get_user()
persona = get_persona()
crons = load_crons(persona)
crons = load_crons(user, persona)
before = len(crons)
crons = [c for c in crons if c["id"] != cron_id]
if len(crons) == before:
return f"Not found: {cron_id}"
save_crons(crons, persona)
_scheduler_remove(f"{persona}:{cron_id}")
save_crons(crons, user, persona)
_scheduler_remove(f"{user}:{persona}:{cron_id}")
return f"Removed: {cron_id}"
def _cron_toggle(cron_id: str) -> str:
user = get_user()
persona = get_persona()
crons = load_crons(persona)
crons = load_crons(user, persona)
for c in crons:
if c["id"] == cron_id:
c["enabled"] = not c.get("enabled", True)
save_crons(crons, persona)
save_crons(crons, user, persona)
action = "resumed" if c["enabled"] else "paused"
sched_id = f"{persona}:{cron_id}"
sched_id = f"{user}:{persona}:{cron_id}"
_scheduler_resume(sched_id) if c["enabled"] else _scheduler_pause(sched_id)
return f"{action.capitalize()}: {cron_id} {c['label']}"
return f"Not found: {cron_id}"
@@ -125,7 +129,7 @@ def _scheduler_add(job: dict, sched_kwargs: dict) -> None:
from cron_runner import run_job
s = sched_module.get_scheduler()
if s and s.running:
sched_id = f"{job.get('persona', 'inara')}:{job['id']}"
sched_id = f"{job.get('user', 'scott')}:{job.get('persona', 'inara')}:{job['id']}"
s.add_job(
lambda j=job: asyncio.ensure_future(run_job(j)),
"cron",