Skip to content

cost

Cost Intelligence — async Cortex usage aggregation per product/session.

CostIntelligence

Aggregates Cortex AI costs by (chapter_id, session_type) from INFORMATION_SCHEMA.

Populated by a serverless task refreshed every 60 minutes. It reads INFORMATION_SCHEMA.CORTEX_FUNCTIONS_QUERY_USAGE filtered by QUERY_TAG and computes per-chapter, per-session-type cost averages.

QUERY_TAG contract — every SP wrapper that calls a Cortex function must set a structured JSON tag before the call:

ALTER SESSION SET QUERY_TAG = '{
  "product":      "PLUTO_SCHOOL",
  "client":       ":client_id",
  "chapter":      ":chapter_id",
  "session_type": ":session_type",
  "session_id":   ":session_id"
}';

Without this tag the data is unusable for cost attribution.

Table layout in DB_{APP}.PUBLIC (no PII — safe to keep after client offboarding):

  • SESSION_COST_AVG — avg and p75 credits per (chapter_id, session_type).
  • SESSION_COST_DEFAULTS — fallback when sample_size < 10.
  • CHAPTER_CONFIG — deterministic config (nb questions, session_type, …).

Individual session data (profile_id, session_id) stays in {client}.SESSION_HISTORY and is dropped with the client schema at unsubscription.

Parameters:

Name Type Description Default
session Any

Active Snowpark session.

 required
schema str

Fully-qualified public schema, e.g. "DB_PLUTO_SCHOOL.PUBLIC".

 required
Example 
ci = CostIntelligence(session, schema="DB_PLUTO_SCHOOL.PUBLIC")
cost = ci.estimate("chapter_001", session_type="tutoring")
# → {"avg_credits": 0.012, "p75_credits": 0.018, "sample_size": 42}
Source code in src/pinky_ai/cost.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
class CostIntelligence:
    """Aggregates Cortex AI costs by (chapter_id, session_type) from INFORMATION_SCHEMA.

    Populated by a serverless task refreshed every 60 minutes. It reads
    `INFORMATION_SCHEMA.CORTEX_FUNCTIONS_QUERY_USAGE` filtered by `QUERY_TAG`
    and computes per-chapter, per-session-type cost averages.

    **QUERY_TAG contract** — every SP wrapper that calls a Cortex function must set
    a structured JSON tag before the call:

    ```sql
    ALTER SESSION SET QUERY_TAG = '{
      "product":      "PLUTO_SCHOOL",
      "client":       ":client_id",
      "chapter":      ":chapter_id",
      "session_type": ":session_type",
      "session_id":   ":session_id"
    }';
    ```

    Without this tag the data is unusable for cost attribution.

    **Table layout in `DB_{APP}.PUBLIC`** (no PII — safe to keep after client offboarding):

    - `SESSION_COST_AVG` — avg and p75 credits per `(chapter_id, session_type)`.
    - `SESSION_COST_DEFAULTS` — fallback when `sample_size < 10`.
    - `CHAPTER_CONFIG` — deterministic config (nb questions, session_type, …).

    Individual session data (`profile_id`, `session_id`) stays in
    `{client}.SESSION_HISTORY` and is dropped with the client schema at
    unsubscription.

    Args:
        session: Active Snowpark session.
        schema: Fully-qualified public schema, e.g. `"DB_PLUTO_SCHOOL.PUBLIC"`.

    Example:
        ```python
        ci = CostIntelligence(session, schema="DB_PLUTO_SCHOOL.PUBLIC")
        cost = ci.estimate("chapter_001", session_type="tutoring")
        # → {"avg_credits": 0.012, "p75_credits": 0.018, "sample_size": 42}
        ```
    """

    def __init__(self, session: Any, schema: str) -> None: ...

    def estimate(
        self,
        chapter_id: str,
        session_type: str,
        conservative: bool = True,
    ) -> dict[str, float | int]:
        """Return estimated credits for one session of a given chapter.

        Falls back to `SESSION_COST_DEFAULTS` when `sample_size < 10`.
        The UI should surface a confidence note when falling back to defaults.

        Args:
            chapter_id: Chapter identifier matching `CHAPTER_CONFIG`.
            session_type: Session type: `"tutoring"`, `"CAN"`, `"scoring"`, etc.
            conservative: If `True` (default), return `p75_credits` instead of
                `avg_credits`. The p75 estimate under-promises cost to the user.

        Returns:
            Dict with keys: `avg_credits`, `p75_credits`, `sample_size`.
        """
        ...

    def bulk_load(self, profile_id: str) -> list[dict[str, Any]]:
        """Load all chapters with cost estimates in a single query (WH call 1).

        Joins `USER_PROFILES`, `CHAPTER_CONFIG`, `SESSION_COST_AVG`, and
        `SESSION_COST_DEFAULTS` in one SELECT. Intended to populate
        `session_state` before the first user interaction, avoiding repeated
        round-trips during the session.

        Args:
            profile_id: User profile identifier from the client schema.

        Returns:
            List of dicts with keys: `profile_id`, `token_quota`, `chapter_id`,
            `chapter_name`, `session_type`, `questions_count`,
            `estimated_cost`, `estimated_cost_p75`, `sample_size`.
        """
        ...

