
Sf Ai Agentforce Observability
- 1.2k installs
- 423 repo stars
- Updated April 27, 2026
- jaganpro/sf-skills
sf-ai-agentforce-observability is an agent skill for agentforce session tracing extraction and analysis. trigger when: user extracts stdm data from data cloud, analyzes agent session traces, debugs agent conversations vi
About
The sf-ai-agentforce-observability skill is designed for agentforce session tracing extraction and analysis. TRIGGER when: user extracts STDM data from Data Cloud, analyzes agent session traces, debugs agent conversations via. Verify setup and auth Confirm Data 360 tracing exists and JWT/ECA auth is working. Extract to Parquet Use the provided scripts under scripts/ rather than reimplementing extraction logic. Invoke when the user asks about sf ai agentforce observability or related SKILL.md workflows.
- Data 360 / Session Tracing extraction.
- .parquet files from Agentforce telemetry.
- session timeline reconstruction.
- trace-driven debugging of topic routing, action failures, or latency.
- Polars / PyArrow-based analysis of large telemetry datasets.
Sf Ai Agentforce Observability by the numbers
- 1,190 all-time installs (skills.sh)
- +4 installs in the week ending Aug 5, 2026 (Skillselion tracking)
- Ranked #169 of 1,435 DevOps & CI/CD skills by installs in the Skillselion catalog
- Security screen: HIGH risk (skills.sh audit)
- Data as of Aug 5, 2026 (Skillselion catalog sync)
sf-ai-agentforce-observability capabilities & compatibility
- Capabilities
- data 360 / session tracing extraction · .parquet files from agentforce telemetry · session timeline reconstruction · trace driven debugging of topic routing, action
What sf-ai-agentforce-observability says it does
Agentforce session tracing extraction and analysis. TRIGGER when: user extracts STDM data from Data Cloud, analyzes agent session traces, debugs agent conversations via telemetry,
Agentforce session tracing extraction and analysis. TRIGGER when: user extracts STDM data from Data Cloud, analyzes agent session traces, debugs agent conversat
npx skills add https://github.com/jaganpro/sf-skills --skill sf-ai-agentforce-observabilityAdd your badge
Show developers this skill is listed on Skillselion. Paste this into your README.
| Installs | 1.2k |
|---|---|
| repo stars | ★ 423 |
| Security audit | 3 / 3 scanners passed |
| Last updated | April 27, 2026 |
| Repository | jaganpro/sf-skills ↗ |
How do I agentforce session tracing extraction and analysis. trigger when: user extracts stdm data from data cloud, analyzes agent session traces, debugs agent conversations via?
Agentforce session tracing extraction and analysis. TRIGGER when: user extracts STDM data from Data Cloud, analyzes agent session traces, debugs agent conversations via.
Who is it for?
Developers using sf ai agentforce observability workflows documented in SKILL.md.
Skip if: Skip when the task falls outside sf-ai-agentforce-observability scope or needs a different stack.
When should I use this skill?
User asks about sf ai agentforce observability or related SKILL.md workflows.
What you get
Completed sf-ai-agentforce-observability workflow with documented commands, files, and expected deliverables.
- Chronological message timeline
- Step-by-step agent trace output
By the numbers
- Includes message-timeline.py CLI for per-session reconstruction
- Reads parquet exports from stdm_data and trace-results directories
Files
sf-ai-agentforce-observability: Agentforce Session Tracing Extraction & Analysis
Use this skill when the user needs trace-based observability, not just testing: extract Session Tracing Data Model (STDM) records, work with Parquet datasets, reconstruct session timelines, analyze topic/action latency, or debug agent behavior from Data 360 telemetry.
When This Skill Owns the Task
Use sf-ai-agentforce-observability when the work involves:
- Data 360 / Session Tracing extraction
.parquetfiles from Agentforce telemetry- session timeline reconstruction
- trace-driven debugging of topic routing, action failures, or latency
- Polars / PyArrow-based analysis of large telemetry datasets
Delegate elsewhere when the user is:
- formally testing agents → sf-ai-agentforce-testing
- debugging Apex logs → sf-debug
- authoring or reconfiguring the agent itself → sf-ai-agentforce or sf-ai-agentscript
---
Prerequisites That Must Exist
Before extraction, verify:
- Data 360 is enabled
- Session Tracing is enabled
- the Salesforce Standard Data Model version is sufficient
- Einstein / Agentforce capabilities are enabled in the org
- JWT / ECA auth for Data 360 access is configured
If auth is missing, hand off to:
- sf-connected-apps
Deep setup guide:
- references/auth-setup.md
---
What This Skill Works With
Core storage / analysis model
- extraction via Data 360 APIs
- Parquet for storage efficiency
- Polars for large-scale lazy analysis
Core STDM entities
At minimum, expect work around:
- session
- interaction / turn
- interaction step
- moment
- message
GenAI Trust Layer / audit records may also be relevant for content-quality and generation debugging.
Full schema:
- references/data-model-reference.md
---
Required Context to Gather First
Ask for or infer:
- target org alias
- time window or date range
- agent filter, if any
- whether the goal is extraction, summary analysis, or single-session debugging
- output location for extracted data
- whether the user already has Parquet files on disk
---
Recommended Workflow
1. Verify setup and auth
Confirm Data 360 tracing exists and JWT/ECA auth is working.
2. Choose the extraction mode
| Need | Default approach |
|---|---|
| recent telemetry snapshot | extract last N days |
| focused investigation | filtered extraction by date and agent |
| one broken conversation | extract or debug a single session tree |
| ongoing usage analytics | incremental extraction |
3. Extract to Parquet
Use the provided scripts under scripts/ rather than reimplementing extraction logic.
4. Analyze with Polars
Common analysis goals:
- session volume and duration
- topic distribution
- action step failures
- latency hotspots
- abandonment / escalation patterns
- session-level timeline reconstruction
5. Convert findings into next actions
Typical outcomes:
- topic mismatch → improve routing or descriptions
- action failure → inspect Flow / Apex implementation
- latency issue → optimize downstream action path
- test gap → add targeted agent tests
---
High-Signal Operational Rules
- treat STDM as read-only telemetry
- expect ingestion lag; this is not perfect real-time debugging
- use date filters and focused extraction to avoid unnecessary volume / query cost
- prefer Parquet over ad hoc JSON for durable analysis
- use lazy Polars patterns for large datasets
Common pitfalls:
- assuming missing data means no issue, when tracing may simply not be enabled
- running huge broad queries without date or agent filters
- trying to fix the agent inside this skill instead of handing off to authoring / testing skills
---
Output Format
When finishing, report in this order: 1. What data was extracted or analyzed 2. Scope (org, dates, agent filter, session IDs) 3. Key findings 4. Likely root causes 5. Recommended next skill / next action
Suggested shape:
Observability task: <extract / analyze / debug-session>
Scope: <org, dates, agents, session ids>
Artifacts: <directories / parquet files>
Findings: <latency, routing, action, quality, abandonment patterns>
Root cause: <best current explanation>
Next step: <testing, agent fix, flow fix, apex fix>---
Cross-Skill Integration
| Need | Delegate to | Reason |
|---|---|---|
| auth / JWT setup | sf-connected-apps | Data 360 access |
| fix agent routing / behavior | sf-ai-agentscript | authoring corrections |
| formal regression / coverage tests | sf-ai-agentforce-testing | reproducible test loops |
| Flow-backed action debugging | sf-flow | declarative repair |
| Apex-backed action debugging | sf-debug or sf-apex | code / log investigation |
---
Reference Map
Start here
- README.md
- references/basic-extraction.md
- references/filtered-extraction.md
- references/cli-reference.md
Data model / querying
- references/data-model-reference.md
- references/query-patterns.md
- references/client-demo-queries.md
Analysis / debugging
- references/analysis-cookbook.md
- references/analysis-examples.md
- references/debugging-sessions.md
- references/polars-cheatsheet.md
- references/agent-execution-lifecycle.md
Auth / troubleshooting
- references/auth-setup.md
- references/troubleshooting.md
- references/billing-and-troubleshooting.md
- references/builder-trace-api.md
- scripts/
---
Score Guide
| Score | Meaning |
|---|---|
| 90+ | strong telemetry-backed diagnosis |
| 75–89 | useful analysis with minor gaps |
| 60–74 | partial visibility only |
| < 60 | insufficient evidence; gather more telemetry |
# Network capture data — contains session tokens, org URLs, and client-specific content
captures/
*.har
# Python
__pycache__/
*.pyc
.trace-venv/
# Dev-only test harness (contains hardcoded org references)
scripts/test_v65.py
# Output data
stdm_data/
*.parquet
trace-results/
# Builder URL cache (contains org-specific URLs)
.builder-url-cache.json
# Debug screenshots
debug-*.png
#!/usr/bin/env python3
"""
Message Timeline Analysis Template
Reconstructs conversation timelines for debugging agent behavior.
Shows all messages and steps in chronological order for a session.
Usage:
python3 message-timeline.py --data-dir ./stdm_data --session-id "a0x..."
python3 message-timeline.py --data-dir ./stdm_data --list-failed
Output includes:
- Chronological message flow
- Step details (LLM reasoning, action inputs/outputs)
- Topic transitions
- Timing information
"""
import argparse
import json
from pathlib import Path
from datetime import datetime
import polars as pl
from rich.console import Console
from rich.panel import Panel
from rich.text import Text
console = Console()
def load_data(data_dir: Path) -> dict:
"""Load all STDM entities as lazy frames."""
return {
"sessions": pl.scan_parquet(data_dir / "sessions" / "**/*.parquet"),
"interactions": pl.scan_parquet(data_dir / "interactions" / "**/*.parquet"),
"steps": pl.scan_parquet(data_dir / "steps" / "**/*.parquet"),
"messages": pl.scan_parquet(data_dir / "messages" / "**/*.parquet"),
}
def get_session_info(data: dict, session_id: str) -> dict:
"""Get session metadata."""
session = (
data["sessions"]
.filter(pl.col("ssot__Id__c") == session_id)
.collect()
)
if session.is_empty():
return None
row = session.row(0, named=True)
return {
"id": row.get("ssot__Id__c"),
"agent": row.get("ssot__AiAgentApiName__c", "Unknown"),
"start": row.get("ssot__StartTimestamp__c"),
"end": row.get("ssot__EndTimestamp__c"),
"end_type": row.get("ssot__AiAgentSessionEndType__c"),
}
def get_timeline(data: dict, session_id: str) -> list:
"""Build chronological timeline of events."""
# Get interactions for this session
interactions = (
data["interactions"]
.filter(pl.col("ssot__AiAgentSessionId__c") == session_id)
.collect()
)
interaction_ids = interactions["ssot__Id__c"].to_list()
# Get messages
messages = (
data["messages"]
.filter(pl.col("ssot__AiAgentInteractionId__c").is_in(interaction_ids))
.collect()
)
# Get steps
steps = (
data["steps"]
.filter(pl.col("ssot__AiAgentInteractionId__c").is_in(interaction_ids))
.collect()
)
# Build timeline
timeline = []
# Add messages
for row in messages.iter_rows(named=True):
timeline.append({
"type": "message",
"timestamp": row.get("ssot__MessageSentTimestamp__c", ""),
"interaction_id": row.get("ssot__AiAgentInteractionId__c"),
"message_type": row.get("ssot__AiAgentInteractionMessageType__c"),
"content": row.get("ssot__ContentText__c", ""),
})
# Add steps
for row in steps.iter_rows(named=True):
timeline.append({
"type": "step",
"timestamp": "", # Steps don't have timestamps
"interaction_id": row.get("ssot__AiAgentInteractionId__c"),
"step_type": row.get("ssot__AiAgentInteractionStepType__c"),
"name": row.get("ssot__Name__c"),
"input": row.get("ssot__InputValueText__c"),
"output": row.get("ssot__OutputValueText__c"),
})
# Add interactions (for topic info)
for row in interactions.iter_rows(named=True):
timeline.append({
"type": "interaction",
"timestamp": row.get("ssot__StartTimestamp__c", ""),
"interaction_id": row.get("ssot__Id__c"),
"interaction_type": row.get("ssot__AiAgentInteractionType__c"),
"topic": row.get("ssot__TopicApiName__c"),
})
# Sort by timestamp (messages and interactions)
timeline.sort(key=lambda x: x.get("timestamp") or "")
return timeline
def print_timeline(session_info: dict, timeline: list, verbose: bool = False):
"""Print formatted timeline."""
console.print("\n[bold cyan]🔍 SESSION TIMELINE[/bold cyan]")
console.print("═" * 70)
# Session header
console.print(f"\nSession: [cyan]{session_info['id']}[/cyan]")
console.print(f"Agent: {session_info['agent']}")
console.print(f"Started: {session_info['start']}")
console.print(f"Ended: {session_info['end']}")
console.print(f"End Type: {session_info['end_type']}")
console.print("\n" + "─" * 70)
console.print("[bold]Timeline[/bold]")
console.print("─" * 70)
current_topic = None
for event in timeline:
if event["type"] == "interaction":
if event.get("topic") != current_topic:
current_topic = event.get("topic")
console.print(f"\n[yellow]═══ Topic: {current_topic} ═══[/yellow]\n")
elif event["type"] == "message":
timestamp = event.get("timestamp", "")[:19] if event.get("timestamp") else ""
msg_type = event.get("message_type", "")
content = event.get("content", "")
if msg_type == "INPUT":
icon = "[green]→[/green]"
label = "[green][INPUT][/green]"
else:
icon = "[blue]←[/blue]"
label = "[blue][OUTPUT][/blue]"
# Truncate long content
if len(content) > 100 and not verbose:
content = content[:97] + "..."
console.print(f"{timestamp} {icon} {label}")
console.print(f" {content}\n")
elif event["type"] == "step" and verbose:
step_type = event.get("step_type", "")
name = event.get("name", "")
if step_type == "LLM_STEP":
icon = "[magenta]🧠[/magenta]"
else:
icon = "[cyan]⚡[/cyan]"
console.print(f" {icon} [{step_type}] {name}")
if event.get("input") and verbose:
try:
input_data = json.loads(event["input"])
console.print(f" Input: {json.dumps(input_data, indent=2)[:200]}")
except:
console.print(f" Input: {event['input'][:200]}")
if event.get("output") and verbose:
try:
output_data = json.loads(event["output"])
console.print(f" Output: {json.dumps(output_data, indent=2)[:200]}")
except:
console.print(f" Output: {event['output'][:200]}")
console.print()
def list_failed_sessions(data: dict, limit: int = 10) -> pl.DataFrame:
"""List sessions that failed or were escalated."""
result = (
data["sessions"]
.filter(
pl.col("ssot__AiAgentSessionEndType__c").is_in(["Escalated", "Abandoned", "Failed"])
)
.sort("ssot__StartTimestamp__c", descending=True)
.head(limit)
)
return result.collect()
def main():
parser = argparse.ArgumentParser(description="Analyze session message timeline")
parser.add_argument("--data-dir", required=True, type=Path, help="STDM data directory")
parser.add_argument("--session-id", type=str, help="Session ID to analyze")
parser.add_argument("--list-failed", action="store_true", help="List failed/escalated sessions")
parser.add_argument("--verbose", "-v", action="store_true", help="Show step details")
parser.add_argument("--output", type=Path, help="Export timeline to JSON")
args = parser.parse_args()
if not args.data_dir.exists():
console.print(f"[red]Error: Data directory not found: {args.data_dir}[/red]")
return 1
data = load_data(args.data_dir)
if args.list_failed:
console.print("\n[bold]Failed/Escalated Sessions[/bold]")
failed = list_failed_sessions(data)
for row in failed.iter_rows(named=True):
session_id = row.get("ssot__Id__c")
agent = row.get("ssot__AiAgentApiName__c", "Unknown")
end_type = row.get("ssot__AiAgentSessionEndType__c")
start = row.get("ssot__StartTimestamp__c", "")[:19]
icon = "🔄" if end_type == "Escalated" else "❌"
console.print(f" {icon} {session_id} | {agent} | {end_type} | {start}")
console.print(f"\nTo debug a session, run:")
console.print(f" python3 message-timeline.py --data-dir {args.data_dir} --session-id <ID>")
return 0
if not args.session_id:
console.print("[red]Error: --session-id required (or use --list-failed)[/red]")
return 1
session_info = get_session_info(data, args.session_id)
if not session_info:
console.print(f"[red]Error: Session not found: {args.session_id}[/red]")
return 1
timeline = get_timeline(data, args.session_id)
if args.output:
args.output.parent.mkdir(parents=True, exist_ok=True)
with open(args.output, "w") as f:
json.dump({
"session": session_info,
"timeline": timeline
}, f, indent=2)
console.print(f"[green]✓ Timeline exported to {args.output}[/green]")
else:
print_timeline(session_info, timeline, args.verbose)
return 0
if __name__ == "__main__":
exit(main())
#!/usr/bin/env python3
"""
Session Summary Analysis Template
Generates comprehensive session statistics from extracted STDM data.
Uses Polars lazy evaluation for memory-efficient processing.
Usage:
python3 session-summary.py --data-dir ./stdm_data --output ./reports
Output includes:
- Sessions by agent
- End type distribution
- Daily session trends
- Average turns per session
"""
import argparse
from pathlib import Path
from datetime import datetime
import polars as pl
from rich.console import Console
from rich.table import Table
console = Console()
def load_data(data_dir: Path) -> dict:
"""Load all STDM entities as lazy frames."""
return {
"sessions": pl.scan_parquet(data_dir / "sessions" / "**/*.parquet"),
"interactions": pl.scan_parquet(data_dir / "interactions" / "**/*.parquet"),
}
def sessions_by_agent(data: dict) -> pl.DataFrame:
"""Calculate session counts and metrics by agent."""
sessions = data["sessions"]
interactions = data["interactions"]
# Count turns per session
turns = (
interactions
.filter(pl.col("ssot__AiAgentInteractionType__c") == "TURN")
.group_by("ssot__AiAgentSessionId__c")
.agg(pl.count().alias("turn_count"))
)
# Join with sessions and aggregate
result = (
sessions
.join(
turns,
left_on="ssot__Id__c",
right_on="ssot__AiAgentSessionId__c",
how="left"
)
.group_by("ssot__AiAgentApiName__c")
.agg([
pl.count().alias("session_count"),
pl.col("turn_count").mean().alias("avg_turns"),
pl.col("turn_count").max().alias("max_turns"),
pl.col("turn_count").min().alias("min_turns"),
])
.sort("session_count", descending=True)
)
return result.collect()
def end_type_distribution(data: dict) -> pl.DataFrame:
"""Calculate session end type distribution."""
result = (
data["sessions"]
.group_by("ssot__AiAgentSessionEndType__c")
.agg(pl.count().alias("count"))
.with_columns([
(pl.col("count") / pl.col("count").sum() * 100)
.round(1)
.alias("percentage")
])
.sort("count", descending=True)
)
return result.collect()
def daily_sessions(data: dict) -> pl.DataFrame:
"""Calculate daily session counts."""
result = (
data["sessions"]
.with_columns([
pl.col("ssot__StartTimestamp__c")
.str.slice(0, 10)
.alias("date")
])
.group_by("date")
.agg(pl.count().alias("session_count"))
.sort("date")
)
return result.collect()
def print_summary(data_dir: Path):
"""Print comprehensive session summary."""
console.print("\n[bold cyan]📊 SESSION SUMMARY REPORT[/bold cyan]")
console.print(f"Data: {data_dir}")
console.print(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
console.print("═" * 60)
data = load_data(data_dir)
# Sessions by Agent
console.print("\n[bold]Sessions by Agent[/bold]")
agent_stats = sessions_by_agent(data)
table = Table()
table.add_column("Agent", style="cyan")
table.add_column("Sessions", justify="right")
table.add_column("Avg Turns", justify="right")
table.add_column("Max Turns", justify="right")
total_sessions = 0
for row in agent_stats.iter_rows(named=True):
agent = row.get("ssot__AiAgentApiName__c", "Unknown")
sessions = row.get("session_count", 0)
avg_turns = row.get("avg_turns", 0) or 0
max_turns = row.get("max_turns", 0) or 0
total_sessions += sessions
table.add_row(
str(agent),
f"{sessions:,}",
f"{avg_turns:.1f}",
str(max_turns)
)
table.add_row("─" * 20, "─" * 10, "─" * 10, "─" * 10)
table.add_row("Total", f"{total_sessions:,}", "", "", style="bold")
console.print(table)
# End Type Distribution
console.print("\n[bold]End Type Distribution[/bold]")
end_types = end_type_distribution(data)
for row in end_types.iter_rows(named=True):
end_type = row.get("ssot__AiAgentSessionEndType__c", "Unknown")
count = row.get("count", 0)
pct = row.get("percentage", 0)
icon = "✅" if end_type == "Completed" else "🔄" if end_type == "Escalated" else "❌"
console.print(f" {icon} {end_type}: {count:,} ({pct}%)")
# Daily Trend
console.print("\n[bold]Daily Session Trend[/bold]")
daily = daily_sessions(data)
if len(daily) > 0:
# Show last 7 days or all if less
recent = daily.tail(7)
for row in recent.iter_rows(named=True):
date = row.get("date", "")
count = row.get("session_count", 0)
bar = "█" * min(count // 100, 50) # Simple bar chart
console.print(f" {date}: {count:>6,} {bar}")
def export_csv(data_dir: Path, output_dir: Path):
"""Export summary data to CSV files."""
output_dir.mkdir(parents=True, exist_ok=True)
data = load_data(data_dir)
# Agent stats
agent_stats = sessions_by_agent(data)
agent_stats.write_csv(output_dir / "sessions_by_agent.csv")
# End type distribution
end_types = end_type_distribution(data)
end_types.write_csv(output_dir / "end_type_distribution.csv")
# Daily sessions
daily = daily_sessions(data)
daily.write_csv(output_dir / "daily_sessions.csv")
console.print(f"\n[green]✓ Reports exported to {output_dir}[/green]")
def main():
parser = argparse.ArgumentParser(description="Generate session summary statistics")
parser.add_argument("--data-dir", required=True, type=Path, help="STDM data directory")
parser.add_argument("--output", type=Path, help="Output directory for CSV export")
parser.add_argument("--format", choices=["table", "csv", "json"], default="table")
args = parser.parse_args()
if not args.data_dir.exists():
console.print(f"[red]Error: Data directory not found: {args.data_dir}[/red]")
return 1
if args.output:
export_csv(args.data_dir, args.output)
else:
print_summary(args.data_dir)
return 0
if __name__ == "__main__":
exit(main())
#!/usr/bin/env python3
"""
Step Distribution Analysis Template
Analyzes the distribution of LLM and ACTION steps across agents and topics.
Helps identify which actions are most frequently used and potential bottlenecks.
Usage:
python3 step-distribution.py --data-dir ./stdm_data
python3 step-distribution.py --data-dir ./stdm_data --agent Customer_Support_Agent
Output includes:
- LLM vs ACTION step ratio
- Most common actions
- Steps per turn distribution
- Action success patterns
"""
import argparse
from pathlib import Path
from datetime import datetime
import polars as pl
from rich.console import Console
from rich.table import Table
console = Console()
def load_data(data_dir: Path) -> dict:
"""Load STDM entities as lazy frames."""
return {
"sessions": pl.scan_parquet(data_dir / "sessions" / "**/*.parquet"),
"interactions": pl.scan_parquet(data_dir / "interactions" / "**/*.parquet"),
"steps": pl.scan_parquet(data_dir / "steps" / "**/*.parquet"),
}
def step_type_ratio(data: dict, agent_name: str = None) -> pl.DataFrame:
"""Calculate LLM vs ACTION step ratio."""
steps = data["steps"]
if agent_name:
# Filter by agent
sessions = data["sessions"]
interactions = data["interactions"]
session_ids = (
sessions
.filter(pl.col("ssot__AiAgentApiName__c") == agent_name)
.select("ssot__Id__c")
)
interaction_ids = (
interactions
.join(session_ids, left_on="ssot__AiAgentSessionId__c", right_on="ssot__Id__c")
.select(pl.col("ssot__Id__c").alias("interaction_id"))
)
steps = steps.join(
interaction_ids,
left_on="ssot__AiAgentInteractionId__c",
right_on="interaction_id"
)
result = (
steps
.group_by("ssot__AiAgentInteractionStepType__c")
.agg(pl.count().alias("count"))
.with_columns([
(pl.col("count") / pl.col("count").sum() * 100)
.round(1)
.alias("percentage")
])
.sort("count", descending=True)
)
return result.collect()
def action_distribution(data: dict, agent_name: str = None, top_n: int = 20) -> pl.DataFrame:
"""Get most common action names."""
steps = data["steps"]
# Filter to ACTION_STEP only
action_steps = steps.filter(
pl.col("ssot__AiAgentInteractionStepType__c") == "ACTION_STEP"
)
if agent_name:
sessions = data["sessions"]
interactions = data["interactions"]
session_ids = (
sessions
.filter(pl.col("ssot__AiAgentApiName__c") == agent_name)
.select("ssot__Id__c")
)
interaction_ids = (
interactions
.join(session_ids, left_on="ssot__AiAgentSessionId__c", right_on="ssot__Id__c")
.select(pl.col("ssot__Id__c").alias("interaction_id"))
)
action_steps = action_steps.join(
interaction_ids,
left_on="ssot__AiAgentInteractionId__c",
right_on="interaction_id"
)
result = (
action_steps
.group_by("ssot__Name__c")
.agg(pl.count().alias("count"))
.sort("count", descending=True)
.head(top_n)
)
return result.collect()
def steps_per_turn(data: dict) -> pl.DataFrame:
"""Calculate steps per turn distribution."""
result = (
data["steps"]
.group_by("ssot__AiAgentInteractionId__c")
.agg(pl.count().alias("step_count"))
.group_by("step_count")
.agg(pl.count().alias("turn_count"))
.sort("step_count")
)
return result.collect()
def print_analysis(data_dir: Path, agent_name: str = None):
"""Print step distribution analysis."""
console.print("\n[bold cyan]⚡ STEP DISTRIBUTION ANALYSIS[/bold cyan]")
console.print(f"Data: {data_dir}")
if agent_name:
console.print(f"Agent: {agent_name}")
console.print(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
console.print("═" * 60)
data = load_data(data_dir)
# Step Type Ratio
console.print("\n[bold]Step Type Ratio[/bold]")
ratio = step_type_ratio(data, agent_name)
for row in ratio.iter_rows(named=True):
step_type = row.get("ssot__AiAgentInteractionStepType__c", "Unknown")
count = row.get("count", 0)
pct = row.get("percentage", 0)
icon = "🧠" if step_type == "LLM_STEP" else "⚡"
console.print(f" {icon} {step_type}: {count:,} ({pct}%)")
# Action Distribution
console.print("\n[bold]Top Actions[/bold]")
actions = action_distribution(data, agent_name)
table = Table()
table.add_column("#", justify="right", style="dim")
table.add_column("Action Name", style="cyan")
table.add_column("Count", justify="right")
for i, row in enumerate(actions.iter_rows(named=True), 1):
action = row.get("ssot__Name__c", "Unknown")
count = row.get("count", 0)
table.add_row(str(i), str(action), f"{count:,}")
console.print(table)
# Steps per Turn
console.print("\n[bold]Steps per Turn Distribution[/bold]")
spt = steps_per_turn(data)
for row in spt.head(10).iter_rows(named=True):
step_count = row.get("step_count", 0)
turn_count = row.get("turn_count", 0)
bar = "█" * min(turn_count // 10, 40)
console.print(f" {step_count:>2} steps: {turn_count:>6,} turns {bar}")
def main():
parser = argparse.ArgumentParser(description="Analyze step distribution")
parser.add_argument("--data-dir", required=True, type=Path, help="STDM data directory")
parser.add_argument("--agent", type=str, help="Filter by agent API name")
parser.add_argument("--output", type=Path, help="Output directory for CSV export")
args = parser.parse_args()
if not args.data_dir.exists():
console.print(f"[red]Error: Data directory not found: {args.data_dir}[/red]")
return 1
if args.output:
args.output.mkdir(parents=True, exist_ok=True)
data = load_data(args.data_dir)
ratio = step_type_ratio(data, args.agent)
ratio.write_csv(args.output / "step_type_ratio.csv")
actions = action_distribution(data, args.agent)
actions.write_csv(args.output / "action_distribution.csv")
console.print(f"[green]✓ Reports exported to {args.output}[/green]")
else:
print_analysis(args.data_dir, args.agent)
return 0
if __name__ == "__main__":
exit(main())
-- Extract interactions from Data Cloud
-- DMO: ssot__AIAgentInteraction__dlm
--
-- Usage: Replace {{SESSION_IDS}} with comma-separated session IDs
--
-- Interactions represent individual turns in the conversation.
-- Each turn starts with user input and ends with agent response.
-- SESSION_END interactions mark when the session concluded.
SELECT
ssot__Id__c,
ssot__AiAgentSessionId__c,
ssot__AiAgentInteractionType__c,
ssot__TopicApiName__c,
ssot__StartTimestamp__c,
ssot__EndTimestamp__c
FROM ssot__AIAgentInteraction__dlm
WHERE ssot__AiAgentSessionId__c IN ({{SESSION_IDS}})
ORDER BY ssot__StartTimestamp__c;
-- ============================================================================
-- EXAMPLE QUERIES
-- ============================================================================
-- Interactions for specific session
-- SELECT * FROM ssot__AIAgentInteraction__dlm
-- WHERE ssot__AiAgentSessionId__c = 'a0x1234567890ABC';
-- Count turns per session
-- SELECT
-- ssot__AiAgentSessionId__c,
-- COUNT(*) as turn_count
-- FROM ssot__AIAgentInteraction__dlm
-- WHERE ssot__AiAgentInteractionType__c = 'TURN'
-- GROUP BY ssot__AiAgentSessionId__c;
-- Topic distribution
-- SELECT
-- ssot__TopicApiName__c,
-- COUNT(*) as count
-- FROM ssot__AIAgentInteraction__dlm
-- WHERE ssot__AiAgentInteractionType__c = 'TURN'
-- GROUP BY ssot__TopicApiName__c
-- ORDER BY count DESC;
-- Find sessions with topic switches (multiple topics)
-- SELECT
-- ssot__AiAgentSessionId__c,
-- COUNT(DISTINCT ssot__TopicApiName__c) as topic_count
-- FROM ssot__AIAgentInteraction__dlm
-- WHERE ssot__AiAgentInteractionType__c = 'TURN'
-- GROUP BY ssot__AiAgentSessionId__c
-- HAVING COUNT(DISTINCT ssot__TopicApiName__c) > 1;
-- Extract messages from Data Cloud
-- DMO: ssot__AIAgentMoment__dlm
--
-- Usage: Replace {{INTERACTION_IDS}} with comma-separated interaction IDs
--
-- Messages are the actual conversation content:
-- - INPUT: User messages
-- - OUTPUT: Agent responses
--
-- Used for reconstructing conversation timelines for debugging.
SELECT
ssot__Id__c,
ssot__AiAgentInteractionId__c,
ssot__ContentText__c,
ssot__AiAgentInteractionMessageType__c,
ssot__MessageSentTimestamp__c
FROM ssot__AIAgentMoment__dlm
WHERE ssot__AiAgentInteractionId__c IN ({{INTERACTION_IDS}})
ORDER BY ssot__MessageSentTimestamp__c;
-- ============================================================================
-- EXAMPLE QUERIES
-- ============================================================================
-- Messages for specific interaction
-- SELECT * FROM ssot__AIAgentMoment__dlm
-- WHERE ssot__AiAgentInteractionId__c = 'a0y1234567890ABC'
-- ORDER BY ssot__MessageSentTimestamp__c;
-- Message type distribution
-- SELECT
-- ssot__AiAgentInteractionMessageType__c,
-- COUNT(*) as count
-- FROM ssot__AIAgentMoment__dlm
-- GROUP BY ssot__AiAgentInteractionMessageType__c;
-- Find long user messages (potential complex requests)
-- SELECT
-- ssot__Id__c,
-- ssot__ContentText__c,
-- LENGTH(ssot__ContentText__c) as length
-- FROM ssot__AIAgentMoment__dlm
-- WHERE ssot__AiAgentInteractionMessageType__c = 'INPUT'
-- AND LENGTH(ssot__ContentText__c) > 500
-- ORDER BY length DESC;
-- Search messages for specific keyword
-- SELECT * FROM ssot__AIAgentMoment__dlm
-- WHERE ssot__ContentText__c LIKE '%order%'
-- AND ssot__AiAgentInteractionMessageType__c = 'INPUT';
-- Average message length by type
-- SELECT
-- ssot__AiAgentInteractionMessageType__c,
-- AVG(LENGTH(ssot__ContentText__c)) as avg_length
-- FROM ssot__AIAgentMoment__dlm
-- GROUP BY ssot__AiAgentInteractionMessageType__c;
-- Extract sessions from Data Cloud
-- DMO: ssot__AIAgentSession__dlm
--
-- Usage: Replace {{START_DATE}}, {{END_DATE}}
--
-- This query extracts session-level data including:
-- - Session ID and timestamps
-- - Channel type (how user connected)
-- - How the session ended (Completed, Abandoned, Escalated, etc.)
-- - Related messaging session (if applicable)
--
-- NOTE: Agent name is NOT on Session table. Join with Moment to get agent info.
SELECT
ssot__Id__c,
ssot__AiAgentChannelType__c,
ssot__StartTimestamp__c,
ssot__EndTimestamp__c,
ssot__AiAgentSessionEndType__c,
ssot__RelatedMessagingSessionId__c,
ssot__InternalOrganizationId__c
FROM ssot__AIAgentSession__dlm
WHERE ssot__StartTimestamp__c >= '{{START_DATE}}'
AND ssot__StartTimestamp__c < '{{END_DATE}}'
ORDER BY ssot__StartTimestamp__c;
-- ============================================================================
-- EXAMPLE QUERIES
-- ============================================================================
-- Last 7 days of sessions
-- SELECT * FROM ssot__AIAgentSession__dlm
-- WHERE ssot__StartTimestamp__c >= '2026-01-21T00:00:00.000Z'
-- ORDER BY ssot__StartTimestamp__c;
-- Sessions by agent (requires Moment join)
-- SELECT DISTINCT s.*
-- FROM ssot__AIAgentSession__dlm s
-- JOIN ssot__AiAgentMoment__dlm m
-- ON m.ssot__AiAgentSessionId__c = s.ssot__Id__c
-- WHERE m.ssot__AiAgentApiName__c = 'Customer_Support_Agent'
-- AND s.ssot__StartTimestamp__c >= '2026-01-01T00:00:00.000Z';
-- Failed/escalated sessions only
-- SELECT * FROM ssot__AIAgentSession__dlm
-- WHERE ssot__AiAgentSessionEndType__c IN ('Escalated', 'Abandoned', 'Failed')
-- AND ssot__StartTimestamp__c >= '2026-01-01T00:00:00.000Z';
-- Session count by end type
-- SELECT
-- ssot__AiAgentSessionEndType__c,
-- COUNT(*) as session_count
-- FROM ssot__AIAgentSession__dlm
-- WHERE ssot__StartTimestamp__c >= '2026-01-01T00:00:00.000Z'
-- GROUP BY ssot__AiAgentSessionEndType__c;
-- Extract interaction steps from Data Cloud
-- DMO: ssot__AIAgentInteractionStep__dlm
--
-- Usage: Replace {{INTERACTION_IDS}} with comma-separated interaction IDs
--
-- Steps are the individual processing units within a turn:
-- - LLM_STEP: Language model reasoning and generation
-- - ACTION_STEP: Flow/Apex action execution
--
-- Each step includes input/output data and variable state changes.
SELECT
ssot__Id__c,
ssot__AiAgentInteractionId__c,
ssot__AiAgentInteractionStepType__c,
ssot__Name__c,
ssot__InputValueText__c,
ssot__OutputValueText__c,
ssot__PreStepVariableText__c,
ssot__PostStepVariableText__c,
ssot__GenerationId__c
FROM ssot__AIAgentInteractionStep__dlm
WHERE ssot__AiAgentInteractionId__c IN ({{INTERACTION_IDS}});
-- ============================================================================
-- EXAMPLE QUERIES
-- ============================================================================
-- Steps for specific interaction
-- SELECT * FROM ssot__AIAgentInteractionStep__dlm
-- WHERE ssot__AiAgentInteractionId__c = 'a0y1234567890ABC';
-- Action step distribution
-- SELECT
-- ssot__Name__c as action_name,
-- COUNT(*) as count
-- FROM ssot__AIAgentInteractionStep__dlm
-- WHERE ssot__AiAgentInteractionStepType__c = 'ACTION_STEP'
-- GROUP BY ssot__Name__c
-- ORDER BY count DESC;
-- LLM vs Action step ratio
-- SELECT
-- ssot__AiAgentInteractionStepType__c,
-- COUNT(*) as count
-- FROM ssot__AIAgentInteractionStep__dlm
-- GROUP BY ssot__AiAgentInteractionStepType__c;
-- Find steps with large outputs (potential issues)
-- SELECT
-- ssot__Id__c,
-- ssot__Name__c,
-- LENGTH(ssot__OutputValueText__c) as output_length
-- FROM ssot__AIAgentInteractionStep__dlm
-- WHERE LENGTH(ssot__OutputValueText__c) > 10000
-- ORDER BY output_length DESC;
-- Steps with specific action name
-- SELECT * FROM ssot__AIAgentInteractionStep__dlm
-- WHERE ssot__Name__c = 'Get_Order_Status'
-- AND ssot__AiAgentInteractionStepType__c = 'ACTION_STEP';
Credits & Dependencies
Core Dependencies
Data Processing
| Package | Version | License | Purpose |
|---|---|---|---|
| Polars | >=1.0.0 | MIT | High-performance DataFrame library with lazy evaluation |
| PyArrow | >=15.0.0 | Apache 2.0 | Parquet file format support and columnar processing |
| Pydantic | >=2.6.0 | MIT | Data validation and settings management |
Authentication
| Package | Version | License | Purpose |
|---|---|---|---|
| PyJWT | >=2.8.0 | MIT | JWT token generation for Salesforce auth |
| cryptography | >=42.0.0 | Apache 2.0 / BSD | Certificate handling for JWT Bearer flow |
HTTP & CLI
| Package | Version | License | Purpose |
|---|---|---|---|
| httpx | >=0.27.0 | BSD | Modern async HTTP client for Data Cloud API |
| Click | >=8.1.0 | BSD | CLI framework |
| Rich | >=13.0.0 | MIT | Terminal formatting, progress bars, tables |
Salesforce APIs Used
| API | Version | Purpose |
|---|---|---|
| Data Cloud Query API | v60.0+ | Execute SQL queries against DMOs |
| Data Cloud Profile API | v60.0+ | Retrieve metadata about DMOs |
| OAuth 2.0 JWT Bearer | - | Server-to-server authentication |
Related Salesforce Documentation
Inspiration & Patterns
- sf-data skill: Data extraction patterns and validation scoring
- sf-connected-apps skill: JWT authentication templates
- sf-ai-agentforce-testing skill: Agentforce integration patterns
Acknowledgments
| Contributor | Role | Contribution |
|---|---|---|
| Alejandro Raigon | Forward Deployed Engineer Director, Anthropic | Domain expertise on Agentforce observability patterns, Session Tracing data model insights, and quality analysis approaches |
License
This skill is released under the MIT License.
MIT License
Copyright (c) 2024-2026 Jag Valaiyapathy
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.#!/usr/bin/env python3
"""
PostToolUse hook: Suggest relevant Polars analysis based on extraction results.
After extracting STDM data, suggests appropriate analysis commands
based on the data volume and patterns found.
Usage (called automatically by hook system):
Receives tool result via stdin, outputs suggestions to stdout.
"""
import sys
import json
from pathlib import Path
from typing import Dict, Any, List, Optional
def analyze_extraction_and_suggest(data_dir: str) -> List[str]:
"""
Analyze extracted data and suggest relevant analysis commands.
Args:
data_dir: Path to extraction directory
Returns:
List of suggested commands/actions
"""
suggestions = []
path = Path(data_dir)
if not path.exists():
return suggestions
# Check metadata for extraction info
metadata_file = path / "metadata" / "extraction.json"
if metadata_file.exists():
with open(metadata_file) as f:
metadata = json.load(f)
results = metadata.get("results", {})
# Suggest based on record counts
session_count = results.get("sessions_count", 0)
interaction_count = results.get("interactions_count", 0)
step_count = results.get("steps_count", 0)
if session_count > 0:
suggestions.append(
f"📊 **Session Summary**: Run `python3 scripts/cli.py analyze --data-dir {data_dir}` "
f"to see summary statistics for {session_count} sessions"
)
if session_count > 100:
suggestions.append(
"📈 **Trend Analysis**: Consider analyzing sessions by date with "
"`analyzer.sessions_by_date()` to identify patterns"
)
if interaction_count > 0 and step_count > 0:
suggestions.append(
"🎯 **Topic Analysis**: Run `python3 scripts/cli.py topics --data-dir {data_dir}` "
"to see which topics are handling the most conversations"
)
suggestions.append(
"⚡ **Action Analysis**: Run `python3 scripts/cli.py actions --data-dir {data_dir}` "
"to see most frequently used actions"
)
# Check for failed sessions
if session_count > 0:
suggestions.append(
"❌ **Failed Sessions**: Use `analyzer.find_failed_sessions()` "
"to identify sessions that ended with escalation or abandonment"
)
# Check for specific session IDs in extraction
sessions_dir = path / "sessions"
if sessions_dir.exists():
try:
import pyarrow.parquet as pq
parquet_files = list(sessions_dir.glob("**/*.parquet"))
if parquet_files:
table = pq.read_table(parquet_files[0], columns=["ssot__Id__c"])
sample_ids = table.column("ssot__Id__c").to_pylist()[:3]
if sample_ids:
suggestions.append(
f"🔍 **Debug Session**: To investigate a specific session, run:\n"
f" `python3 scripts/cli.py debug-session --data-dir {data_dir} "
f"--session-id \"{sample_ids[0]}\"`"
)
except ImportError:
pass
except Exception:
pass
return suggestions
def suggest_next_skills(data_dir: str) -> List[str]:
"""
Suggest related skills based on analysis findings.
Args:
data_dir: Path to extraction directory
Returns:
List of skill suggestions
"""
suggestions = []
path = Path(data_dir)
if not path.exists():
return suggestions
# Always suggest testing after observability
suggestions.append(
"🧪 **Create Tests**: Use `Skill(skill=\"sf-ai-agentforce-testing\")` "
"to create test cases based on observed patterns"
)
# Suggest agent script fixes if needed
suggestions.append(
"🔧 **Fix Issues**: If you find topic routing issues, use "
"`Skill(skill=\"sf-ai-agentscript\")` to improve agent definitions"
)
return suggestions
def format_suggestions(analysis_suggestions: List[str], skill_suggestions: List[str]) -> str:
"""Format all suggestions for display."""
lines = []
if analysis_suggestions:
lines.append("\n💡 **Suggested Analysis:**")
for suggestion in analysis_suggestions:
lines.append(f"\n{suggestion}")
if skill_suggestions:
lines.append("\n\n🔗 **Related Skills:**")
for suggestion in skill_suggestions:
lines.append(f"\n{suggestion}")
return "\n".join(lines) if lines else ""
def extract_data_dir_from_tool_result(tool_result: Dict[str, Any]) -> Optional[str]:
"""Extract data directory path from tool result."""
# Check for output directory in command
output = tool_result.get("output", "")
if isinstance(output, str):
# Look for output directory mentions
import re
dir_match = re.search(r'Output:\s*(/[^\s]+|\.\/[^\s]+)', output)
if dir_match:
return dir_match.group(1)
# Look for stdm_data pattern
dir_match = re.search(r'(/[^\s]+/stdm_data|/[^\s]+/stdm_debug|\.\/stdm_data|\.\/stdm_debug)', output)
if dir_match:
return dir_match.group(1)
return None
def main():
"""Main hook entry point."""
# Read tool result from stdin
try:
input_data = sys.stdin.read()
if not input_data.strip():
sys.exit(0)
tool_result = json.loads(input_data)
except json.JSONDecodeError:
sys.exit(0)
# Extract data directory from tool result
data_dir = extract_data_dir_from_tool_result(tool_result)
if not data_dir:
# Try common default paths
for default in ["./stdm_data", "./stdm_debug"]:
if Path(default).exists():
data_dir = default
break
if not data_dir or not Path(data_dir).exists():
sys.exit(0)
# Generate suggestions
analysis_suggestions = analyze_extraction_and_suggest(data_dir)
skill_suggestions = suggest_next_skills(data_dir)
# Only output if we have suggestions
output = format_suggestions(analysis_suggestions, skill_suggestions)
if output:
print(output)
if __name__ == "__main__":
main()
#!/usr/bin/env python3
"""
PostToolUse hook: Validate Parquet extraction output.
Follows sf-data validation patterns to ensure extracted data is valid
before analysis.
Validates:
- Schema matches expected DMO structure
- No null primary keys
- Timestamps are valid ISO format
- Record count > 0
Usage (called automatically by hook system):
Receives tool result via stdin, outputs validation result to stdout.
"""
import sys
import json
import re
from pathlib import Path
from typing import Dict, Any, List, Optional
def validate_parquet_file(file_path: str) -> Dict[str, Any]:
"""
Validate a Parquet file against expected STDM schema.
Args:
file_path: Path to Parquet file
Returns:
Validation result dict with:
- valid: bool
- record_count: int
- warnings: list of warning messages
- errors: list of error messages
"""
result = {
"valid": True,
"record_count": 0,
"warnings": [],
"errors": [],
}
path = Path(file_path)
if not path.exists():
result["valid"] = False
result["errors"].append(f"File not found: {file_path}")
return result
try:
import pyarrow.parquet as pq
# Read metadata without loading data
parquet_file = pq.ParquetFile(file_path)
metadata = parquet_file.metadata
result["record_count"] = metadata.num_rows
if metadata.num_rows == 0:
result["warnings"].append("File contains 0 records")
# Validate schema
schema = parquet_file.schema_arrow
field_names = [field.name for field in schema]
# Check for expected STDM fields based on file type
if "sessions" in str(file_path).lower():
expected = ["ssot__Id__c", "ssot__AiAgentApiName__c", "ssot__StartTimestamp__c"]
elif "interactions" in str(file_path).lower():
expected = ["ssot__Id__c", "ssot__AiAgentSessionId__c", "ssot__AiAgentInteractionType__c"]
elif "steps" in str(file_path).lower():
expected = ["ssot__Id__c", "ssot__AiAgentInteractionId__c", "ssot__AiAgentInteractionStepType__c"]
elif "messages" in str(file_path).lower():
expected = ["ssot__Id__c", "ssot__AiAgentInteractionId__c", "ssot__ContentText__c"]
else:
expected = ["ssot__Id__c"] # At minimum, expect an ID field
missing = [f for f in expected if f not in field_names]
if missing:
result["warnings"].append(f"Missing expected fields: {missing}")
# Check for null primary keys (sample first batch)
table = pq.read_table(file_path, columns=["ssot__Id__c"])
id_column = table.column("ssot__Id__c")
null_count = id_column.null_count
if null_count > 0:
result["valid"] = False
result["errors"].append(f"Found {null_count} null primary key values")
except ImportError:
result["warnings"].append("PyArrow not installed, skipping schema validation")
except Exception as e:
result["valid"] = False
result["errors"].append(f"Validation error: {str(e)}")
return result
def validate_extraction_directory(dir_path: str) -> Dict[str, Any]:
"""
Validate an extraction output directory.
Args:
dir_path: Path to extraction directory
Returns:
Aggregated validation result
"""
result = {
"valid": True,
"entity_counts": {},
"warnings": [],
"errors": [],
}
path = Path(dir_path)
if not path.exists():
result["valid"] = False
result["errors"].append(f"Directory not found: {dir_path}")
return result
# Check for expected subdirectories
expected_entities = ["sessions", "interactions", "steps", "messages"]
for entity in expected_entities:
entity_dir = path / entity
if entity_dir.exists():
parquet_files = list(entity_dir.glob("**/*.parquet"))
if parquet_files:
# Validate first parquet file
file_result = validate_parquet_file(str(parquet_files[0]))
result["entity_counts"][entity] = file_result["record_count"]
if not file_result["valid"]:
result["valid"] = False
result["errors"].extend(file_result["errors"])
result["warnings"].extend(file_result["warnings"])
else:
result["warnings"].append(f"No Parquet files found in {entity}")
else:
result["warnings"].append(f"Missing {entity} directory")
# Check for metadata
metadata_file = path / "metadata" / "extraction.json"
if not metadata_file.exists():
result["warnings"].append("Missing extraction metadata file")
return result
def format_output(result: Dict[str, Any]) -> str:
"""Format validation result for display."""
lines = []
if result["valid"]:
lines.append("✅ Validation passed")
else:
lines.append("❌ Validation failed")
# Entity counts
if result.get("entity_counts"):
lines.append("\nRecord counts:")
for entity, count in result["entity_counts"].items():
lines.append(f" • {entity}: {count:,}")
# Single file count
if result.get("record_count"):
lines.append(f"\nRecords: {result['record_count']:,}")
# Warnings
if result.get("warnings"):
lines.append("\n⚠️ Warnings:")
for warning in result["warnings"]:
lines.append(f" • {warning}")
# Errors
if result.get("errors"):
lines.append("\n❌ Errors:")
for error in result["errors"]:
lines.append(f" • {error}")
return "\n".join(lines)
def extract_path_from_tool_result(tool_result: Dict[str, Any]) -> Optional[str]:
"""Extract file/directory path from tool result."""
# Check for file_path in Write tool result
if "file_path" in tool_result:
return tool_result["file_path"]
# Check for output path in command output
output = tool_result.get("output", "")
if isinstance(output, str):
# Look for parquet file paths
parquet_match = re.search(r'(/[^\s]+\.parquet)', output)
if parquet_match:
return parquet_match.group(1)
# Look for directory paths ending with stdm_data or similar
dir_match = re.search(r'(/[^\s]+/stdm_data|/[^\s]+/stdm_debug)', output)
if dir_match:
return dir_match.group(1)
return None
def main():
"""Main hook entry point."""
# Read tool result from stdin
try:
input_data = sys.stdin.read()
if not input_data.strip():
# No input, nothing to validate
sys.exit(0)
tool_result = json.loads(input_data)
except json.JSONDecodeError:
# Not JSON input, skip validation
sys.exit(0)
# Extract path from tool result
path = extract_path_from_tool_result(tool_result)
if not path:
# No relevant path found, skip validation
sys.exit(0)
path_obj = Path(path)
# Skip if path doesn't look like STDM data
if not any(x in str(path).lower() for x in ["stdm", "sessions", "interactions", "steps", "messages", ".parquet"]):
sys.exit(0)
# Determine validation type
if path_obj.is_file() and path.endswith(".parquet"):
result = validate_parquet_file(path)
elif path_obj.is_dir():
result = validate_extraction_directory(path)
else:
# File doesn't exist yet or not relevant
sys.exit(0)
# Output result
print(format_output(result))
# Exit with error code if validation failed
if not result["valid"]:
sys.exit(1)
if __name__ == "__main__":
main()
MIT License
Copyright (c) 2024-2026 Jag Valaiyapathy
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
sf-ai-agentforce-observability
Extract and analyze Agentforce session tracing data from Salesforce Data Cloud.
Status: General Availability (GA)
This skill has been validated against live Salesforce orgs and is production-ready.
| Metric | Value |
|---|---|
| Test Coverage | 260+ tests across 6 tiers |
| DMO Discovery | 24 DMOs verified, 3 not found (RAG Quality) |
| Live API Validation | All SQL patterns tested against Data Cloud |
| Schema Accuracy | Verified column names match actual API |
| Last Validated | January 30, 2026 |
Features
- High-Volume Extraction: Handle 1-10M records/day via Data Cloud Query API
- Parquet Storage: Efficient columnar storage (10x smaller than JSON)
- Polars Analysis: Lazy evaluation for memory-efficient analysis of 100M+ rows
- Session Debugging: Reconstruct session timelines for troubleshooting
- Incremental Sync: Watermark-based extraction for continuous monitoring
- GenAI Quality Analysis: Extract Trust Layer metrics (toxicity, adherence, resolution)
Quick Start
1. Prerequisites
# Install Python dependencies
pip install polars pyarrow pyjwt cryptography httpx rich click pydantic
# Verify Data Cloud access
sf org display --target-org myorg2. Configure Authentication
Session tracing extraction requires JWT Bearer auth to the Data Cloud Query API.
# Generate certificate
openssl req -x509 -sha256 -nodes -days 365 -newkey rsa:2048 \
-keyout ~/.sf/jwt/myorg.key \
-out ~/.sf/jwt/myorg.crt \
-subj "/CN=DataCloudAuth"
# Create External Client App (via sf-connected-apps skill)
# Required scopes: cdp_query_api, cdp_profile_apiSee references/auth-setup.md for detailed instructions.
3. Extract Session Data
# Extract last 7 days
python3 scripts/cli.py extract --org myorg --days 7 --output ./data
# Extract specific date range
python3 scripts/cli.py extract --org myorg --since 2026-01-01 --until 2026-01-15
# Extract complete session tree for debugging
python3 scripts/cli.py extract-tree --org myorg --session-id "a0x..."4. Analyze Data
# Session summary
python3 scripts/cli.py analyze --data-dir ./data
# Debug specific session
python3 scripts/cli.py debug-session --data-dir ./data --session-id "a0x..."
# Topic analysis
python3 scripts/cli.py topics --data-dir ./data
# Extract GenAI quality metrics
python3 scripts/cli.py extract-quality --data-dir ./data5. Use with Python
from scripts.analyzer import STDMAnalyzer
from pathlib import Path
analyzer = STDMAnalyzer(Path("./data"))
# Session summary
print(analyzer.session_summary())
# Step distribution
print(analyzer.step_distribution())
# Message timeline for debugging
print(analyzer.message_timeline("a0x..."))Data Model
24 DMOs verified via T6 live API testing (January 2026):
Session Tracing DMOs (5)
AIAgentSession (18 fields)
├── AIAgentSessionParticipant (12 fields) - Roles: USER, AGENT
├── AIAgentInteraction (20 fields) - Types: TURN, SESSION_END
│ ├── AIAgentInteractionStep (23 fields) - Types: LLM_STEP, ACTION_STEP, TOPIC_STEP
│ └── AIAgentInteractionMessage (21 fields) - Types: Input, Output
└── AIAgentMoment (13 fields) - Contains Agent API nameGenAI Audit & Feedback DMOs (13) ✅ T6 Verified
GenAIGatewayRequest (30 fields) - LLM call details, token usage
├── GenAIGatewayResponse (8 fields)
├── GenAIGeneration (11 fields) - LLM output text
│ ├── GenAIContentQuality (10 fields) - Trust Layer assessment
│ │ └── GenAIContentCategory (10 fields) - Detector results
│ └── GenAIFeedback (16 fields) - User thumbs up/down
│ └── GenAIFeedbackDetail (10 fields) - Feedback comments
└── GenAIGatewayRequestTag, GenAIGtwyRequestMetadata, GenAIGtwyObjRecord...Key Enum Values (Live API Verified)
| Entity | Field | Values |
|---|---|---|
| Session | ChannelType | E & O, Builder, SCRT2 - EmbeddedMessaging, Voice, NGC |
| Participant | AgentType | EinsteinServiceAgent, AgentforceEmployeeAgent, AgentforceServiceAgent |
| Participant | Role | USER, AGENT |
| Step | StepType | LLM_STEP, ACTION_STEP, TOPIC_STEP, SESSION_END |
| ContentCategory | DetectorType | TOXICITY, PII, PROMPT_DEFENSE, InstructionAdherence |
DMOs NOT Found (3)
GenAIRetrieverResponse__dlm❌GenAIRetrieverRequest__dlm❌GenAIRetrieverQualityMetric__dlm❌
Important: Data Cloud uses AiAgent (lowercase 'i') in field names, not AIAgent.
See references/data-model-reference.md for full schema.
Output Format
Data is stored in Parquet format:
stdm_data/
├── sessions/data.parquet
├── interactions/data.parquet
├── steps/data.parquet
├── messages/data.parquet
├── generations/data.parquet # GenAI quality
├── content_quality/data.parquet # Trust Layer
├── content_categories/data.parquet # Toxicity/Adherence
└── metadata/
├── extraction.json
└── watermark.jsonReal-Time Trace Analysis
For real-time agent trace capture and analysis (v1.1 PlanSuccessResponse with 13 step types), use sf-ai-agentforce-testing Phase F via sf agent preview start/send/end. This replaces the previous browser-based capture approach with a simpler CLI workflow.
CLI Reference
| Command | Description |
|---|---|
extract | Extract session data for time range |
extract-tree | Extract full tree for specific session |
extract-incremental | Continue from last extraction |
extract-quality | Extract GenAI Trust Layer metrics |
analyze | Generate summary statistics |
debug-session | Show session timeline |
topics | Topic routing analysis |
count | Count records per DMO |
trace-test | Capture Builder execution traces |
See references/cli-reference.md for all options.
Validation
This skill includes comprehensive validation testing:
| Tier | Category | Tests | Description |
|---|---|---|---|
| T1 | Auth & Connectivity | 5 | JWT auth, API access, DMO existence |
| T2 | Extraction Commands | 35 | CLI extract, tree, incremental |
| T3 | Analysis Commands | 46 | Analyze, debug-session, topics |
| T4 | Schema/Documentation | 96 | Field validation, query patterns |
| T5 | Negative Cases | 12 | Error handling, invalid args |
| T6 | Live SQL Execution | 39 | All SQL patterns against live API |
| T6 | DMO Discovery | 27 | Probe all 27 DMOs for existence |
| T6 | Field Discovery | 47 | Discover all fields per DMO |
Total: 260+ tests | 100% pass rate (34 discovery tests skip as expected for non-existent DMOs)
Run validation:
cd validation
source .venv/bin/activate
pytest scenarios/ -v --org YourOrgAliasIntegration with Other Skills
| Skill | Use Case |
|---|---|
sf-connected-apps | Set up JWT Bearer auth |
sf-ai-agentscript | Fix agents based on trace analysis |
sf-ai-agentforce-testing | Create tests from observed patterns |
sf-debug | Deep-dive into action failures |
Requirements
- Python 3.10+
- Salesforce org with Data Cloud and Agentforce enabled
- Session Tracing enabled in Agentforce settings
- JWT Bearer auth configured (via External Client App)
- Salesforce Standard Data Model v1.124+
Resources
- Data Model Reference - Full STDM schema
- Query Patterns - SQL examples for Data Cloud
- Analysis Cookbook - Polars analysis patterns
- Polars Cheatsheet - Quick reference
License
MIT License - See LICENSE file.
Author
Jag Valaiyapathy
---
Last updated: January 2026 | Validated against live org
<!-- Parent: sf-ai-agentforce-observability/SKILL.md -->
Agent Execution Lifecycle — 6-Phase Model
How the Agentforce platform processes each user message internally, based on live Builder trace captures. Maps AgentScript syntax to the 13 real-time trace step types.
Visual version: Generated runtime artifact at ~/.agent/diagrams/agent-execution-lifecycle.html with Mermaid flowcharts, timing bars, and side-by-side script-to-trace mapping.Data source: Live Builder traces from a test agent, sandbox org, 2026-02-25.
---
The 6 Phases
Every user message triggers this pipeline:
Phase 1: Input & Init (automatic — platform-managed)
Phase 2: Topic Selection (script-driven — AgentScript start_agent)
↓ Decision: LLM invoked go_* tool?
├─ Yes → Phase 3 + Phase 4
└─ No → Phase 5 (short-circuit)
Phase 3: Topic Transition (automatic — TransitionStep)
Phase 4: Topic Execution (LLM + actions — AgentScript topic)
Phase 5: Trust Layer (automatic — grounding + safety)
Phase 6: Response Delivery (automatic — final message)Full path (Turn 1): 1 → 2 → 3 → 4 → 5 → 6 — 46 steps, 4.6s Short-circuit (Turn 2): 1 → 2 → 5 → 6 — 15 steps, 2.6s
The short-circuit occurs when the Topic Selector has enough conversation context to respond directly without routing to a topic.
---
Phase → Step Type Mapping
Phase 1: Input & Init
| Step Type | Count | Duration | Data |
|---|---|---|---|
| UserInputStep | 1 | 0ms | User message text |
| SessionInitialStateStep | 1 | 6ms | Locale, input, conversation context |
Phase 2: Topic Selection
| Step Type | Count | Duration | Data |
|---|---|---|---|
| NodeEntryStateStep | 1 | 5ms | 24 state variables for topic_selector |
| VariableUpdateStep | 3 | ~1ms each | Condition checks (routable_id, mutable_case_id) |
| BeforeReasoningStep | 1 | 5ms | 5 setup actions |
| VariableUpdateStep | 6 | ~1ms each | Progressive instruction assembly |
| BeforeReasoningIterationStep | 1 | 9ms | 8 actions (setup + instruction assembly) |
| EnabledToolsStep | 1 | 2ms | 5 tools (go_product_help, go_escalation, 3 guardrails) |
| LLMStep | 1 | 624ms | Tool invocation: go_product_help |
| VariableUpdateStep | 2 | 0ms | next_topic: "__EMPTY__" → "product_help" → "__EMPTY__" |
Phase 3: Topic Transition
| Step Type | Count | Duration | Data |
|---|---|---|---|
| TransitionStep | 1 | 0ms | topic_selector → product_help, type: handoff, mode: manual |
Phase 4: Topic Execution
| Step Type | Count | Duration | Data |
|---|---|---|---|
| NodeEntryStateStep | 1 | 0ms | State inherited from topic_selector |
| BeforeReasoningStep | 1 | 3ms | 3 setup actions |
| VariableUpdateStep | 1 | 0ms | Condition check |
| FunctionStep | 1 | 1313ms | "Update Session Routing" Flow action |
| VariableUpdateStep | 10+ | ~1ms each | session_routed → true, instruction assembly |
| BeforeReasoningIterationStep | 1 | 1363ms | 13 actions total |
| EnabledToolsStep | 1 | 4ms | 7 tools (confirm_at_home, confirm_not_home, wrap_up, exit_to_escalation, 3 guardrails) |
| LLMStep | 1 | 1045ms | Text response: "Are you currently at home..." |
| AfterReasoningStep | 1 | 2ms | turn_count: 0 → 1 |
Phase 5: Trust Layer
| Step Type | Count | Duration | Data |
|---|---|---|---|
| ReasoningStep | 1 | 1003ms | category: GROUNDED, reason: "follows mandatory first question" |
| PlannerResponseStep | 1 | 0ms | safety_score: 0.99, all category_scores: 0 |
Phase 6: Response Delivery
The PlannerResponseStep message is delivered to the user. No additional trace steps.
---
Script → Trace Mapping Quick Reference
| AgentScript Construct | Trace Step Type | Notes |
|---|---|---|
start_agent entry: | NodeEntryStateStep | Initializes topic_selector state |
set @variables.x = ... | VariableUpdateStep | One step per set statement |
instructions: -> with set | VariableUpdateStep chain | Progressive instruction assembly |
actions: block | EnabledToolsStep | Lists all available tools for the LLM |
reasoning: | LLMStep | The actual LLM call with full prompt |
before_reasoning: | BeforeReasoningStep | Setup actions before LLM call |
after_reasoning: | AfterReasoningStep | Cleanup after LLM response |
run @actions.X | FunctionStep | Flow/Apex action execution |
transition to / go_* | TransitionStep | Topic-to-topic handoff |
if condition | VariableUpdateStep | Condition → true/false on AgentScriptInternal_condition |
template(...) expression | VariableUpdateStep | Evaluates template, updates agent_instructions |
---
Timing Analysis
Turn 1 (Full Path — 46 steps, 4,556ms)
| Component | Duration | Percentage |
|---|---|---|
| LLM calls (624 + 1045ms) | 1,669ms | 37% |
| Action calls (FunctionStep) | 1,313ms | 29% |
| Grounding evaluation | 1,003ms | 22% |
| Overhead (state, transitions) | 571ms | 12% |
Turn 2 (Short-Circuit — 15 steps, 2,564ms)
| Component | Duration | Percentage |
|---|---|---|
| LLM calls (1034ms) | 1,034ms | 40% |
| Action calls | 0ms | 0% |
| Grounding evaluation | 1,241ms | 49% |
| Overhead | 289ms | 11% |
Key insight: Grounding evaluation costs ~1s on every turn — as expensive as an LLM call. It's the fixed cost of Agentforce's safety guarantees.
---
Conversation Context Growth
The messages_sent[] array in LLMStep grows per turn:
- Turn 1: 3 messages (system prompt, user input, assembled instructions)
- Turn 2: 7 messages (system prompt, Turn 1 user/assistant/tool/assistant, Turn 2 user, instructions)
Each turn adds ~4 messages. After 10 turns, the LLM processes 30+ messages.
---
Dynamic Instruction Assembly
Each set on AgentScriptInternal_agent_instructions generates one VariableUpdateStep that appends a line via template evaluation. The instructions are reset to "" at the start of each phase and rebuilt from scratch.
Topic Selector assembly (Turn 1, 6 steps):
Step 7: "" → "\nCustomer: Test Customer"
Step 8: → + "This agent specializes in product troubleshooting..."
Step 9: → + "ROUTING:"
Step 10: → + "- ANY product issue → Product Help"
Step 11: → + "- Customer requests human agent → Escalation"
Step 12: → + "- ANY non-product request → Escalation"The final value becomes the last system message in the LLM prompt.
---
State Variables (24 at Entry)
Full state captured at NodeEntryStateStep (Turn 1):
| Variable | Type | Initial Value | Purpose |
|---|---|---|---|
authenticated | boolean | true | Pre-verified in session init |
customer_name | string | "Test Customer" | Display name for templates |
at_home | string | "unknown" | Controls instruction branching |
products_fetched | boolean | false | Gates product lookup |
product_confirmed | string | "pending" | Product selection state |
session_routed | boolean | false | Set true after FunctionStep |
turn_count | integer | 0 | Incremented in after_reasoning |
__user_input__ | string | "My camera..." | Current message (auto-set) |
__resolved_locale__ | string | "en_US" | Language for response |
__current_date_time__ | string | ISO timestamp | Auto-updated each turn |
---
STDM Span Types (Testing Center Trace View)
When viewing session traces in Testing Center (not Builder), steps are collapsed into 6 color-coded span types:
| Span Type | Color | Maps to Builder Steps |
|---|---|---|
| TOPIC_STEP | Blue | NodeEntryStateStep, TransitionStep |
| LLM_STEP | Purple | LLMStep, EnabledToolsStep |
| ACTION_STEP | Green | FunctionStep |
| TRUST_GUARDRAILS_STEP | Orange | ReasoningStep, PlannerResponseStep |
| SESSION_END | Gray | Session termination |
| SYSTEM_STEP | Light gray | SessionInitialStateStep, VariableUpdateStep |
This mapping is critical for correlating Builder-level debugging with Testing Center trace analysis. A single TOPIC_STEP span in Testing Center may represent multiple NodeEntryStateStep + TransitionStep events at the Builder level.
---
Related Documents
- builder-trace-api.md — Reverse-engineered API endpoints and capture methodology
- data-model-reference.md — Complete 24-DMO STDM schema (persisted layer)
- query-patterns.md — Data Cloud SQL for persisted trace data
<!-- Parent: sf-ai-agentforce-observability/SKILL.md -->
Analysis Cookbook
Common analysis patterns using Polars for session tracing data.
Getting Started
from pathlib import Path
import polars as pl
# Load data as lazy frames (memory efficient)
data_dir = Path("./stdm_data")
sessions = pl.scan_parquet(data_dir / "sessions" / "**/*.parquet")
interactions = pl.scan_parquet(data_dir / "interactions" / "**/*.parquet")
steps = pl.scan_parquet(data_dir / "steps" / "**/*.parquet")
messages = pl.scan_parquet(data_dir / "messages" / "**/*.parquet")---
Session Analysis
Basic Statistics
# Session count by agent
sessions.group_by("ssot__AiAgentApiName__c").agg(
pl.count().alias("session_count")
).sort("session_count", descending=True).collect()Completion Rate by Agent
sessions.group_by("ssot__AiAgentApiName__c").agg([
pl.count().alias("total"),
pl.col("ssot__AiAgentSessionEndType__c")
.filter(pl.col("ssot__AiAgentSessionEndType__c") == "Completed")
.count().alias("completed")
]).with_columns([
(pl.col("completed") / pl.col("total") * 100).round(1).alias("completion_rate")
]).collect()Session Duration Analysis
# Note: Requires timestamp parsing
sessions.with_columns([
pl.col("ssot__StartTimestamp__c").str.to_datetime().alias("start"),
pl.col("ssot__EndTimestamp__c").str.to_datetime().alias("end"),
]).with_columns([
(pl.col("end") - pl.col("start")).alias("duration")
]).group_by("ssot__AiAgentApiName__c").agg([
pl.col("duration").mean().alias("avg_duration"),
pl.col("duration").max().alias("max_duration"),
]).collect()---
Interaction Analysis
Turns per Session
turns_per_session = (
interactions
.filter(pl.col("ssot__AiAgentInteractionType__c") == "TURN")
.group_by("ssot__AiAgentSessionId__c")
.agg(pl.count().alias("turn_count"))
)
# Distribution
turns_per_session.group_by("turn_count").agg(
pl.count().alias("sessions")
).sort("turn_count").collect()Topic Routing
# Most common topics
interactions.filter(
pl.col("ssot__AiAgentInteractionType__c") == "TURN"
).group_by("ssot__TopicApiName__c").agg(
pl.count().alias("turn_count"),
pl.col("ssot__AiAgentSessionId__c").n_unique().alias("session_count")
).sort("turn_count", descending=True).collect()Topic Switches (Multi-Topic Sessions)
# Sessions that used multiple topics
topic_counts = (
interactions
.filter(pl.col("ssot__AiAgentInteractionType__c") == "TURN")
.group_by("ssot__AiAgentSessionId__c")
.agg(pl.col("ssot__TopicApiName__c").n_unique().alias("topic_count"))
)
# Sessions with 2+ topics
topic_counts.filter(pl.col("topic_count") > 1).collect()---
Step Analysis
LLM vs Action Ratio
steps.group_by("ssot__AiAgentInteractionStepType__c").agg(
pl.count().alias("count")
).with_columns([
(pl.col("count") / pl.col("count").sum() * 100).round(1).alias("percentage")
]).collect()Most Used Actions
steps.filter(
pl.col("ssot__AiAgentInteractionStepType__c") == "ACTION_STEP"
).group_by("ssot__Name__c").agg(
pl.count().alias("invocations")
).sort("invocations", descending=True).head(20).collect()Steps per Turn Distribution
steps_per_turn = (
steps
.group_by("ssot__AiAgentInteractionId__c")
.agg(pl.count().alias("step_count"))
)
steps_per_turn.group_by("step_count").agg(
pl.count().alias("turns")
).sort("step_count").collect()Action Input/Output Analysis
# Parse JSON in action inputs (example for specific action)
action_steps = steps.filter(
(pl.col("ssot__AiAgentInteractionStepType__c") == "ACTION_STEP") &
(pl.col("ssot__Name__c") == "Get_Order_Status")
).collect()
# Parse inputs
import json
for row in action_steps.iter_rows(named=True):
input_data = json.loads(row["ssot__InputValueText__c"] or "{}")
output_data = json.loads(row["ssot__OutputValueText__c"] or "{}")
print(f"Input: {input_data}, Output: {output_data}")---
Message Analysis
Message Length Distribution
messages.with_columns([
pl.col("ssot__ContentText__c").str.len_chars().alias("length")
]).group_by("ssot__AiAgentInteractionMessageType__c").agg([
pl.col("length").mean().alias("avg_length"),
pl.col("length").max().alias("max_length"),
]).collect()Common User Phrases
# Word frequency in user messages (simple)
user_messages = messages.filter(
pl.col("ssot__AiAgentInteractionMessageType__c") == "INPUT"
).select("ssot__ContentText__c").collect()
# Count words
from collections import Counter
words = Counter()
for row in user_messages.iter_rows(named=True):
content = row["ssot__ContentText__c"] or ""
words.update(content.lower().split())
print(words.most_common(20))---
Time-Based Analysis
Sessions by Date
sessions.with_columns([
pl.col("ssot__StartTimestamp__c").str.slice(0, 10).alias("date")
]).group_by("date").agg(
pl.count().alias("sessions")
).sort("date").collect()Sessions by Hour
sessions.with_columns([
pl.col("ssot__StartTimestamp__c").str.slice(11, 2).alias("hour")
]).group_by("hour").agg(
pl.count().alias("sessions")
).sort("hour").collect()Day of Week Analysis
# Requires datetime conversion
sessions.with_columns([
pl.col("ssot__StartTimestamp__c")
.str.to_datetime()
.dt.weekday()
.alias("weekday")
]).group_by("weekday").agg(
pl.count().alias("sessions")
).sort("weekday").collect()---
Debugging Patterns
Find Failed Sessions
failed = sessions.filter(
pl.col("ssot__AiAgentSessionEndType__c").is_in(["Escalated", "Abandoned", "Failed"])
).sort("ssot__StartTimestamp__c", descending=True).collect()
for row in failed.head(5).iter_rows(named=True):
print(f"Session: {row['ssot__Id__c']}")
print(f" Agent: {row['ssot__AiAgentApiName__c']}")
print(f" End: {row['ssot__AiAgentSessionEndType__c']}")
print()Session Timeline Reconstruction
def get_timeline(session_id: str) -> pl.DataFrame:
"""Reconstruct message timeline for a session."""
# Get interaction IDs
interaction_df = interactions.filter(
pl.col("ssot__AiAgentSessionId__c") == session_id
).collect()
interaction_ids = interaction_df["ssot__Id__c"].to_list()
# Get messages
msg_df = messages.filter(
pl.col("ssot__AiAgentInteractionId__c").is_in(interaction_ids)
).sort("ssot__MessageSentTimestamp__c").collect()
return msg_df
# Usage
timeline = get_timeline("a0x1234567890ABC")
for row in timeline.iter_rows(named=True):
msg_type = row["ssot__AiAgentInteractionMessageType__c"]
content = row["ssot__ContentText__c"]
icon = "→" if msg_type == "INPUT" else "←"
print(f"{icon} {content[:100]}...")---
Performance Tips
Use Lazy Evaluation
# Good: Lazy evaluation, deferred execution
result = (
sessions
.filter(pl.col("ssot__AiAgentApiName__c") == "My_Agent")
.group_by("ssot__AiAgentSessionEndType__c")
.agg(pl.count())
.collect() # Execute here
)
# Avoid: Eager loading of everything
df = pl.read_parquet(data_dir / "sessions" / "**/*.parquet") # Loads all data
result = df.filter(...) # Then filterSelect Only Needed Columns
# Good: Select specific columns
sessions.select([
"ssot__Id__c",
"ssot__AiAgentApiName__c",
"ssot__AiAgentSessionEndType__c"
]).collect()
# Avoid: Select all columns
sessions.collect() # Includes all columnsUse Streaming for Large Results
# For very large datasets
for batch in sessions.collect(streaming=True):
# Process batch
pass<!-- Parent: sf-ai-agentforce-observability/SKILL.md -->
Analysis Examples
Examples for analyzing extracted STDM data with Polars and the CLI.
---
CLI Analysis Commands
Summary Statistics
Get high-level overview of extracted data:
stdm-extract analyze --data-dir ./stdm_dataOutput:
📊 SESSION SUMMARY
═══════════════════════════════════════════════════════════
Agent Sessions Completed Escalated Completion %
───────────────────────────────────────────────────────────────────────────
Customer_Support_Agent 450 382 45 84.9%
Order_Tracking_Agent 312 298 8 95.5%
FAQ_Agent 156 152 2 97.4%
───────────────────────────────────────────────────────────────────────────
TOTAL 918 832 55 90.6%Topic Analysis
See which topics handle the most turns:
stdm-extract topics --data-dir ./stdm_dataOutput:
📊 TOPIC ROUTING
═══════════════════════════════════════════════════════════
Topic Turns Sessions Avg Turns/Session
───────────────────────────────────────────────────────────────
Order_Status 1234 312 4.0
Return_Policy 856 245 3.5
Account_Help 654 198 3.3
General_FAQ 432 156 2.8Action Analysis
See which actions are invoked most:
stdm-extract actions --data-dir ./stdm_dataOutput:
📊 ACTION INVOCATIONS
═══════════════════════════════════════════════════════════
Action Count Avg/Session
─────────────────────────────────────────────────
Get_Order_Status 1856 2.0
Search_Knowledge_Base 1245 1.4
Create_Case 567 0.6
Update_Contact 234 0.3---
Python Analysis
Setup
import polars as pl
from pathlib import Path
data_dir = Path("./stdm_data")
# Lazy loading (memory efficient)
sessions = pl.scan_parquet(data_dir / "sessions" / "**/*.parquet")
interactions = pl.scan_parquet(data_dir / "interactions" / "**/*.parquet")
steps = pl.scan_parquet(data_dir / "steps" / "**/*.parquet")
messages = pl.scan_parquet(data_dir / "messages" / "**/*.parquet")Session Analysis
Sessions by End Type:
sessions.group_by("ssot__AiAgentSessionEndType__c").agg(
pl.count().alias("count")
).sort("count", descending=True).collect()Daily Session Trend:
sessions.with_columns(
pl.col("ssot__StartTimestamp__c").str.slice(0, 10).alias("date")
).group_by("date").agg(
pl.count().alias("sessions")
).sort("date").collect()Hourly Distribution:
sessions.with_columns(
pl.col("ssot__StartTimestamp__c").str.slice(11, 2).alias("hour")
).group_by("hour").agg(
pl.count().alias("sessions")
).sort("hour").collect()Turn Analysis
Turns Per Session:
turns_per_session = (
interactions
.filter(pl.col("ssot__AiAgentInteractionType__c") == "TURN")
.group_by("ssot__AiAgentSessionId__c")
.agg(pl.count().alias("turns"))
)
# Distribution
turns_per_session.group_by("turns").agg(
pl.count().alias("sessions")
).sort("turns").collect()Multi-Topic Sessions (Topic Switches):
topic_counts = (
interactions
.filter(pl.col("ssot__AiAgentInteractionType__c") == "TURN")
.group_by("ssot__AiAgentSessionId__c")
.agg(pl.col("ssot__TopicApiName__c").n_unique().alias("topics"))
)
# Sessions with 2+ topics
topic_counts.filter(pl.col("topics") > 1).collect()Step Analysis
LLM vs Action Ratio:
steps.group_by("ssot__AiAgentInteractionStepType__c").agg(
pl.count().alias("count")
).with_columns(
(pl.col("count") / pl.col("count").sum() * 100).round(1).alias("percent")
).collect()Most Used Actions:
steps.filter(
pl.col("ssot__AiAgentInteractionStepType__c") == "ACTION_STEP"
).group_by("ssot__Name__c").agg(
pl.count().alias("invocations")
).sort("invocations", descending=True).head(10).collect()Steps Per Turn:
steps.group_by("ssot__AiAgentInteractionId__c").agg(
pl.count().alias("steps")
).group_by("steps").agg(
pl.count().alias("turns")
).sort("steps").collect()Message Analysis
Average Message Length by Type:
messages.with_columns(
pl.col("ssot__ContentText__c").str.len_chars().alias("length")
).group_by("ssot__AiAgentInteractionMessageType__c").agg([
pl.col("length").mean().round(0).alias("avg_length"),
pl.col("length").max().alias("max_length"),
]).collect()Common User Phrases:
user_msgs = messages.filter(
pl.col("ssot__AiAgentInteractionMessageType__c") == "INPUT"
).select("ssot__ContentText__c").collect()
from collections import Counter
words = Counter()
for row in user_msgs.iter_rows(named=True):
text = row["ssot__ContentText__c"] or ""
words.update(text.lower().split())
print(words.most_common(20))---
Using the Analyzer Class
The skill includes a helper class for common analyses:
from scripts.analyzer import STDMAnalyzer
analyzer = STDMAnalyzer(Path("./stdm_data"))
# Quick summary
summary = analyzer.session_summary()
print(summary)
# Step breakdown
steps_df = analyzer.step_distribution()
print(steps_df)
# Topic patterns
topics_df = analyzer.topic_analysis()
print(topics_df)
# Find failed sessions
failed = analyzer.find_failed_sessions(limit=10)
print(failed)---
Exporting Results
To CSV
result = sessions.group_by(...).agg(...).collect()
result.write_csv("output.csv")To JSON
result.write_json("output.json")To Parquet (for further analysis)
result.write_parquet("output.parquet")---
Visualization (with Matplotlib)
import matplotlib.pyplot as plt
# Daily trend
daily = sessions.with_columns(
pl.col("ssot__StartTimestamp__c").str.slice(0, 10).alias("date")
).group_by("date").agg(
pl.count().alias("sessions")
).sort("date").collect()
plt.figure(figsize=(12, 4))
plt.bar(daily["date"], daily["sessions"])
plt.xticks(rotation=45)
plt.title("Sessions per Day")
plt.tight_layout()
plt.savefig("daily_trend.png")---
See Also
- Polars Cheatsheet - Quick reference
- Analysis Cookbook - More recipes
- Debugging Sessions - Session-level debugging
<!-- Parent: sf-ai-agentforce-observability/SKILL.md -->
Authentication Setup
This skill uses JWT Bearer authentication to access the Data Cloud Query API. Authentication is configured via an External Client App (ECA) in Salesforce.
Prerequisites
1. Salesforce CLI installed and authenticated to your org 2. OpenSSL for certificate generation 3. Data Cloud enabled in your org 4. System Administrator profile (or appropriate permissions for ECA setup)
---
Quick Setup (5 Steps)
Step 1: Generate JWT Certificate
# Create directory for JWT keys
mkdir -p ~/.sf/jwt
# Generate private key and self-signed certificate (valid 1 year)
# Naming convention: {org_alias}-agentforce-observability.key
openssl req -x509 -sha256 -nodes -days 365 \
-newkey rsa:2048 \
-keyout ~/.sf/jwt/YourOrg-agentforce-observability.key \
-out ~/.sf/jwt/YourOrg-agentforce-observability.crt \
-subj "/CN=AgentforceObservability/O=YourOrg"
# Secure the private key (required - Salesforce rejects world-readable keys)
chmod 600 ~/.sf/jwt/YourOrg-agentforce-observability.keyStep 2: Create External Client App
In Salesforce Setup:
1. Setup → External Client App Manager → New External Client App 2. Fill in the basic details:
| Field | Value |
|---|---|
| Name | Agentforce Observability |
| API Name | Agentforce_Observability |
| Description | JWT Bearer auth for Agentforce STDM extraction via Claude Code |
| Distribution State | Local |
3. Click Save
Step 3: Configure OAuth Settings
In the ECA → OAuth Settings tab:
| Setting | Value |
|---|---|
| Enable OAuth | ✅ Checked |
| Callback URL | https://login.salesforce.com/services/oauth2/callback |
| Selected OAuth Scopes | cdp_query_api, refresh_token, offline_access |
| Require PKCE | ❌ Unchecked (not needed for JWT Bearer) |
| Enable Client Credentials Flow | ❌ Optional |
Upload Certificate: 1. Check Use digital signatures 2. Click Choose File 3. Upload ~/.sf/jwt/{org}-agentforce-observability.crt 4. Click Save
Step 4: Configure App Policies
In the ECA → Policies tab:
| Setting | Value |
|---|---|
| Permitted Users | Admin approved users are pre-authorized |
| IP Relaxation | Relax IP restrictions (for CLI usage) |
Add Your User: 1. Click Manage → Manage Profiles or Manage Permission Sets 2. Add your user's profile or an appropriate permission set 3. Click Save
Step 5: Get Consumer Key & Test
1. Go to ECA → OAuth Settings tab 2. Copy the Consumer Key (starts with 3MVG9...) 3. Test authentication:
# Option 1: Pass consumer key directly
python3 scripts/cli.py test-auth \
--org YourOrg \
--consumer-key "3MVG9..."
# Option 2: Use environment variable
export SF_CONSUMER_KEY="3MVG9..."
python3 scripts/cli.py test-auth --org YourOrgExpected output:
Testing Authentication
Org: YourOrg
Key: /Users/you/.sf/jwt/YourOrg-agentforce-observability.key
Getting org info...
Instance: https://your-org.my.salesforce.com
Username: your.user@example.com
Sandbox: True
Testing token generation...
✓ Token generated
Testing Data Cloud access...
✓ Data Cloud accessible
Authentication successful!---
Key Path Resolution
The CLI resolves the private key path in this order:
1. Explicit `--key-path` argument (highest priority) 2. App-specific key: ~/.sf/jwt/{org_alias}-agentforce-observability.key 3. Generic org key: ~/.sf/jwt/{org_alias}.key (fallback)
Examples
# Uses app-specific key automatically
python3 scripts/cli.py test-auth --org YourOrg --consumer-key "..."
# → Resolves to: ~/.sf/jwt/YourOrg-agentforce-observability.key
# Explicit key path (overrides all defaults)
python3 scripts/cli.py test-auth --org YourOrg \
--consumer-key "..." \
--key-path ~/.sf/jwt/custom-key.key---
File Locations
| File | Location | Description |
|---|---|---|
| Private Key | ~/.sf/jwt/{org}-agentforce-observability.key | RSA private key (chmod 600) |
| Certificate | ~/.sf/jwt/{org}-agentforce-observability.crt | X.509 cert uploaded to Salesforce |
| Consumer Key | $SF_CONSUMER_KEY or --consumer-key | From ECA OAuth Settings |
---
Required OAuth Scopes
| Scope | Purpose | Required |
|---|---|---|
cdp_query_api | Execute Data Cloud SQL queries | ✅ Yes |
refresh_token, offline_access | Server-to-server access | ✅ Yes |
cdp_profile_api | Access profile and DMO metadata | Optional |
---
Troubleshooting
invalid_grant Error
RuntimeError: Token exchange failed: invalid_grantCauses & Fixes:
| Cause | Fix |
|---|---|
| Certificate mismatch | Re-upload .crt file to ECA |
| Expired certificate | Regenerate with openssl (see Step 1) |
| User not authorized | Add user to ECA policies (see Step 4) |
| Wrong login URL | Verify sandbox detection is working |
Verify certificate expiry:
openssl x509 -enddate -noout -in ~/.sf/jwt/YourOrg-agentforce-observability.crt403 Forbidden
RuntimeError: Access denied: Ensure ECA has cdp_query_api scopeFix: Add cdp_query_api scope to the ECA OAuth Settings.
Key Not Found
FileNotFoundError: Private key not found at ~/.sf/jwt/myorg.keyCauses: 1. Key file doesn't exist → Generate with Step 1 2. Wrong org alias → Check sf org list for correct alias 3. Using old naming convention → Rename to {org}-agentforce-observability.key
Permission Denied on Key
PermissionError: [Errno 13] Permission denied: '~/.sf/jwt/...'Fix:
chmod 600 ~/.sf/jwt/*.key---
Sandbox vs Production
The skill automatically detects sandbox orgs:
1. Reads isSandbox from sf org display --json 2. Uses https://test.salesforce.com for token exchange (sandboxes) 3. Uses https://login.salesforce.com for production
No additional configuration needed.
---
Security Best Practices
1. Protect private keys: Always chmod 600 ~/.sf/jwt/*.key 2. Rotate certificates annually: Regenerate before expiration 3. Use separate ECAs: One per org/environment for isolation 4. Never commit keys: Add *.key to .gitignore 5. Audit access: Review ECA usage in Setup → Security → Connected Apps OAuth Usage
---
Environment Variables
For automation/CI, set these environment variables:
# Consumer key (org-specific takes priority)
export SF_YOURORG_CONSUMER_KEY="3MVG9..."
# or generic fallback
export SF_CONSUMER_KEY="3MVG9..."The CLI checks in this order: 1. --consumer-key argument 2. SF_{ORG_ALIAS}_CONSUMER_KEY (uppercase, hyphens to underscores) 3. SF_CONSUMER_KEY
---
See Also
- sf-connected-apps skill - General ECA patterns
- CLI Reference - All command options
- Troubleshooting - More error solutions
<!-- Parent: sf-ai-agentforce-observability/SKILL.md -->
Basic Extraction Examples
Simple examples to get started with STDM extraction.
Prerequisites
1. JWT authentication configured (see Auth Setup) 2. Consumer key set: export SF_CONSUMER_KEY="3MVG9..." 3. Org alias available in sf org list
---
Example 1: Extract Last 7 Days
The simplest extraction - all sessions from the last 7 days:
stdm-extract extract --org prodOutput structure:
./stdm_data/
├── sessions/
│ └── data.parquet
├── interactions/
│ └── data.parquet
├── steps/
│ └── data.parquet
└── messages/
└── data.parquet---
Example 2: Extract Last 24 Hours
stdm-extract extract --org prod --days 1---
Example 3: Custom Output Directory
stdm-extract extract --org prod --output /data/agentforce/prod---
Example 4: Verbose Mode
See detailed progress and timing:
stdm-extract extract --org prod --verboseSample output:
🔐 Authenticating to Data Cloud...
Instance URL: https://myorg.my.salesforce.com
📊 Extracting sessions...
Query: SELECT ... FROM ssot__AIAgentSession__dlm WHERE ...
Records: 1,234 (2.3s)
📊 Extracting interactions...
Records: 5,678 (4.1s)
📊 Extracting steps...
Records: 12,345 (8.7s)
📊 Extracting messages...
Records: 9,876 (5.2s)
✅ Extraction complete!
Total records: 29,133
Duration: 20.3s
Output: ./stdm_data/---
Example 5: Test Authentication First
Before extracting, verify your setup:
stdm-extract test-auth --org prodSuccess:
✅ Authentication successful
Instance URL: https://myorg.my.salesforce.com
Token valid for: 3599 secondsFailure:
❌ Authentication failed
Error: invalid_grant
Hint: Check certificate expiration with:
openssl x509 -enddate -noout -in ~/.sf/jwt/prod.key---
Example 6: Count Records Before Extraction
Check how much data exists without downloading:
stdm-extract count --org prod --dmo sessionsOutput:
📊 Record counts for prod:
Sessions: 12,345---
Example 7: Extract from Sandbox
Works the same as production - just use the sandbox alias:
# List your orgs
sf org list
# Extract from sandbox
stdm-extract extract --org mysandbox --days 3---
What's Next?
- Filtered Extraction - Filter by agent, date range
- Analysis Examples - Analyze extracted data
- Debugging Sessions - Debug specific sessions
---
Quick Reference
| Task | Command |
|---|---|
| Last 7 days | stdm-extract extract --org prod |
| Last N days | stdm-extract extract --org prod --days N |
| Test auth | stdm-extract test-auth --org prod |
| Check counts | stdm-extract count --org prod |
| Verbose mode | Add --verbose to any command |
<!-- Parent: sf-ai-agentforce-observability/SKILL.md -->
Billing Considerations & Troubleshooting
Billing Considerations
Reference: Billing Considerations for Agentforce Session Tracing
Agentforce Session Tracing consumes Data 360 credits for ingestion, storage, and processing.
Credit Consumption
| Usage Type | Digital Wallet Card | Description |
|---|---|---|
| Batch Data Pipeline | Data Services | Records ingested via data streams. ~24 records per LLM round-trip. Primary cost driver. |
| Data Queries | Data Services | Records processed when running queries, reports, dashboards |
| Streaming Calculated Insights | Data Services | Used for Prompt Builder usage and feedback metrics |
| Storage Beyond Allocation | Data Storage | Storage consumed above allocated amount |
Cost Estimation
Records per session ≈ Turns × 24 (avg per LLM call)
Daily records ≈ Sessions/day × Avg turns × 24
Example:
1,000 sessions/day × 4 turns × 24 = 96,000 records/day ingestedTip: Use Digital Wallet for near real-time consumption tracking.
---
Common Issues & Fixes
| Error | Cause | Fix |
|---|---|---|
401 Unauthorized | JWT auth expired/invalid | Refresh token or reconfigure ECA |
No session data | Tracing not enabled | Enable Session Tracing in Agent Settings |
Query timeout | Too much data | Add date filters, use incremental |
Memory error | Loading all data | Use Polars lazy frames |
Missing DMO | Wrong API version | Use API v60.0+ |
See references/troubleshooting.md for detailed solutions.
---
Agentforce Credit Costs (Cross-Reference)
When analyzing traces for performance optimization, understanding credit costs per step type is critical:
| Step Type | Credit Cost | Notes |
|---|---|---|
Framework operations (before/after_reasoning, set, if/else, @utils.*) | FREE | No credit consumption |
| Prompt Template invocations | 2-16 credits/call | Varies by template complexity |
| Flow/Apex action executions | 20 credits/call | Most expensive step type |
Optimization insight: When trace analysis reveals excessive action calls, consider consolidating multiple Flow actions into a single Apex action or using framework-level set operations where possible.
See sf-ai-agentscript/references/production-gotchas.md for the full credit consumption model and optimization strategies.
<!-- Parent: sf-ai-agentforce-observability/SKILL.md -->
Builder Trace API — Reverse-Engineered Internal Endpoint
How the Agentforce Builder's "Trace" tab renders real-time execution telemetry, the exact internal API endpoints, request/response schemas, and how to capture trace data programmatically.
Status: ✅ Confirmed via live network capture (February 2026).
---
Two-Layer Trace Architecture
The Agentforce Builder's Trace tab is powered by two distinct systems:
Layer 1: Real-Time Execution Trace (Aura Controller)
The live trace rendered during Builder testing uses an Aura controller action — confirmed via CDP network capture:
| Aspect | Detail |
|---|---|
| Endpoint | serviceComponent://ui.agent.authoring.components.aura.controller.AgentAuthoringController/ACTION$getSimulationPlanTraces |
| Protocol | Aura framework (POST to /aura) |
| Auth | Cookie-based browser session (no explicit Authorization header) |
| Trigger | Called automatically after SSE stream delivers INFORM event |
| Input | { planId, sessionId, version: "1.0" } |
| Output | PlanSuccessResponse with plan[] array of 13 step types |
| Latency | Real-time (called after agent responds) |
| Persistence | Transient — not stored in STDM |
Layer 2: Persisted Session Tracing (STDM in Data Cloud)
After 5-15 minutes, a subset of trace data is persisted to Data Cloud across 24 DMOs:
| Aspect | Detail |
|---|---|
| Storage | Data Cloud (Data 360) |
| Query API | Data Cloud Query API (SQL-like) |
| Auth | JWT Bearer via External Client App |
| Step Types | 5 (vs 13 in real-time trace) |
| Retention | 13 months (prod) / 30 days (sandbox) |
See data-model-reference.md and query-patterns.md for full STDM documentation.
---
Confirmed Architecture
The Builder uses a two-phase request pattern per user message:
Phase 1: Message Delivery (SSE Stream)
┌─────────────────────────────────────────────────────────────────┐
│ Builder LWC → POST /einstein/ai-agent/v1/sessions/{id}/ │
│ messages/stream │
│ Auth: Bearer token (JWT) │
│ Body: {"message":{"type":"Text","text":"...","sequenceId":"N"}} │
│ │
│ Response: SSE stream with events: │
│ event: TEXT_CHUNK → streaming word-by-word response │
│ event: INFORM → final message + planId + feedbackId │
│ event: END_OF_TURN → turn complete │
└─────────────────────────────────────────────────────────────────┘
│
│ planId from INFORM event
▼
Phase 2: Trace Retrieval (Aura Controller)
┌─────────────────────────────────────────────────────────────────┐
│ Builder LWC → POST /aura │
│ Descriptor: AgentAuthoringController/getSimulationPlanTraces │
│ Auth: Browser cookies (X-SFDC-Page-Cache, X-SFDC-Request-Id) │
│ Params: { planId, sessionId, version: "1.0" } │
│ │
│ Response: PlanSuccessResponse with plan[] array │
│ → 13 step types, full variable state, LLM prompts, timing │
└─────────────────────────────────────────────────────────────────┘Key discovery: The planId returned in the SSE INFORM event is the same ID used to fetch the trace. This links the message stream to the execution trace.
---
SSE Stream Schema (/messages/stream)
Request
POST https://api.salesforce.com/einstein/ai-agent/v1/sessions/{sessionId}/messages/stream
Authorization: Bearer {jwt_token}
Content-Type: application/json
x-salesforce-region: us-west-2
X-B3-TraceId: {distributed_trace_id}
X-B3-SpanId: {span_id}{
"message": {
"type": "Text",
"text": "Where is my order?",
"sequenceId": "1"
}
}SSE Event Types
| Event Type | Purpose | Key Fields |
|---|---|---|
TEXT_CHUNK | Streaming response text | offset, message.message (word/phrase) |
INFORM | Final complete response | message.message, planId, feedbackId, isContentSafe, citedReferences[] |
END_OF_TURN | Turn completion marker | message.type: "EndOfTurn" |
INFORM Event Schema (Critical — Contains planId)
{
"timestamp": 1772000000000,
"originEventId": "1772000000000-REQ",
"traceId": "00000000000000001234567890abcdef",
"offset": 0,
"message": {
"type": "Inform",
"feedbackId": "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
"isContentSafe": true,
"message": "I can help you with that. Let me look into it...",
"id": "11111111-2222-3333-4444-555555555555",
"metrics": {},
"planId": "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
"result": [],
"citedReferences": []
}
}Note:planIdandfeedbackIdare identical in the observed capture.
---
Trace API Schema (getSimulationPlanTraces)
Request (Aura)
POST /aura
Content-Type: application/x-www-form-urlencoded
X-SFDC-Page-Cache: {page_cache_id}
X-SFDC-Request-Id: {request_id}
X-B3-TraceId: {trace_id}URL-encoded body (decoded):
{
"actions": [{
"id": "48;a",
"descriptor": "serviceComponent://ui.agent.authoring.components.aura.controller.AgentAuthoringController/ACTION$getSimulationPlanTraces",
"callingDescriptor": "UNKNOWN",
"params": {
"planId": "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
"sessionId": "019xxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
"version": "1.0"
}
}]
}Response: PlanSuccessResponse
{
"type": "PlanSuccessResponse",
"planId": "aaaaaaaa-...",
"sessionId": "019xxxxx-...",
"intent": "DefaultTopic",
"topic": "DefaultTopic",
"plan": [
{ "type": "UserInputStep", ... },
{ "type": "SessionInitialStateStep", ... },
{ "type": "EnabledToolsStep", ... },
{ "type": "LLMStep", ... },
...
]
}---
Complete Step Type Taxonomy (13 Types)
For the full execution lifecycle and how these steps map to AgentScript syntax, see agent-execution-lifecycle.md.
Captured from a single agent turn (46 steps total):
| Step Type | Count | Description | Key Data Fields |
|---|---|---|---|
VariableUpdateStep | 28 | Variable state change | variable_updates[].{variable_name, variable_past_value, variable_new_value, variable_change_reason, directive_context} |
BeforeReasoningIterationStep | 3 | Pre-iteration setup | agent_name, action_names[] |
NodeEntryStateStep | 2 | Topic/agent entry | agent_name, directive_context, state_variables{} |
BeforeReasoningStep | 2 | Pre-reasoning setup | agent_name, action_names[] |
EnabledToolsStep | 2 | Available tools list | agent_name, enabled_tools[], directive_context |
LLMStep | 2 | LLM call with full prompt | agent_name, prompt_name, prompt_content, execution_latency, messages_sent[], tools_sent[], response_messages[] |
UserInputStep | 1 | User message | message |
SessionInitialStateStep | 1 | Session init state | variable_values{}, directive_context |
TransitionStep | 1 | Topic transition | from_agent, to_agent, current_state{}, transition_type, transition_mode, directive_context |
FunctionStep | 1 | Flow/Apex action | function.{name, input{}, output{}, errors}, executionLatency |
AfterReasoningStep | 1 | Post-reasoning cleanup | agent_name, action_names[] |
ReasoningStep | 1 | Grounding evaluation | category (GROUNDED/UNGROUNDED), reason |
PlannerResponseStep | 1 | Final response | message, responseType, isContentSafe, safetyScore.category_scores{} |
Common Fields (All Step Types)
Every step has:
{
"startExecutionTime": 1772000000000,
"endExecutionTime": 1772000000000,
"type": "StepTypeName"
}Timestamps are Unix epoch milliseconds.
---
Detailed Step Schemas
LLMStep (Most Data-Rich)
Contains the full LLM prompt, tool definitions, and response — the most valuable step for debugging.
{
"startExecutionTime": 1772000001000,
"endExecutionTime": 1772000002880,
"type": "LLMStep",
"data": {
"agent_name": "Topic Selector",
"prompt_name": "topic_selector_prompt",
"prompt_content": "[full prompt string, 4000+ chars]",
"prompt_response": null,
"execution_latency": 1879
},
"messages_sent": [
{"role": "system", "content": "Topic Selector & Safety Router..."},
{"role": "user", "content": "I need help with my order"},
{"role": "system", "content": "Customer: Jane Doe..."}
],
"tools_sent": [
"go_order_support",
"go_escalation",
"Inappropriate_Content",
"Prompt_Injection",
"Reverse_Engineering"
],
"response_messages": [
{"role": "assistant", "content": "", "tool_invocation": {"name": "go_order_support", "arguments": "{}"}}
]
}FunctionStep (Action Execution)
Shows Flow/Apex invocation with input/output:
{
"startExecutionTime": 1772000003000,
"endExecutionTime": 1772000005675,
"type": "FunctionStep",
"function": {
"name": "Get_Order_Status",
"input": {"orderId": "ORD-12345"},
"output": {"status": "Shipped", "trackingNumber": "1Z999..."},
"errors": null
},
"executionLatency": 2675
}TransitionStep (Topic Routing)
Shows topic-to-topic transitions with full state:
{
"startExecutionTime": 1772000002900,
"endExecutionTime": 1772000002900,
"type": "TransitionStep",
"data": {
"from_agent": "Topic Selector",
"to_agent": "order_support",
"current_state": {"authenticated": true, "customer_name": "Jane Doe", "...": "..."},
"transition_type": "ROUTING",
"transition_mode": "DIRECT",
"directive_context": "on_message"
}
}PlannerResponseStep (Final Output + Safety)
Contains the safety score breakdown:
{
"startExecutionTime": 1772000006800,
"endExecutionTime": 1772000006800,
"type": "PlannerResponseStep",
"message": "I can help you track your order. Let me look that up...",
"responseType": "Inform",
"isContentSafe": true,
"safetyScore": {
"safetyScore": {
"safety_score": 0.99,
"category_scores": {
"toxicity": 0, "hate": 0, "identity": 0, "violence": 0,
"physical": 0, "sexual": 0, "profanity": 0, "biased": 0
}
}
}
}ReasoningStep (Grounding Evaluation)
The agent's self-assessment of response quality:
{
"startExecutionTime": 1772000005600,
"endExecutionTime": 1772000006800,
"type": "ReasoningStep",
"category": "GROUNDED",
"reason": "The response follows the mandatory first question as instructed..."
}---
Real-Time → STDM Field Mapping (Confirmed)
| Real-Time Step Type (13) | Persisted STDM Step Type (5) | Data Loss |
|---|---|---|
UserInputStep | AIAgentInteractionMessage (Input type) | None |
SessionInitialStateStep | Not directly persisted | Full state lost |
NodeEntryStateStep | TOPIC_STEP (partially) | State variables lost |
VariableUpdateStep | PreStepVariableText__c / PostStepVariableText__c | Change reasons lost |
BeforeReasoningStep | Not persisted | Action list lost |
BeforeReasoningIterationStep | Not persisted | Iteration data lost |
EnabledToolsStep | Not persisted | Tool list lost |
LLMStep | LLM_STEP | Full prompt content lost; only input/output summaries kept |
TransitionStep | TOPIC_STEP (partially) | Transition mode/type lost |
FunctionStep | ACTION_STEP | Input/output preserved; latency lost |
AfterReasoningStep | Not persisted | Cleanup data lost |
ReasoningStep | LLM_STEP (ReactValidationPrompt) | Grounding category/reason lost |
PlannerResponseStep | AIAgentInteractionMessage (Output type) | Safety scores lost; content preserved |
Key finding: The real-time trace contains 13 step types vs 5 in STDM. The most valuable debugging data (full LLM prompts, variable change reasons, safety scores, grounding evaluations) is not persisted to Data Cloud.
---
Other Discovered Endpoints
postInteractionSummary (Aura)
Called after trace retrieval. Sends the full agent script + conversation back for analytics:
Descriptor: aura://AuthoringAgentFamilyController/ACTION$postInteractionSummary
Params: {
authoringAgentInteractionSummaryInputRepresentation: {
projectId: "{project_id}",
currentAFScript: "[full agent script JSON]",
...
}
}EvfSdkController/publishEvent (Aura Telemetry)
Client-side observability events tagged COPILOT_O11Y:
Descriptor: aura://EvfSdkController/ACTION$publishEvent
Params: {
event: {
contextualData: {
loggerName: "COPILOT_O11Y",
loggerAppName: "AgentAuthoring:agentAuthoringBuilder"
},
feature: "EINSTEIN_..."
}
}---
Auth Patterns (Confirmed)
| Endpoint | Auth Method | Header |
|---|---|---|
SSE Stream (/messages/stream) | JWT Bearer token | Authorization: Bearer eyJ... |
Aura Trace (/aura + getSimulationPlanTraces) | Browser cookies | X-SFDC-Page-Cache, X-SFDC-Request-Id (no Authorization header) |
Aura Telemetry (/aura + instrumentation) | Browser cookies | Same as above |
Important: The trace endpoint (getSimulationPlanTraces) uses cookie-based auth only. It cannot be called with a standalone JWT token. You must maintain a browser session.---
Capture Methodology
Option 1: CLI Preview Commands (Recommended)
The sf agent preview start/send/end commands provide programmatic access to the same v1.1 trace data without browser automation:
# Start session → send utterance → end and get traces
SESSION_ID=$(sf agent preview start --api-name My_Agent --target-org myOrg --json 2>/dev/null | jq -r '.result.sessionId')
PLAN_ID=$(sf agent preview send --session-id "$SESSION_ID" --api-name My_Agent --utterance "test" --target-org myOrg --json 2>/dev/null | jq -r '.result.messages[-1].planId')
TRACES_PATH=$(sf agent preview end --session-id "$SESSION_ID" --api-name My_Agent --target-org myOrg --json 2>/dev/null | jq -r '.result.tracesPath')
# Analyze trace
jq '.' "$TRACES_PATH/$PLAN_ID.json"For full workflow details, see sf-ai-agentforce-testing Phase F: trace-enriched preview testing. If you use--authoring-bundleinstead of--api-name, remember thatpreview startnow requires an explicit mode flag:--simulate-actionsor--use-live-actions.
Option 2: Manual HAR Export (Fallback)
1. Open DevTools (Cmd+Opt+I) → Network tab 2. Enable: ✅ Preserve log, ✅ Disable cache, Filter: Fetch/XHR 3. Send test messages in Builder 4. Right-click → "Save all as HAR with content" 5. Analyze with jq or the trace analyzer in sf-ai-agentforce-testing
Key Findings
- The trace endpoint (
getSimulationPlanTraces) returns the fullPlanSuccessResponsewith 13 step types - CLI approach (
sf agent preview) saves traces as JSON files automatically — no network interception needed - Trace files contain complete LLM prompts, safety scores, grounding assessments, and action I/O
---
Capture Log
Use this section to track your own capture sessions.
Org: [your org alias]
Agent: [agent API name]
Session ID: [from SSE stream URL]
Plan ID: [from INFORM event planId]
User message: [test utterance]
Total requests: [seen], [captured]
SSE streams: [count]
Step types found: [count]
Steps in trace: [count]Key findings from initial investigation:
- Architecture is Aura-based (Hypothesis A confirmed)
- SSE stream at
api.salesforce.comdelivers messages; trace is fetched separately via Aura - 13 step types in real-time vs 5 in persisted STDM
- Full LLM prompts, tool lists, and safety scores available only in real-time trace
planIdfrom INFORM event links messages to trace- Cookie auth required for trace (no standalone API access)
---
Programmatic Access Comparison
| API | Trace Depth | Step Types | Full Prompts | Safety Scores | Auth | Docs |
|---|---|---|---|---|---|---|
| Builder Trace (this doc) | 13 step types, full state | 13 | ✅ Yes | ✅ Yes | Browser cookies | Internal |
| Data Cloud STDM | 5 step types, summaries | 5 | ❌ No | ❌ No | JWT Bearer | query-patterns.md |
| Testing API | Actions + topic only | N/A | ❌ No | ❌ No | OAuth | Salesforce Docs |
| Agent Runtime API | Messages only — no trace | N/A | ❌ No | ❌ No | OAuth | Salesforce Docs |
---
Related Documents
- Data Model Reference — Complete 24-DMO STDM schema (persisted layer)
- Query Patterns — Data Cloud SQL examples for persisted trace data
- Debugging Sessions — Session timeline reconstruction from STDM
- Auth Setup — JWT Bearer configuration for Data Cloud access
<!-- Parent: sf-ai-agentforce-observability/SKILL.md -->
CLI Reference
Complete command reference for the stdm-extract CLI tool.
Global Options
| Option | Description |
|---|---|
--help | Show help message and exit |
--version | Show version number |
---
Commands
extract
Extract session tracing data for a date range.
stdm-extract extract --org <alias> [options]Options:
| Option | Type | Default | Description |
|---|---|---|---|
--org | String | Required | Salesforce org alias (from sf org list) |
--consumer-key | String | $SF_CONSUMER_KEY | ECA consumer key |
--days | Integer | 7 | Extract last N days |
--since | DateTime | None | Start date (ISO format) |
--until | DateTime | Now | End date (ISO format) |
--agent | String | None | Filter by agent API name (repeatable) |
--output | Path | ./stdm_data | Output directory |
--verbose | Flag | False | Enable verbose logging |
Examples:
# Last 7 days for all agents
stdm-extract extract --org prod
# Specific date range
stdm-extract extract --org prod --since 2026-01-01 --until 2026-01-15
# Filter by agent
stdm-extract extract --org prod --agent Customer_Support_Agent
# Multiple agents
stdm-extract extract --org prod --agent Agent1 --agent Agent2---
extract-tree
Extract complete session tree for specific session IDs.
stdm-extract extract-tree --org <alias> --session-ids <ids> [options]Options:
| Option | Type | Default | Description |
|---|---|---|---|
--org | String | Required | Salesforce org alias |
--consumer-key | String | $SF_CONSUMER_KEY | ECA consumer key |
--session-ids | String | Required | Comma-separated session IDs |
--output | Path | ./stdm_data | Output directory |
--verbose | Flag | False | Enable verbose logging |
Example:
stdm-extract extract-tree --org prod --session-ids "a0x001,a0x002,a0x003"---
extract-incremental
Incremental extraction based on last run timestamp.
stdm-extract extract-incremental --org <alias> [options]Options:
| Option | Type | Default | Description |
|---|---|---|---|
--org | String | Required | Salesforce org alias |
--consumer-key | String | $SF_CONSUMER_KEY | ECA consumer key |
--output | Path | ./stdm_data | Output directory |
--verbose | Flag | False | Enable verbose logging |
Notes:
- Watermark stored at
~/.sf/observability/{org}/watermark.json - First run extracts last 24 hours
- Subsequent runs extract since last watermark
Example:
# First run: extracts last 24 hours
stdm-extract extract-incremental --org prod
# Subsequent runs: extracts only new data
stdm-extract extract-incremental --org prod---
analyze
Generate summary statistics from extracted data.
stdm-extract analyze --data-dir <path> [options]Options:
| Option | Type | Default | Description |
|---|---|---|---|
--data-dir | Path | Required | Directory containing Parquet files |
--format | Choice | table | Output format: table, json, csv |
Example:
stdm-extract analyze --data-dir ./stdm_data --format table---
debug-session
Debug a specific session with full message timeline.
stdm-extract debug-session --data-dir <path> --session-id <id> [options]Options:
| Option | Type | Default | Description |
|---|---|---|---|
--data-dir | Path | Required | Directory containing Parquet files |
--session-id | String | Required | Session ID to debug |
--verbose | Flag | False | Show step details (LLM, actions) |
--output | Path | None | Export timeline to JSON |
Example:
# View timeline in terminal
stdm-extract debug-session --data-dir ./stdm_data --session-id "a0x123"
# Export to JSON with full details
stdm-extract debug-session --data-dir ./stdm_data --session-id "a0x123" \
--verbose --output ./debug/session.json---
topics
Analyze topic routing patterns.
stdm-extract topics --data-dir <path> [options]Options:
| Option | Type | Default | Description |
|---|---|---|---|
--data-dir | Path | Required | Directory containing Parquet files |
--format | Choice | table | Output format: table, json |
Example:
stdm-extract topics --data-dir ./stdm_data---
actions
Analyze action invocation patterns.
stdm-extract actions --data-dir <path> [options]Options:
| Option | Type | Default | Description |
|---|---|---|---|
--data-dir | Path | Required | Directory containing Parquet files |
--agent | String | None | Filter by agent API name |
--format | Choice | table | Output format: table, json |
Example:
stdm-extract actions --data-dir ./stdm_data --agent Customer_Support_Agent---
count
Count records in Data Cloud (quick check without extraction).
stdm-extract count --org <alias> [options]Options:
| Option | Type | Default | Description |
|---|---|---|---|
--org | String | Required | Salesforce org alias |
--consumer-key | String | $SF_CONSUMER_KEY | ECA consumer key |
--dmo | Choice | sessions | DMO to count |
Example:
stdm-extract count --org prod --dmo sessions---
test-auth
Test authentication to Data Cloud.
stdm-extract test-auth --org <alias> [options]Options:
| Option | Type | Default | Description |
|---|---|---|---|
--org | String | Required | Salesforce org alias |
--consumer-key | String | $SF_CONSUMER_KEY | ECA consumer key |
Example:
stdm-extract test-auth --org prod---
Environment Variables
| Variable | Description |
|---|---|
SF_CONSUMER_KEY | Default consumer key for ECA |
SF_JWT_KEY_PATH | Override JWT private key location (default: ~/.sf/jwt/{org}.key) |
---
Exit Codes
| Code | Meaning |
|---|---|
| 0 | Success |
| 1 | General error |
| 2 | Authentication failed |
| 3 | Data not found |
| 4 | Invalid arguments |
---
See Also
- Auth Setup - Setting up JWT authentication
- Polars Cheatsheet - Quick analysis commands
- Troubleshooting - Common issues
<!-- Parent: sf-ai-agentforce-observability/SKILL.md -->
STDM Query Walkthrough - Client Demo
Step-by-step queries for exploring Session Tracing Data Model in Data Cloud Query Console.
Data Model Hierarchy
ssot__AIAgentSession__dlm (Session)
└── ssot__AIAgentInteraction__dlm (Turn)
└── ssot__AIAgentInteractionStep__dlm (LLM/Actions)
└── ssot__AIAgentMoment__dlm (Messages)---
Tab S1: All Sessions
Purpose: Overview of all agent sessions
-- S1: All Sessions (Overview)
SELECT
ssot__Id__c AS SessionID,
ssot__StartTimestamp__c AS StartTime,
ssot__EndTimestamp__c AS EndTime,
ssot__RelatedMessagingSessionId__c AS MessagingSessionID,
ssot__AiAgentSessionEndType__c AS EndType
FROM ssot__AIAgentSession__dlm
ORDER BY ssot__StartTimestamp__c DESC
LIMIT 100📋 Copy a `SessionID` for the next query
---
Tab S2: Interactions for Session
Purpose: All turns/interactions within a session
-- S2: Interactions for Session
SELECT
ssot__Id__c AS InteractionID,
ssot__AiAgentSessionId__c AS SessionID,
ssot__AiAgentInteractionType__c AS InteractionType,
ssot__TopicApiName__c AS Topic,
ssot__StartTimestamp__c AS StartTime,
ssot__EndTimestamp__c AS EndTime
FROM ssot__AIAgentInteraction__dlm
WHERE ssot__AiAgentSessionId__c = '{{PASTE_SESSION_ID_HERE}}'
ORDER BY ssot__StartTimestamp__c📋 Copy an `InteractionID` for S3 and S4
---
Tab S3: Steps for Interaction
Purpose: LLM reasoning and action execution within a turn
-- S3: Steps for Interaction
SELECT
ssot__Id__c AS StepID,
ssot__AiAgentInteractionId__c AS InteractionID,
ssot__AiAgentInteractionStepType__c AS StepType,
ssot__Name__c AS ActionName,
ssot__InputValueText__c AS InputJSON,
ssot__OutputValueText__c AS OutputJSON
FROM ssot__AIAgentInteractionStep__dlm
WHERE ssot__AiAgentInteractionId__c = '{{PASTE_INTERACTION_ID_HERE}}'StepType values: LLM_STEP, ACTION_STEP
---
Tab S4: Messages for Interaction
Purpose: Actual user/agent conversation content
-- S4: Messages for Interaction
SELECT
ssot__Id__c AS MessageID,
ssot__AiAgentInteractionId__c AS InteractionID,
ssot__AiAgentInteractionMessageType__c AS MessageType,
ssot__ContentText__c AS Content,
ssot__MessageSentTimestamp__c AS SentTime
FROM ssot__AIAgentMoment__dlm
WHERE ssot__AiAgentInteractionId__c = '{{PASTE_INTERACTION_ID_HERE}}'
ORDER BY ssot__MessageSentTimestamp__cMessageType values: INPUT (user), OUTPUT (agent)
---
Tab A1: Session Summary by End Type
Purpose: Aggregate sessions by how they ended
-- A1: Session Summary by End Type
SELECT
ssot__AiAgentSessionEndType__c AS EndType,
COUNT(*) AS SessionCount
FROM ssot__AIAgentSession__dlm
GROUP BY ssot__AiAgentSessionEndType__c
ORDER BY SessionCount DESC---
Tab A2: Topic Usage
Purpose: Which topics handle the most turns
-- A2: Topic Usage Analysis
SELECT
ssot__TopicApiName__c AS Topic,
COUNT(*) AS TurnCount
FROM ssot__AIAgentInteraction__dlm
WHERE ssot__AiAgentInteractionType__c = 'TURN'
GROUP BY ssot__TopicApiName__c
ORDER BY TurnCount DESC---
Tab A3: Action Invocations
Purpose: Which actions are called most frequently
-- A3: Action Invocation Frequency
SELECT
ssot__Name__c AS ActionName,
COUNT(*) AS InvocationCount
FROM ssot__AIAgentInteractionStep__dlm
WHERE ssot__AiAgentInteractionStepType__c = 'ACTION_STEP'
GROUP BY ssot__Name__c
ORDER BY InvocationCount DESC---
Demo Flow
1. S1: Run query, pick an interesting session (look for Escalated/Failed in EndType) 2. S2: Paste SessionID, see the conversation turns and topics 3. S3: Paste InteractionID, see LLM reasoning + action I/O 4. S4: Paste InteractionID, read the actual messages 5. A1-A3: Show aggregate insights
---
Field Reference
| Friendly Name | Actual Field |
|---|---|
| SessionID | ssot__Id__c |
| StartTime | ssot__StartTimestamp__c |
| EndTime | ssot__EndTimestamp__c |
| EndType | ssot__AiAgentSessionEndType__c |
| InteractionID | ssot__Id__c |
| Topic | ssot__TopicApiName__c |
| StepType | ssot__AiAgentInteractionStepType__c |
| ActionName | ssot__Name__c |
| MessageType | ssot__AiAgentInteractionMessageType__c |
| Content | ssot__ContentText__c |
Related skills
Forks & variants (1)
Sf Ai Agentforce Observability has 1 known copy in the catalog totaling 32 installs. They canonicalize to this original listing.
- jaganpro - 32 installs
How it compares
Pick sf-ai-agentforce-observability when Salesforce Agentforce session parquet or HAR captures exist; use generic LLM tracing tools for non-Salesforce agents.
FAQ
What does sf-ai-agentforce-observability do?
Agentforce session tracing extraction and analysis. TRIGGER when: user extracts STDM data from Data Cloud, analyzes agent session traces, debugs agent conversations via.
When should I use sf-ai-agentforce-observability?
User asks about sf ai agentforce observability or related SKILL.md workflows.
Is sf-ai-agentforce-observability safe to install?
Review the Security Audits panel on this page before installing in production.