manager

manager

Persistent agent lifecycle manager.

Composition layer — stores agent state in SQLite, delegates all computation to the five existing primitives (Intelligence, Agent, Tools, Engine, Learning).

Classes

AgentManager

AgentManager(db_path: str, *, clear_stale_running: bool = False)

Persistent agent lifecycle manager with SQLite backing.

Source code in src/openjarvis/agents/manager.py

def __init__(self, db_path: str, *, clear_stale_running: bool = False) -> None:
    self._db_path = str(db_path)
    self._db_lock = threading.RLock()
    self._conn = sqlite3.connect(self._db_path, check_same_thread=False)
    self._conn.row_factory = sqlite3.Row
    self._conn.execute("PRAGMA journal_mode=WAL")
    self._conn.execute("PRAGMA foreign_keys=ON")
    self._conn.execute(_CREATE_AGENTS)
    self._conn.execute(_CREATE_TASKS)
    self._conn.execute(_CREATE_BINDINGS)
    self._conn.executescript(_CREATE_CHECKPOINTS)
    self._conn.executescript(_CREATE_MESSAGES)
    self._conn.executescript(_CREATE_LEARNING_LOG)
    self._conn.commit()
    # Schema migrations for runtime columns
    _MIGRATIONS = [
        "ALTER TABLE managed_agents ADD COLUMN total_tokens INTEGER DEFAULT 0",
        "ALTER TABLE managed_agents ADD COLUMN total_cost REAL DEFAULT 0",
        "ALTER TABLE managed_agents ADD COLUMN total_runs INTEGER DEFAULT 0",
        "ALTER TABLE managed_agents ADD COLUMN last_run_at REAL",
        "ALTER TABLE managed_agents ADD COLUMN last_activity_at REAL",
        "ALTER TABLE managed_agents ADD COLUMN stall_retries INTEGER DEFAULT 0",
        "ALTER TABLE managed_agents ADD COLUMN current_activity TEXT DEFAULT ''",
        "ALTER TABLE managed_agents ADD COLUMN input_tokens INTEGER DEFAULT 0",
        "ALTER TABLE managed_agents ADD COLUMN output_tokens INTEGER DEFAULT 0",
        # JSON-encoded array of {tool, arguments, result, success, latency}
        "ALTER TABLE agent_messages ADD COLUMN tool_calls TEXT",
    ]
    for migration in _MIGRATIONS:
        try:
            self._conn.execute(migration)
        except sqlite3.OperationalError:
            pass  # Column already exists
    self._conn.commit()
    # Only the authoritative long-running process (the API server, which
    # owns the scheduler) may sweep running→idle on boot. Short-lived CLI
    # commands (`jarvis agents list/info/...`) and the SystemBuilder path
    # used by `run`/`ask` MUST NOT: they share this DB with a server that
    # may be mid-tick, and an unconditional sweep here flips an actively
    # running agent back to "idle" — which is exactly why `list` reported
    # "idle" while a tick was running elsewhere. Zombies left by a crashed
    # worker are recovered lazily by start_tick()'s stale-lock overtake.
    if clear_stale_running:
        self._clear_stale_running_state()

Functions

start_tick

start_tick(agent_id: str) -> None

Mark agent as running. Raises ValueError if already running.

A row that has been running longer than _STALE_TICK_SECONDS without any update is treated as a zombie left by a dead worker and overtaken rather than refused — otherwise a crash with no server around to sweep it would wedge the agent forever.

The state transition is a single conditional SQLite update. The read-before-write approach is racy when two workers start together: both can observe idle and then mark the row running. The conditional update lets SQLite choose exactly one winner, including when workers use separate connections or processes.

Source code in src/openjarvis/agents/manager.py

@_db_locked
def start_tick(self, agent_id: str) -> None:
    """Mark agent as running. Raises ValueError if already running.

    A row that has been ``running`` longer than ``_STALE_TICK_SECONDS``
    without any update is treated as a zombie left by a dead worker and
    overtaken rather than refused — otherwise a crash with no server
    around to sweep it would wedge the agent forever.

    The state transition is a single conditional SQLite update.  The
    read-before-write approach is racy when two workers start together:
    both can observe ``idle`` and then mark the row ``running``.  The
    conditional update lets SQLite choose exactly one winner, including
    when workers use separate connections or processes.
    """
    while True:
        now = time.time()
        stale_before = now - self._STALE_TICK_SECONDS
        agent = self.get_agent(agent_id)
        if agent is None:
            return

        was_stale = (
            agent["status"] == "running"
            and (agent.get("updated_at") or 0) <= stale_before
        )
        cursor = self._conn.execute(
            "UPDATE managed_agents SET status = 'running', "
            "updated_at = ? WHERE id = ? AND "
            "(status != 'running' OR updated_at <= ?)",
            (now, agent_id, stale_before),
        )
        self._conn.commit()
        if cursor.rowcount:
            if was_stale:
                age = time.time() - (agent.get("updated_at") or 0)
                logger.warning(
                    "Agent %s: overtaking stale tick lock "
                    "(running, idle for %.0fs)",
                    agent_id,
                    age,
                )
            return

        # Another connection acquired the lock after our read.  If that
        # lock is still fresh, this caller loses; if it ended between the
        # failed update and this read, retry so the caller never proceeds
        # without owning the tick lock.
        current = self.get_agent(agent_id)
        if current is None:
            return
        if current["status"] == "running":
            age = time.time() - (current.get("updated_at") or 0)
            if age < self._STALE_TICK_SECONDS:
                raise ValueError(f"Agent {agent_id} is already executing a tick")

