Per-User Scheduler Isolation

Pass a principal when creating a scheduled automation or a suggestion, and every list / accept / dismiss / remove call is automatically scoped to that owner — one gateway user’s jobs are invisible to another’s.

Quick Start

Create a job for a specific user

Set principal on a ScheduleJob. store.list(principal="alice") returns only Alice’s job, while store.list() returns everything.

from praisonaiagents.scheduler import ScheduleJob, Schedule
from praisonaiagents.scheduler.config_store import ConfigYamlScheduleStore

store = ConfigYamlScheduleStore()

store.add(ScheduleJob(
    name="morning-brief",
    schedule=Schedule(kind="cron", cron_expr="0 8 * * *"),
    message="Send Alice her morning brief",
    principal="alice",
))

alice_jobs = store.list(principal="alice")   # → just Alice's job
everything = store.list()                     # → every job, all owners

Scope a suggestion to a user

Set principal on a Suggestion. Each owner sees only their own pending queue.

from praisonaiagents.scheduler.suggestion_store import SuggestionStore, Suggestion

store = SuggestionStore()

store.add(Suggestion(
    id="sug_abc",
    blueprint_name="morning-brief",
    slots={"hour": 8},
    principal="alice",
))

alice_pending = store.list_pending(principal="alice")   # → Alice's suggestion
bob_pending = store.list_pending(principal="bob")        # → empty

Backward compatible (single-user)

Omit principal everywhere. Behaviour is unchanged — the store is one shared pool.

from praisonaiagents.scheduler import ScheduleJob, Schedule
from praisonaiagents.scheduler.config_store import ConfigYamlScheduleStore

store = ConfigYamlScheduleStore()

store.add(ScheduleJob(
    name="daily-report",
    schedule=Schedule(kind="cron", cron_expr="0 9 * * *"),
    message="Send the daily report",
))

store.list()   # → every job in one shared pool

How It Works

The gateway resolves the caller to a stable identity, then threads that string into every store call as principal. The store filters reads and refuses cross-owner mutations.

Call	Scoped by `principal`	Cross-owner result
`list` / `list_pending`	Yes	Other owners’ items are hidden
`get_by_name`	Yes	Returns `None` (not found)
`remove_by_name`	Yes	Returns `False` (not removed)
`accept` / `dismiss`	Yes	Returns `False` (refused)
`get(job_id)` / `remove(job_id)`	No	Direct-id access is capability-shaped

Configuration Options

Every method below gained an optional principal filter. principal=None is the default and preserves global / single-tenant behaviour.

`Suggestion` field

Field	Type	Default	Description
`principal`	`Optional[str]`	`None`	Canonical identity of the owner. `None` = global / single-tenant.

`SuggestionStore` methods

Method	Signature	Behaviour
`add`	`add(self, suggestion: Suggestion) -> bool`	Pending cap and dedup window are evaluated within the suggestion’s own `principal`.
`list_pending`	`list_pending(self, principal: Optional[str] = None) -> List[Suggestion]`	Returns only that owner’s active suggestions; `None` = global.
`accept`	`accept(self, suggestion_id: str, principal: Optional[str] = None) -> bool`	Returns `False` if the suggestion is owned by a different principal.
`dismiss`	`dismiss(self, suggestion_id: str, principal: Optional[str] = None) -> bool`	Returns `False` if the suggestion is owned by a different principal.

`ScheduleJob` field

Field	Type	Default	Description
`principal`	`Optional[str]`	`None`	Canonical identity of the end-user who owns this job. Distinct from `agent_id` (the owning agent) and `origin` (where it was created). Omitted from `to_dict()` when `None`.

`FileScheduleStore` / `ConfigYamlScheduleStore` methods

Method	Signature	Behaviour
`list`	`list(self, agent_id: Optional[str] = None, principal: Optional[str] = None) -> List[ScheduleJob]`	Filters to jobs whose `principal` matches; `None` = global.
`get_by_name`	`get_by_name(self, name: str, principal: Optional[str] = None) -> Optional[ScheduleJob]`	A job owned by a different identity is treated as not found (`None`).
`remove_by_name`	`remove_by_name(self, name: str, principal: Optional[str] = None) -> bool`	A job owned by a different identity is skipped (returns `False`).

Semantics Reference