bulk_load(profile_id)

Load all chapters with cost estimates in a single query (WH call 1).

Joins USER_PROFILES, CHAPTER_CONFIG, SESSION_COST_AVG, and SESSION_COST_DEFAULTS in one SELECT. Intended to populate session_state before the first user interaction, avoiding repeated round-trips during the session.

Parameters:

Name Type Description Default
profile_id str

User profile identifier from the client schema.

 required

Returns:

Type Description
list[dict[str, Any]]

List of dicts with keys: profile_id, token_quota, chapter_id,
list[dict[str, Any]]

chapter_name, session_type, questions_count,
list[dict[str, Any]]

estimated_cost, estimated_cost_p75, sample_size.
Source code in src/pinky_ai/cost.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def bulk_load(self, profile_id: str) -> list[dict[str, Any]]:
    """Load all chapters with cost estimates in a single query (WH call 1).

    Joins `USER_PROFILES`, `CHAPTER_CONFIG`, `SESSION_COST_AVG`, and
    `SESSION_COST_DEFAULTS` in one SELECT. Intended to populate
    `session_state` before the first user interaction, avoiding repeated
    round-trips during the session.

    Args:
        profile_id: User profile identifier from the client schema.

    Returns:
        List of dicts with keys: `profile_id`, `token_quota`, `chapter_id`,
        `chapter_name`, `session_type`, `questions_count`,
        `estimated_cost`, `estimated_cost_p75`, `sample_size`.
    """
    ...

estimate(chapter_id, session_type, conservative=True)

Return estimated credits for one session of a given chapter.

Falls back to SESSION_COST_DEFAULTS when sample_size < 10. The UI should surface a confidence note when falling back to defaults.

Parameters:

Name Type Description Default
chapter_id str

Chapter identifier matching CHAPTER_CONFIG.

 required
session_type str

Session type: "tutoring", "CAN", "scoring", etc.

 required
conservative bool

If True (default), return p75_credits instead of avg_credits. The p75 estimate under-promises cost to the user.

 True

Returns:

Type Description
dict[str, float | int]

Dict with keys: avg_credits, p75_credits, sample_size.
Source code in src/pinky_ai/cost.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def estimate(
    self,
    chapter_id: str,
    session_type: str,
    conservative: bool = True,
) -> dict[str, float | int]:
    """Return estimated credits for one session of a given chapter.

    Falls back to `SESSION_COST_DEFAULTS` when `sample_size < 10`.
    The UI should surface a confidence note when falling back to defaults.

    Args:
        chapter_id: Chapter identifier matching `CHAPTER_CONFIG`.
        session_type: Session type: `"tutoring"`, `"CAN"`, `"scoring"`, etc.
        conservative: If `True` (default), return `p75_credits` instead of
            `avg_credits`. The p75 estimate under-promises cost to the user.

    Returns:
        Dict with keys: `avg_credits`, `p75_credits`, `sample_size`.
    """
    ...