find_binding_for_channel

find_binding_for_channel(channel_type: str, channel_id: str) -> Optional[Dict[str, Any]]

Find a dedicated binding for a specific channel.

Source code in src/openjarvis/agents/manager.py

@_db_locked
def find_binding_for_channel(
    self, channel_type: str, channel_id: str
) -> Optional[Dict[str, Any]]:
    """Find a dedicated binding for a specific channel."""
    rows = self._conn.execute(
        "SELECT * FROM channel_bindings WHERE channel_type = ?",
        (channel_type,),
    ).fetchall()
    for row in rows:
        binding = self._row_to_binding(row)
        config = binding.get("config", {})
        if config.get("channel") == channel_id:
            return binding
    return None

list_templates staticmethod

list_templates() -> List[Dict[str, Any]]

Discover built-in and user templates.

Source code in src/openjarvis/agents/manager.py

@staticmethod
def list_templates() -> List[Dict[str, Any]]:
    """Discover built-in and user templates."""
    import importlib.resources

    try:
        import tomllib
    except ModuleNotFoundError:
        import tomli as tomllib  # type: ignore[no-redef]

    templates: List[Dict[str, Any]] = []

    # Built-in templates
    try:
        tpl_dir = importlib.resources.files("openjarvis.agents") / "templates"
        for item in tpl_dir.iterdir():
            if str(item).endswith(".toml"):
                data = tomllib.loads(item.read_text(encoding="utf-8"))
                tpl = data.get("template", {})
                tpl["source"] = "built-in"
                templates.append(tpl)
    except Exception:
        pass

    # User templates
    user_dir = get_config_dir() / "templates"
    if user_dir.is_dir():
        for f in user_dir.glob("*.toml"):
            try:
                data = tomllib.loads(f.read_text(encoding="utf-8"))
                tpl = data.get("template", {})
                tpl["source"] = "user"
                templates.append(tpl)
            except Exception:
                pass

    return templates

create_from_template

create_from_template(template_id: str, name: str, overrides: Optional[Dict[str, Any]] = None) -> Dict[str, Any]

Create an agent from a template with optional overrides.

Source code in src/openjarvis/agents/manager.py

def create_from_template(
    self, template_id: str, name: str, overrides: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
    """Create an agent from a template with optional overrides."""
    templates = self.list_templates()
    tpl = next((t for t in templates if t.get("id") == template_id), None)
    if not tpl:
        raise ValueError(f"Template not found: {template_id}")
    skip = {"id", "name", "description", "source"}
    config = {k: v for k, v in tpl.items() if k not in skip}
    if overrides:
        config.update(overrides)
    agent_type = config.pop("agent_type", "monitor_operative")

    # Expand system_prompt_template with instruction
    prompt_tpl = config.pop("system_prompt_template", "")
    if prompt_tpl:
        instruction = config.get("instruction", "")
        config["system_prompt"] = prompt_tpl.format(
            instruction=instruction or "(No specific instruction provided)",
        )

    return self.create_agent(name=name, agent_type=agent_type, config=config)

store_agent_response

store_agent_response(agent_id: str, content: str, tool_calls: Optional[list] = None) -> dict

Store an agent-to-user response message.

tool_calls is an optional list of {tool, arguments, result, success, latency} dicts captured during the turn. They are stored as JSON alongside the message so the UI can replay them after a page reload.

Source code in src/openjarvis/agents/manager.py

@_db_locked
def store_agent_response(
    self,
    agent_id: str,
    content: str,
    tool_calls: Optional[list] = None,
) -> dict:
    """Store an agent-to-user response message.

    ``tool_calls`` is an optional list of ``{tool, arguments, result,
    success, latency}`` dicts captured during the turn. They are stored
    as JSON alongside the message so the UI can replay them after a
    page reload.
    """
    msg_id = uuid4().hex[:16]
    now = time.time()
    tool_calls_json = json.dumps(tool_calls) if tool_calls else None
    self._conn.execute(
        "INSERT INTO agent_messages"
        " (id, agent_id, direction, content, mode, status, created_at,"
        " tool_calls)"
        " VALUES (?, ?, 'agent_to_user', ?, 'immediate', 'delivered', ?, ?)",
        (msg_id, agent_id, content, now, tool_calls_json),
    )
    self._conn.commit()
    return {
        "id": msg_id,
        "agent_id": agent_id,
        "direction": "agent_to_user",
        "content": content,
        "mode": "immediate",
        "status": "delivered",
        "created_at": now,
        "tool_calls": tool_calls or None,
    }

Functions