Situation	Result
`principal=None` on any reader / mutator	Global / single-tenant behaviour (unchanged).
`list_pending(principal="alice")` / `list(principal="alice")`	Only items whose `principal == "alice"` are returned.
`accept(id, principal="bob")` where suggestion is owned by `"alice"`	Returns `False`; suggestion is NOT accepted.
`dismiss(id, principal="bob")` where suggestion is owned by `"alice"`	Returns `False`; suggestion is NOT dismissed.
`get_by_name("shared-name", principal="bob")` where job is owned by `"alice"`	Returns `None` (treated as not found).
`remove_by_name("shared-name", principal="bob")` where job is owned by `"alice"`	Returns `False`; job is NOT removed.
`add(Suggestion(principal="alice", ...))` while Alice is at `MAX_PENDING_CAP`	Returns `False`.
`add(Suggestion(principal="bob", ...))` while Alice is at `MAX_PENDING_CAP`	Returns `True` — Bob’s cap is measured independently.
Two identical `blueprint_name`+`slots` from the same principal within 24h	Deduped (2nd rejected).
Two identical `blueprint_name`+`slots` from different principals within 24h	Both accepted — dedup window is per-principal.
Serialising a job with `principal=None`	`principal` key is absent from `to_dict()` output.

Choosing When to Scope

Common Patterns

Thread the gateway-resolved identity down; the core carries the string and enforces the filter but never resolves identity itself.

from praisonaiagents.scheduler import ScheduleJob, Schedule
from praisonaiagents.scheduler.config_store import ConfigYamlScheduleStore

store = ConfigYamlScheduleStore()

def create_for_caller(principal: str, name: str, cron: str, message: str) -> None:
    store.add(ScheduleJob(
        name=name,
        schedule=Schedule(kind="cron", cron_expr=cron),
        message=message,
        principal=principal,
    ))

def list_for_caller(principal: str):
    return store.list(principal=principal)

get(job_id) and remove(job_id) are intentionally NOT scoped. Direct-id access is capability-shaped — the id is an opaque 12-character hex (uuid4().hex[:12]). The guessable surface is the human-readable name, so scoping the name-based path (get_by_name / remove_by_name) is what closes the leak.

Any store implementing ScheduleStoreProtocol MUST accept principal: Optional[str] = None on list, get_by_name, and remove_by_name for isinstance(store, ScheduleStoreProtocol) (runtime-checkable) to keep passing. It MUST also honour the “cross-owner is invisible” contract: hide other owners’ items on filtered reads, and refuse cross-owner mutations by returning None / False.

Backward Compatibility

Existing on-disk jobs.json, config.yaml, and suggestions.json files load unchanged. principal is omitted from to_dict() output when None, and every store method defaults principal to None — the pre-scoping global behaviour.

from praisonaiagents.scheduler import ScheduleJob, Schedule

job = ScheduleJob(
    name="daily-report",
    schedule=Schedule(kind="cron", cron_expr="0 9 * * *"),
    message="Send the daily report",
)

assert "principal" not in job.to_dict()   # key absent when None

Best Practices

Reuse the gateway-resolved identity

Thread the identity the gateway already resolved (IdentityResolverProtocol output) down as principal. Do not invent a new identity in the store layer — the core carries the string and enforces the filter, it does not resolve identity.

Keep the store shared, not sharded

The store is designed for one file with per-record ownership, not one file per user. Pass principal on each record instead of creating a separate store per user.

Match the mutation to the read

If you list per-principal in the UI, always mutate per-principal in the callback. Never accept a naked sug_id from an untrusted UI without re-scoping to the caller’s principal.

Treat the direct id as a capability

get(id) and remove(id) are intentionally unscoped. Surface a job’s opaque id only to its owner — the id itself is the access token for direct-id operations.

Automation Suggestions

Per-user consent-first automation proposals.

Async Agent Scheduler

The scheduler that owns these jobs.

Scheduled Run Policy

Safety gate for unattended runs — a different layer.

Bot Command Access Control

Who may call /automations at all.

Scheduled Run Policy

Package Tier Architecture

Getting Started

Install

Core Concepts

Guides

Features

Models

Databases

Observability

Memory

Telemetry & Analytics

Knowledge

RAG

Persistence

Tools

Other Features

Developers

Configuration

Best Practices

Getting Started (No Code)

Per-User Scheduler Isolation

Quick Start

How It Works

Configuration Options

`Suggestion` field

`SuggestionStore` methods

`ScheduleJob` field

`FileScheduleStore` / `ConfigYamlScheduleStore` methods

Semantics Reference

Choosing When to Scope

Common Patterns

Backward Compatibility

Best Practices

Automation Suggestions

Async Agent Scheduler

Scheduled Run Policy

Bot Command Access Control

​Quick Start

​How It Works

​Configuration Options

​Suggestion field

​SuggestionStore methods

​ScheduleJob field

​FileScheduleStore / ConfigYamlScheduleStore methods

​Semantics Reference

​Choosing When to Scope

​Common Patterns

​Backward Compatibility

​Best Practices

​Related

Automation Suggestions

Async Agent Scheduler

Scheduled Run Policy

Bot Command Access Control

Quick Start

How It Works

Configuration Options

`Suggestion` field

`SuggestionStore` methods

`ScheduleJob` field

`FileScheduleStore` / `ConfigYamlScheduleStore` methods

Semantics Reference

Choosing When to Scope

Common Patterns

Backward Compatibility

Best Practices

Related