Automation Suggestions & Blueprints

from praisonai.scheduler.suggestion_engine import SuggestionEngine

engine = SuggestionEngine()

# Feed one observation per completed turn.
# 1st and 2nd calls: returns None (below threshold=3).
engine.observe("morning-brief", slots={"hour": 8}, principal="telegram:alice")
engine.observe("morning-brief", slots={"hour": 8}, principal="telegram:alice")

# 3rd call: threshold met — engine.propose() runs, store returns a suggestion id.
sug_id = engine.observe(
    "morning-brief",
    slots={"hour": 8},
    principal="telegram:alice",
    reason="Alice asks for a morning brief around 8am every day.",
)

# Different slots for the same user = distinct intent. This is turn 1 of its own count.
engine.observe("morning-brief", slots={"hour": 9}, principal="telegram:alice")

# Different principal = fully separate count. Bob's first observation is turn 1.
engine.observe("morning-brief", slots={"hour": 8}, principal="telegram:bob")

# Fire immediately on the first observation — handy for testing.
engine.observe("morning-brief", slots={"hour": 8}, threshold=1)

# Wait for stronger evidence over a longer window (30 days).
engine.observe(
    "morning-brief",
    slots={"hour": 8},
    threshold=5,
    window_seconds=3600 * 24 * 30,
)

# Nested slots are supported — the recurrence key is frozen recursively.
engine.observe(
    "morning-brief",
    slots={"weekdays": ["mon", "tue"], "opts": {"tz": "UTC"}},
)

name: standup-digest
version: "1.0.0"
description: Team standup digest
category: daily
default_deliver: telegram
prompt_template: "Summarise today's standup for {team}."
schedule_template: "cron:{minute} {hour} * * {weekdays_expression}"
slots:
  - name: hour
    type: int
    default: 9
  - name: minute
    type: int
    default: 30
  - name: weekdays
    type: choice
    default: mon-fri
    choices: [mon-fri, daily, weekends]
  - name: team
    type: str
    default: engineering

from praisonai.scheduler.suggestion_engine import SuggestionEngine
from praisonai.scheduler.blueprint_catalogue import BlueprintCatalogue

engine = SuggestionEngine()

sug_id = engine.propose(
    "morning-brief",
    slots={"hour": 8, "weekdays": "mon-fri"},
    deliver="telegram",
    reason="Detected daily morning request pattern",
)

if sug_id:
    catalogue = BlueprintCatalogue()
    bp = catalogue.get_blueprint("morning-brief")
    sug = engine.get_suggestion(sug_id)
    resolved = catalogue.resolve_slots(bp, sug.slots)
    prompt = catalogue.materialize_prompt(bp, resolved)
    schedule = catalogue.materialize_schedule(bp, resolved)
    engine.accept(sug_id)
    print(prompt, schedule)

User → /automations
Bot  → "💡 You have 1 pending automation suggestion:"
Bot  → "💡 Detected daily morning request pattern
        morning-brief · hour=8, weekdays=mon-fri"
        [✓ Accept]  [✕ Dismiss]
User → taps ✓ Accept
Bot  → "✅ Automation scheduled. Added job 'morning-brief' (schedule cron:0 8 * * mon,tue,wed,thu,fri)"

User → /blueprint
Bot  → "📋 Create an automation from a template:
        • morning-brief — Daily morning briefing with news and priorities
        • important-mail — Check for important emails at a regular interval
        • weekly-review — End-of-week summary and review
        Usage: /blueprint <name> [slot=value ...]
        Example: /blueprint morning-brief hour=8 weekdays=mon-fri"
User → /blueprint morning-brief hour=8 weekdays=mon-fri focus=tech
Bot  → "✅ Automation scheduled from 'morning-brief'. Added job 'morning-brief' (schedule cron:0 8 * * mon,tue,wed,thu,fri)"

User → "Give me the morning brief"           (turn 1 → observe → below threshold)
User → next day: "morning brief please"      (turn 2 → observe → below threshold)
User → next day: "brief me on the morning"   (turn 3 → observe → threshold met → propose)
Bot  → "💡 Noticed a recurring 'morning-brief' request (3x); offering to automate it."
       morning-brief · hour=8
       [✓ Accept]  [✕ Dismiss]
User → taps ✓ Accept
Bot  → "✅ Automation scheduled. Added job 'morning-brief' ..."

Alice → DMs the bot: "remind me to file expenses every Friday at 5pm"
Bot   → add(Suggestion(principal="tg:alice", name="weekly-expenses", ...))
Bob   → DMs the same bot: "remind me to file expenses every Friday at 5pm"   (identical text)
Bot   → add(Suggestion(principal="tg:bob", ...))  # admitted — dedup is per-principal
Alice → /automations   → sees only her own pending suggestion
Alice → accepts        → job created with principal="tg:alice"
Bob   → accepts        → his own job created; no name collision, both coexist

from praisonaiagents.scheduler import Suggestion, SuggestionStore

store = SuggestionStore()  # ~/.praisonai/suggestions.json

# In a gateway handler that already knows who's calling:
def handle_user_message(user_id: str) -> None:
    store.add(Suggestion(
        id=f"sug_{user_id}_standup",
        blueprint_name="morning-brief",
        slots={"hour": 9, "weekdays": "mon-fri"},
        reason="Detected daily standup request",
        principal=user_id,          # isolate to this caller
    ))                              # pending cap + dedup window scoped per user_id

# Later, when the same user pulls their pending list:
def list_for(user_id: str) -> list[Suggestion]:
    return store.list_pending(principal=user_id)