approval_store

approval_store

ApprovalStore — SQLite-backed store for proactive agent action approvals.

Two tables: - pending_actions: actions proposed by the proactive agent awaiting user decision - permission_memory: remembered user decisions keyed by action pattern

Permission key format: "{action_type}:{fingerprint}" e.g. "email_delete:domain:noreply.github.com" "sms_draft_reply:contact:+15551234567"

Classes

PendingAction dataclass

PendingAction(
    id: str,
    action_type: str,
    description: str,
    payload: Dict[str, Any],
    permission_key: str,
    tier: str,
    status: str = STATUS_PENDING,
    created_at: str = "",
    expires_at: str = "",
    notification_sent: bool = False,
    decision_at: Optional[str] = None,
)

An action proposed by the proactive agent.

PermissionRule dataclass

PermissionRule(
    permission_key: str,
    decision: str,
    times_approved: int = 0,
    times_denied: int = 0,
    last_updated: str = "",
    notes: str = "",
)

A remembered user decision for a permission pattern.

ApprovalStore

ApprovalStore(db_path: str = '')

SQLite store for proactive agent action approvals and permission memory.

Source code in src/diapason/tools/approval_store.py

def __init__(self, db_path: str = "") -> None:
    if not db_path:
        db_path = str(get_config_dir() / "approvals.db")
    self._db_path = db_path
    Path(db_path).parent.mkdir(parents=True, exist_ok=True)
    self._conn = sqlite3.connect(db_path, check_same_thread=False)
    self._conn.execute("PRAGMA journal_mode=WAL")
    self._create_tables()
    self._conn.commit()

Methods:

queue_action

queue_action(
    action_type: str,
    description: str,
    payload: Dict[str, Any],
    permission_key: str,
    tier: str,
    ttl_hours: int = 24,
) -> PendingAction

Create and persist a new pending action.

Source code in src/diapason/tools/approval_store.py

def queue_action(
    self,
    action_type: str,
    description: str,
    payload: Dict[str, Any],
    permission_key: str,
    tier: str,
    ttl_hours: int = 24,
) -> PendingAction:
    """Create and persist a new pending action."""
    now = datetime.now(timezone.utc)
    action = PendingAction(
        id=uuid.uuid4().hex[:12],
        action_type=action_type,
        description=description,
        payload=payload,
        permission_key=permission_key,
        tier=tier,
        status=STATUS_PENDING,
        created_at=now.isoformat(),
        expires_at=(now + timedelta(hours=ttl_hours)).isoformat(),
    )
    self._conn.execute(
        """
        INSERT OR REPLACE INTO pending_actions
            (id, action_type, description, payload, permission_key,
             tier, status, created_at, expires_at, notification_sent, decision_at)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """,
        (
            action.id,
            action.action_type,
            action.description,
            json.dumps(action.payload),
            action.permission_key,
            action.tier,
            action.status,
            action.created_at,
            action.expires_at,
            int(action.notification_sent),
            action.decision_at,
        ),
    )
    self._conn.commit()
    return action

list_pending

list_pending() -> List[PendingAction]

Return all non-expired pending actions.

Source code in src/diapason/tools/approval_store.py

def list_pending(self) -> List[PendingAction]:
    """Return all non-expired pending actions."""
    now = datetime.now(timezone.utc).isoformat()
    rows = self._conn.execute(
        "SELECT id, action_type, description, payload, permission_key, "
        "tier, status, created_at, expires_at, notification_sent, decision_at "
        "FROM pending_actions WHERE status = ? AND expires_at > ? "
        "ORDER BY created_at",
        (STATUS_PENDING, now),
    ).fetchall()
    return [PendingAction.from_row(r) for r in rows]

list_approved

list_approved() -> List[PendingAction]

Return approved-but-not-yet-executed actions.

Source code in src/diapason/tools/approval_store.py

def list_approved(self) -> List[PendingAction]:
    """Return approved-but-not-yet-executed actions."""
    rows = self._conn.execute(
        "SELECT id, action_type, description, payload, permission_key, "
        "tier, status, created_at, expires_at, notification_sent, decision_at "
        "FROM pending_actions WHERE status = ? ORDER BY created_at",
        (STATUS_APPROVED,),
    ).fetchall()
    return [PendingAction.from_row(r) for r in rows]

expire_stale

expire_stale() -> int

Mark past-TTL pending actions as expired. Returns count.

Source code in src/diapason/tools/approval_store.py

def expire_stale(self) -> int:
    """Mark past-TTL pending actions as expired. Returns count."""
    now = datetime.now(timezone.utc).isoformat()
    cur = self._conn.execute(
        "UPDATE pending_actions SET status = ? "
        "WHERE status = ? AND expires_at <= ?",
        (STATUS_EXPIRED, STATUS_PENDING, now),
    )
    self._conn.commit()
    return cur.rowcount

set_permission

set_permission(
    permission_key: str,
    decision: str,
    *,
    approved: bool = False,
    notes: str = "",
) -> None

Upsert a permission rule, incrementing the relevant counter.

Source code in src/diapason/tools/approval_store.py

def set_permission(
    self,
    permission_key: str,
    decision: str,
    *,
    approved: bool = False,
    notes: str = "",
) -> None:
    """Upsert a permission rule, incrementing the relevant counter."""
    now = datetime.now(timezone.utc).isoformat()
    existing = self.get_permission(permission_key)
    if existing:
        times_approved = existing.times_approved + (1 if approved else 0)
        times_denied = existing.times_denied + (0 if approved else 1)
        self._conn.execute(
            "UPDATE permission_memory SET decision = ?, times_approved = ?, "
            "times_denied = ?, last_updated = ?, notes = ? "
            "WHERE permission_key = ?",
            (
                decision,
                times_approved,
                times_denied,
                now,
                notes or existing.notes,
                permission_key,
            ),
        )
    else:
        self._conn.execute(
            "INSERT INTO permission_memory "
            "(permission_key, decision, times_approved, times_denied, "
            "last_updated, notes) "
            "VALUES (?, ?, ?, ?, ?, ?)",
            (
                permission_key,
                decision,
                1 if approved else 0,
                0 if approved else 1,
                now,
                notes,
            ),
        )
    self._conn.commit()

get_seen_ids

get_seen_ids() -> set

Return all doc_ids and message_ids previously queued (any status).

Used by ProactiveAgent to skip items it has already proposed so they don't resurface on every run while still unread/unanswered.

Source code in src/diapason/tools/approval_store.py

def get_seen_ids(self) -> set:
    """Return all doc_ids and message_ids previously queued (any status).

    Used by ``ProactiveAgent`` to skip items it has already proposed so
    they don't resurface on every run while still unread/unanswered.
    """
    seen: set = set()
    rows = self._conn.execute("SELECT payload FROM pending_actions").fetchall()
    for (payload_json,) in rows:
        try:
            payload = json.loads(payload_json) if payload_json else {}
        except json.JSONDecodeError:
            continue
        doc_id = payload.get("doc_id", "")
        if doc_id:
            seen.add(doc_id)
        msg_id = payload.get("message_id", "")
        if msg_id:
            seen.add(f"gmail:{msg_id}")
    return seen

Functions: