# Project export: Promptetheus This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small. ## Project metadata - Hackathon: UC Berkeley AI Hackathon 2026 - Tagline: Incident response for AI agents: trace runs, catch silent failures, replay the bad step, and turn fixes into regression replays. - Devpost: https://devpost.com/software/promptetheus - GitHub: https://github.com/obro79/promptetheus - Demo: https://promptetheus-console.vercel.app/ - Video: https://www.youtube.com/embed/s41WnOceXRM?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent - Team: 2 GitHub contributor(s) — owenfisher (21 commits), Saksham Adhikari (3 commits) ## Devpost submission (written by the team) ### What it does Promptetheus is an incident response for production AI agents. Traditional monitoring tells teams when a service crashes. Promptetheus tells teams when an agent appears to succeed but silently violates the user’s goal the failure mode dashboards and logs miss entirely. Developers add a lightweight Python SDK to their agentic application. Promptetheus records the run, streams structured events into an incident console in real time, detects likely failures, replays the exact bad step, explains the root cause, packages the fix context, dispatches an autonomous fix, and runs regression checks to prove the agent actually improved. The core loop is: Observe → Detect → Replay → Attribute → Fix → Evaluate → Prevent To prove it generalizes, we instrument three different agents across three modalities and each one fails in a way no traditional monitor would catch: Browser agent (form automation): fills out a job application, every tool call returns, the form submits, and it confidently reports "Application submitted successfully" — while silently leaving a required field empty. Voice agent (real-time speech): mid-conversation the user corrects it, the agent ignores the correction, and still ends the call claiming the task is done. Chat agent (support / refunds): retrieves the correct refund policy, then acts against it — issuing the wrong refund. It had the right evidence and still did the wrong thing. Three modalities, three silent failures, one loop. Each run becomes an incident with evidence, a replayed bad step, and a root-cause attribution — not just a trace. What sets Promptetheus apart from observability tools: it doesn't stop at "here's what happened." It ships the fix. The incident is handed to an autonomous coding agent (Devin) that opens a real GitHub pull request editing the agent's source code. A verification gate — an LLM-as-judge critique plus a regression replay — blocks any fix that doesn't genuinely address the failure, so a wrong fix never ships. On the PR, the previously failing test turns green in CI, and before-and-after regression evidence proves the failure was actually prevented. Two things compound over time. Every verified fix is embedded into Redis vector memory and reused as a warm-start for similar future incidents — a learning flywheel read-only tools can't have. And the healer monitors its own fix quality through Sentry's AI-agent (gen_ai) tracing, catching the same overconfidence in our fixer that Promptetheus exists to catch in agents. The stack: a Python SDK + FastAPI ingestion gateway, Supabase (Postgres + Auth + Storage + RLS) as canonical storage, Redis (vector fix-memory + the live heal timeline), a Next.js incident console, real GitHub PRs, and an agnostic fix-agent layer — with Deepgram powering the voice agent. The goal isn't to show logs. It's to turn an agent failure into an incident with evidence, ownership, a fix path, and proof the failure can't happen again. ### How we built it We designed Promptetheus as a drop-in debugging layer for existing agent stacks. The biggest constraint was adoption. Production teams will not refactor their agents just to add a hackathon tool. So we built the SDK around decorators, lightweight wrappers, typed events, and durable delivery. Developers can install the SDK as a real Python package: Then they can add tracing with minimal code changes: To prove the model generalizes, we instrument three different agents across three modalities — a browser agent, a voice agent, and a chat agent — and run all three through the same incident loop. The system has these core components: Python SDK The SDK captures one trace per user-visible agent task. It emits typed events for user messages, agent messages, tool calls, browser actions, DOM snapshots, screenshots, LLM calls, retrieval, metrics, errors, scores, and final goal checks. Decorator-based instrumentation Promptetheus uses decorators for top-level agent runs, tool calls, and nested spans. This makes the integration backwards compatible because teams can instrument existing code instead of rebuilding around a new framework. Durable local delivery The SDK never crashes the host agent if delivery fails. If HTTP delivery is not configured or the service is unavailable, events spool locally and can be replayed later. That makes the tool safe to add to real agent code. Playwright browser tracing For browser agents, Promptetheus captures clicks, DOM state, screenshots, selected values, warnings, and replay artifacts. This is critical because many agent failures are only obvious visually. Deepgram voice tracing For voice agents, Promptetheus traces the live speech-to-text and text-to-speech turns powered by Deepgram, so spoken corrections the agent ignored become first-class evidence on the timeline — not lost audio. FastAPI ingestion service FastAPI is the trace write gateway. It receives events and artifacts from the SDK, validates the trace state, and streams events into the rest of the system. Supabase storage with workspace isolation Supabase stores trace sessions, incidents, replay metadata, artifacts, and structured evidence. We use the Supabase-backed storage model and RLS-oriented schema to make the hosted version feel like real team infrastructure, not just a local demo. Workspace-filtered event streaming The console receives live trace updates through workspace-filtered streaming. In the hackathon build, this uses server-side streaming from the FastAPI service. In production, this is designed to evolve into Kafka-backed ingestion so high-volume event streams can be processed, clustered, and routed across tenants. Redis vector fix-memory Redis is the learning layer. Every verified fix is embedded and stored in Redis Vector Sets, keyed by the incident's root cause. Before generating a new fix, Promptetheus runs a vector search for the most similar past incident and passes its proven fix to the fix agent as a warm-start — so the system gets better at fixing the more it runs. Redis Streams also carry the live heal-loop timeline the console renders in real time. Orkes Agentspan fix orchestration Orkes Agentspan provides the durable, trackable orchestration layer for the heal loop. It sequences the same diagnose → verify → open-PR steps the in-process loop runs, so every fix run becomes a first-class, replayable execution with the trace, screenshots, replay metadata, memory context, and root-cause summary packaged into a structured task. Devin coding agent (Cognition) Devin handles the actual code-level fix. Instead of stopping at "here is what went wrong," Promptetheus hands the incident brief to Devin, which proposes the fix and opens a real pull request. A verification gate we built — an LLM-as-judge critique plus a regression replay — blocks any fix that does not genuinely address the root cause, so a wrong fix never ships. Sentry AI-agent monitoring Sentry is both our product analogy and a live integration. We instrument the heal loop with Sentry's gen_ai (AI-agent) tracing: every heal run is a gen_ai.invoke_agent transaction and each fix-quality eval is a span — so Promptetheus monitors its own fixer's quality in Sentry, catching the same overconfidence in our fix agent that Promptetheus exists to catch in agents. Sentry also fits the production story for monitoring Promptetheus itself across ingestion errors, console failures, latency spikes, and broken fix-dispatch paths. Regression replay and evals After the fix path runs, Promptetheus checks whether the agent actually improved with before-and-after regression evidence, and the previously failing test turns green in CI on the PR. This prevents fake progress where a patch looks reasonable but does not solve the original failure or introduces a new one. Next.js incident console The console shows sessions, logs, replay, incidents, evals, docs, settings, and the guided demo. Developers can inspect the trace, view replay evidence, understand the root cause, see and track the dispatched fix and its Devin PR, and verify whether the fix improved the agent. The architecture is horizontal by design. Promptetheus can support browser agents, chat agents, voice agents, coding agents, or raw tool-using LLM workflows because the underlying event model is not tied to one agent framework. Tech stack We built Promptetheus with: Python for the SDK and event instrumentation uv and PyPI for simple package installation and distribution FastAPI for the ingestion backend Playwright for browser-agent tracing and replay evidence Deepgram for the real-time voice agent (speech-to-text + text-to-speech) Supabase / Postgres for canonical trace and incident storage Supabase RLS for workspace isolation and multi-tenant foundations Redis (Vector Sets + Streams) for vector fix-memory and the live heal-loop timeline Server-sent events for live trace streaming in the console Orkes Agentspan for durable fix-workflow orchestration Devin (Cognition) for the autonomous coding-agent fix step Sentry for AI-agent (gen_ai) monitoring and the incident-response model Next.js, TypeScript, and Tailwind CSS for the incident console Vercel for frontend deployment Railway for backend deployment GitHub for the open-source repo, real fix PRs, issues, and regression artifacts Kafka-backed ingestion (planned) for production-scale event streaming How we used sponsors We wanted every sponsor integration to make the product stronger, not just appear in the stack. Redis is the fix-memory and learning layer — used well beyond caching. Every verified incident→fix pair is embedded and stored in Redis Vector Sets, and before each new fix we run a vector search for the nearest past fix and reuse it as a warm-start, so the system learns over time. Redis Streams also drive the live heal-loop timeline the console renders. This is the data flywheel a read-only observability tool can't have. Anthropic (Claude) powers Promptetheus two ways. As the product: Claude is our LLM-as-judge verification gate — it scores the agent's before-and-after behavior against the violated goal and approves a fix only when it genuinely flips the failure, not just because the patch looks plausible. As the way we built it: we used Claude Code throughout the hackathon to design the event schema and wire the ingestion and heal loop. Using Claude both to judge whether a fix is real and to build the tool that enforces it is what keeps the loop honest. Sentry is both the clearest product analogy and a live integration. Promptetheus is Sentry for AI-agent behavior: instead of only catching exceptions, it catches false success, ignored warnings, goal mismatches, bad tool usage, and semantic failures. We also instrument our own heal loop with Sentry's gen_ai AI-agent tracing, so the fixer's quality is monitored in Sentry, and Sentry fits the production story for monitoring Promptetheus itself across ingestion errors, console failures, latency spikes, and broken fix-dispatch paths. Orkes coordinates the fix workflow. With Orkes Agentspan, each heal run becomes a durable, trackable execution that sequences the diagnose → verify → open-PR steps and packages the incident — trace, evidence, replay metadata, memory context, and root-cause summary — into a structured fix task. Cognition (Devin) handles the actual fixing step. Promptetheus uses Devin to move from diagnosis to a real code-level fix and pull request, so the workflow does not stop at observability, and our verification gate ensures Devin's fix is regression-checked before it can ship. Deepgram powers the real-time voice agent. Voice failures like an agent ignoring a spoken correction and still claiming success are some of the hardest to catch, and Deepgram's speech-to-text and text-to-speech let us trace those turns as structured evidence rather than lost audio. The result is a real loop: record the failure, stream the events, store the evidence, retrieve the context, orchestrate the fix, apply the fix, evaluate the result, and prevent the bug from recurring. ### Challenges we ran into The hardest part was turning raw traces into a useful incident instead of building another log viewer. Agent runs produce a lot of noise: messages, tool calls, browser actions, screenshots, DOM state, memory, retrieval context, model outputs, and final responses. Those details are only useful if they are organized around the failure. We had to decide what evidence actually matters and how to present it so a developer can quickly understand the bug. Another challenge was making the SDK safe and backwards compatible. If a debugging tool requires a major refactor, teams will not install it until after something is already on fire. That pushed us toward decorators, lightweight wrappers, local spooling, and a minimal event contract. We also had to think carefully about production ingestion. The hackathon version uses FastAPI and live event streaming so the demo is real end to end. But the product is designed to grow into Kafka-backed ingestion, multi-tenant workspaces, event fanout, clustering, alerting, and high-volume agent monitoring. The final hard part was closing the loop from detection to verified fix. It is easy to generate a plausible explanation. It is harder to generate a fix path. It is much harder to prove that the fix actually improves the agent. That is why Promptetheus includes Orkes-orchestrated fix dispatch, a Devin coding-agent handoff, a verification gate, and before-and-after regression evidence — and why a wrong fix is blocked rather than shipped. ### Accomplishments we're proud of We are proud that Promptetheus already feels like a real developer tool, not just a hackathon prototype. We were able to get 24 stars and 2 real users that used promptetheus to help debug and fix their apps during the hackathon period! We registered and shipped a real Python package that other developers can install with: That matters because the product only works if it is easy to adopt. We did not want a demo that only runs on our machine. We wanted something other teams could actually try. We are also proud of how backwards compatible the integration is. The decorator-based SDK lets developers add tracing with minimal edits instead of rewriting their agent stack. More importantly, we got buy-in from other teams who immediately understood the debugging pain. Some teams used Promptetheus-style tracing to help debug critical bugs in their own agentic applications. That was the strongest validation for us. Promptetheus was not just useful for our demo. It helped real builders find real failures. During the hackathon, we built the SDK, PyPI package, decorator integration, typed event schema, browser and voice tracing flows, Redis-backed vector fix-memory, FastAPI ingestion backend, Supabase storage layer, workspace-isolated service model, event streaming console, Orkes Agentspan fix orchestration, Devin fix path with a verification gate, Sentry AI-agent monitoring, regression replay loop, and three concrete silent-failure demos end to end — across browser, voice, and chat agents. Most importantly, we built something we would actually want to use while building and debugging production AI agents. ### What we learned We learned that agent reliability is not just about catching exceptions. Many of the most important agent bugs are semantic failures. The agent used the wrong tool. It clicked the wrong button. It ignored a spoken correction. It left a required field empty. It used stale context. It retrieved the right policy and still acted against it. It gave a confident final answer even though the user's goal was not satisfied. Those failures need evidence, replay, memory context, root-cause analysis, fix ownership, and evals. Logs alone are not enough. We also learned that the fix step needs verification. A coding agent can generate a patch, but that does not mean the agent is actually better. The real workflow needs before-and-after regression evidence so teams can see whether the fix improved task success, reduced regressions, and solved the original failure — which is exactly why our loop blocks an unverified fix instead of shipping it. ### What's next Next, we want to turn Promptetheus into a complete hosted workflow for teams building agents in production. Our roadmap includes: Team workspaces Production ingestion with Kafka-backed event streaming Multi-tenant project isolation Incident clustering across repeated failures Sentry-style alerting for semantic agent failures Better browser and voice replay Stronger Redis-backed vector memory and similar-fix retrieval More advanced Orkes Agentspan orchestration for fix workflows Deeper Devin-powered code remediation Before-and-after regression eval suites GitHub issue and PR generation across more repos Support for more agent frameworks More decorators and adapters for common agent stacks Hosted dashboards for teams managing many deployed agents The long-term vision is for Promptetheus to become the default debugging and incident-response layer for production AI agents. When an agent fails silently, developers should not have to dig through scattered logs, screenshots, tool outputs, memory state, retrieval results, and user complaints. Promptetheus should give them the full loop: observe the run, detect the failure, replay the bad step, attribute the cause, generate the fix, evaluate the improvement, and prevent the bug from happening again. ## README (from the GitHub repository) # promptetheus Promptetheus is debugging infrastructure for AI agents: a Python SDK, local replay tooling, hosted trace delivery, and MCP evidence access for coding agents that need to fix failing agent runs. ## What You Get - One trace per user-visible agent task. - Decorators for top-level agent runs, tool calls, and nested spans. - Typed events for user messages, agent messages, tool calls, browser actions, DOM snapshots, screenshots, LLM calls, retrieval, metrics, errors, scores, and final goal checks. - Durable delivery that never crashes the host agent. If HTTP delivery is not configured or fails, events spool locally and can be replayed later. - Local CLI tools for doctor checks, spool inspection, session replay, diffing, and failure fingerprints. - Hosted MCP config snippets for read-only incident evidence scoped to a workspace and Supabase project. ## Install For a normal project, install from PyPI: ```bash pip install promptetheus promptetheus version ``` Create or configure a hosted project key: ```bash export PROMPTETHEUS_CONSOLE_TOKEN=... promptetheus init \ --workspace-name "Acme" \ --project-name "Browser Agent" \ --write-env .env source .env promptetheus doctor ``` For local self-hosted development: ```bash promptetheus init \ --api-url http://127.0.0.1:4318 \ --console-token pt_console_token \ --write-env .env source .env ``` For contributor work from this repository: ```bash pip install -e packages/promptetheus promptetheus version ``` With `transport="auto"`, the SDK sends to the configured API when `PROMPTETHEUS_API_KEY` is present. Without a key, it writes to the local spool so the instrumented agent keeps running. ## Observe With Decorators Use decorators when you want instrumentation to sit directly on agent and tool functions: ```python import promptetheus as pt @pt.tool def search_calendar(day: str) -> list[str]: return ["Tuesday 2pm", "Tuesday 3pm"] @pt.traced("choose-slot") def choose_slot(slots: list[str]) -> str: return "Wednesday 2pm" @pt.observe( agent="calendar-agent", user_goal="Book Tuesday at 2pm", transport="auto", # use "spool" to force local JSONL while trying this ) def run_agent(goal: str) -> str: pt.current().user_message(goal) slots = search_calendar("Tuesday") selected = choose_slot(slots) pt.current().agent_message(f"Booked {selected}") pt.current().goal_check( False, mismatches=["selected Wednesday, not Tuesday"], ) return selected run_agent("Book Tuesday at 2pm") ``` What each decorator does: - `@pt.observe(...)` starts one trace/session around the top-level run. - `@pt.tool` records `tool_call` and `tool_result` events inside the current session. - `@pt.traced("name")` adds a nested span to the replay tree without starting a separate session. - `pt.current()` returns the active session so the agent can record user messages, agent messages, goal checks, errors, metrics, and other events. `goal_check(False)` is visible in replay, fingerprints, and tail sampling. If a failed goal should also make the process fail, record the goal check and then raise an exception so the terminal `session_end` status is `failed`: ```python if not selected.startswith("Tuesday"): pt.current().goal_check(False, mismatches=["selected Wednesday"]) raise RuntimeError("agent selected the wrong day") ``` ## What You Can See When no API key is configured, `transport="auto"` writes local JSONL. While learning, you can also pass `transport="spool"` to force local output. After a local or spooled run, list sessions: ```bash promptetheus sessions ``` Example output: ```text 01KVMZ4T7V2SN61ZWG1XTDBK47: 11 event(s) ``` Replay the timeline: ```bash promptetheus replay 01KVMZ4T7V2SN61ZWG1XTDBK47 ``` Example output: ```text [0] state_change name='session_started' [1] tool_call tool_name='run_agent' [2] user_message content='Book Tuesday at 2pm' [3] tool_call tool_name='search_calendar' [4] tool_result call_id='190a6438979141f5ac11b2e1b2ee29a0' [5] state_change name='span_start' [6] state_change name='span_end' [7] agent_message content='Booked Wednesday 2pm' [8] goal_check passed=False [9] tool_result call_id='a78566297e0a4a309d5ce44cefe0d836' [10] session_end status='completed' ``` Replay the run tree: ```bash promptetheus replay 01KVMZ4T7V2SN61ZWG1XTDBK47 --tree ``` Example output: ```text [0] state_change name='session_started' [1] tool_call tool_name='run_agent' [2] user_message content='Book Tuesday at 2pm' [3] tool_call tool_name='search_calendar' [4] tool_result call_id='190a6438979141f5ac11b2e1b2ee29a0' [7] agent_message content='Booked Wednesday 2pm' [8] goal_check passed=False [9] tool_result call_id='a78566297e0a4a309d5ce44cefe0d836' [10] session_end status='completed' choose-slot span=span_163a8380174647e98bfe1f3fff9e15b9 duration_ms=0.0 ``` Generate a failure fingerprint: ```bash promptetheus fingerprint 01KVMZ4T7V2SN61ZWG1XTDBK47 ``` Example output: ```text 8ae0f41220d0 goal mismatch: selected wednesday, not tuesday - goal:selected wednesday, not tuesday ``` Inspect the local delivery spool: ```bash promptetheus spool list ``` Example output: ```text Spool: .promptetheus/spool pending : 11 event(s) across 1 session file(s), 4082 bytes dead : 0 event(s) across 0 file(s), 0 bytes 01KVMZ4T7V2SN61ZWG1XTDBK47: 11 pending ``` The raw spool is JSONL. Each line is an event envelope: ```json { "type": "tool_call", "session_id": "01KVMZ4T7V2SN61ZWG1XTDBK47", "seq": 1, "idempotency_key": "01KVMZ4T7V2SN61ZWG1XTDBK47:29c5eff0:1", "payload": { "tool_name": "run_agent", "call_id": "a78566297e0a4a309d5ce44cefe0d836", "arguments": { "args": "('Book Tuesday at 2pm',)", "kwargs": "{}" } } } ``` ## Manual Trace API Use `pt.trace.start(...)` when you control the run boundary and want explicit event calls instead of decorators: ```python import promptetheus as pt with pt.trace.start( agent="demo-agent", user_goal="Book a meeting for Tuesday", transport="auto", ) as session: session.user_message("Please book the small room for Tuesday at 2pm") session.tool_call("calendar.search", {"day": "Tuesday"}, call_id="calendar-1") session.tool_result("calendar-1", result={"available": ["2pm", "3pm"]}) session.agent_message("Booking confirmed for Wednesday at 2pm") session.goal_check(False, mismatches=["booked Wednesday, not Tuesday"]) # session_end is emitted automatically; transport flush runs on exit ``` ## Public SDK API The package exposes these primary entry points: ```python import promptetheus as pt pt.trace.start(...) pt.start(...) pt.observe(...) pt.tool pt.traced(...) pt.current() pt.Session pt.AsyncSession pt.AgentRuntime ``` Common session helpers: ```python session.user_message("Book Tuesday at 2pm Pacific") session.agent_message("I found availability") session.tool_call("browser.click", {"selector": "#checkout"}, call_id="click-1") session.tool_result("click-1", result={"ok": True}) session.retrieval("refund policy", documents=[{"id": "doc-1", "score": 0.91}]) session.browser_action("click", "#checkout", url=page.url) session.dom_snapshot(page.url, visible_text, selected_values={"day": "Tuesday"}) session.screenshot(page.screenshot()) session.replay_artifact("trace.webm", artifact_type="screen_recording", event_time_map={}) session.llm_call("gpt-5", input_tokens=100, output_tokens=40, latency_ms=900) session.score("goal_match", 0.2, comment="Selected the wrong day") session.metric("steps", 12, unit="count") session.error(RuntimeError("calendar API timeout"), handled=True) session.goal_check(False, mismatches=["selected Wednesday"]) session.end("failed") session.flush(timeout=2) ``` Every helper writes a schema-valid event envelope with `type`, `session_id`, `timestamp`, `seq`, `idempotency_key`, and `payload`. Use `metadata` for safe, low-cardinality context. Do not put raw secrets, cookies, tokens, or [README truncated for size] ## Detected evidence (automated analysis) Indexed codebase: 120 recognized source files, 1091 KB. - Anthropic (technology) — detected in the code - CrewAI (technology) — detected in the code - FastAPI (technology) — detected in the code - OpenAI (technology) — detected in the code - Python (language) — detected in the code - Next.js (technology) — claimed on Devpost, not found in the code - Redis (technology) — claimed on Devpost, not found in the code - Supabase (technology) — claimed on Devpost, not found in the code - Tailwind CSS (technology) — claimed on Devpost, not found in the code - TypeScript (language) — claimed on Devpost, not found in the code - Vercel (technology) — claimed on Devpost, not found in the code ## Codebase structure (from repository index) ### Files (120 of 125) ``` .agent/skills/promptetheus/SKILL.md .github/workflows/publish.yml .gitignore conftest.py docs/architecture/adr/0001-decorator-first-sdk.md docs/architecture/adr/0002-sdk-contract-hardening.md docs/architecture/adr/0003-agent-runtime-redis-coordination.md docs/architecture/components.md docs/architecture/implementation-plan.md docs/architecture/pypi-setup.md docs/architecture/staged-scope.md docs/architecture/technical-architecture.md docs/archive/archive-start-full.md docs/build-plan.md docs/demo-data-plan.md docs/demo-plan.md docs/linear-execution-plan.md docs/mcp.md docs/product-strategy.md docs/README.md docs/reference/reference-examples.md docs/research/competitive-landscape.md docs/reviews/office-hours-ceo-review.md docs/sdk-architecture.md docs/sdk/adapter-parity-matrix.md docs/sdk/http-level1.md examples/langchain_minimal.py LICENSE packages/promptetheus/promptetheus/__init__.py packages/promptetheus/promptetheus/adapters/__init__.py packages/promptetheus/promptetheus/adapters/_base.py packages/promptetheus/promptetheus/adapters/anthropic.py packages/promptetheus/promptetheus/adapters/autogen.py packages/promptetheus/promptetheus/adapters/crewai.py packages/promptetheus/promptetheus/adapters/dspy.py packages/promptetheus/promptetheus/adapters/haystack.py packages/promptetheus/promptetheus/adapters/langchain.py packages/promptetheus/promptetheus/adapters/langgraph.py packages/promptetheus/promptetheus/adapters/litellm.py packages/promptetheus/promptetheus/adapters/llamaindex.py packages/promptetheus/promptetheus/adapters/openai.py packages/promptetheus/promptetheus/adapters/otel.py packages/promptetheus/promptetheus/adapters/playwright.py packages/promptetheus/promptetheus/adapters/pydantic_ai.py packages/promptetheus/promptetheus/agent_runtime.py packages/promptetheus/promptetheus/cli.py packages/promptetheus/promptetheus/config.py packages/promptetheus/promptetheus/cost.py packages/promptetheus/promptetheus/exporters/__init__.py packages/promptetheus/promptetheus/exporters/otlp.py packages/promptetheus/promptetheus/fingerprint.py packages/promptetheus/promptetheus/propagation.py packages/promptetheus/promptetheus/py.typed packages/promptetheus/promptetheus/redaction.py packages/promptetheus/promptetheus/regression.py packages/promptetheus/promptetheus/sampling.py packages/promptetheus/promptetheus/schema.py packages/promptetheus/promptetheus/server/__init__.py packages/promptetheus/promptetheus/server/mcp.py packages/promptetheus/promptetheus/session_async.py packages/promptetheus/promptetheus/session.py packages/promptetheus/promptetheus/skills/promptetheus/SKILL.md packages/promptetheus/promptetheus/testing.py packages/promptetheus/promptetheus/trace_tree.py packages/promptetheus/promptetheus/trace.py packages/promptetheus/promptetheus/transport/__init__.py packages/promptetheus/promptetheus/transport/async_http.py packages/promptetheus/promptetheus/transport/durable.py packages/promptetheus/promptetheus/transport/http.py packages/promptetheus/promptetheus/transport/local.py packages/promptetheus/pyproject.toml packages/promptetheus/README.md packages/promptetheus/uv.lock README.md start.md tests/adapters/__init__.py tests/adapters/conftest.py tests/adapters/test_anthropic.py tests/adapters/test_autogen.py tests/adapters/test_crewai.py tests/adapters/test_dspy.py tests/adapters/test_haystack.py tests/adapters/test_langchain_adapter_parity.py tests/adapters/test_langchain_driven.py tests/adapters/test_langchain.py tests/adapters/test_langgraph_driven.py tests/adapters/test_langgraph.py tests/adapters/test_litellm.py tests/adapters/test_llamaindex.py tests/adapters/test_openai_streaming_cost.py tests/adapters/test_openai.py tests/adapters/test_otel.py tests/adapters/test_playwright.py tests/adapters/test_pydantic_ai.py tests/cli/__init__.py tests/cli/test_cli.py tests/cli/test_diff_replay.py tests/schema/test_schema.py tests/sdk/test_async.py tests/sdk/test_config.py tests/sdk/test_cost.py tests/sdk/test_fingerprint.py tests/sdk/test_instrumentation_primitives.py tests/sdk/test_integration_features.py tests/sdk/test_observe.py tests/sdk/test_otlp_exporter.py tests/sdk/test_propagation.py tests/sdk/test_redaction.py tests/sdk/test_regression_diff.py tests/sdk/test_sampling_privacy.py tests/sdk/test_sampling.py tests/sdk/test_schema_properties.py tests/sdk/test_session_safety.py tests/sdk/test_spans.py tests/sdk/test_tail_policy.py tests/sdk/test_tail_sampling.py tests/sdk/test_testing_utils.py tests/sdk/test_trace_tree.py tests/server/test_mcp.py tests/transport/test_agent_runtime_http.py [5 more files omitted for size] ``` ### Dependencies - packages/promptetheus/pyproject.toml: anthropic@>=0.25, autogen-agentchat@>=0.2, crewai@>=0.40, cryptography@>=42, cryptography@>=42, dspy-ai@>=2.4, fastapi@>=0.110, fastapi@>=0.110, haystack-ai@>=2.0, httpx@>=0.27, httpx@>=0.27, httpx@>=0.27, hypothesis@>=6.0, langchain-core@>=0.2, langchain-core@>=0.2, langgraph@>=0.2, litellm@>=1.40, llama-index-core@>=0.10, mcp@>=1.0, mypy@>=1.8, openai@>=1.0, opentelemetry-api@>=1.20, opentelemetry-exporter-otlp-proto-http@>=1.20, opentelemetry-sdk@>=1.20, opentelemetry-sdk@>=1.20, playwright@>=1.40, psycopg[binary]@>=3.2, psycopg[binary]@>=3.2, pydantic-ai@>=0.0.13, PyJWT@>=2.8, PyJWT@>=2.8, pytest@>=8.0, tiktoken@>=0.7, uvicorn@>=0.29, uvicorn@>=0.29 ### Recent commits (newest first) - Merge pull request #7 from Tar-ive/saksham-skill - Merge branch 'main' into saksham-skill - Add Promptetheus SDK skill at .agent/skills/promptetheus/SKILL.md - Remove README setup GIF - Add root onboarding README - Improve README decorator onboarding - Add SDK onboarding README - Bump SDK to 2.0.1 - Update bundled skill onboarding guidance - Bump SDK to 2.0.0 - Add CLI project bootstrap - Reserve promptetheus on PyPI with trusted publishing scaffold. - Rename MCP supabase tools/endpoints to promptetheus - Bump SDK to 1.0.0 - Document self-host SDK endpoint override - Merge pull request #2 from Tar-ive/saksham-post-tests - Raise hosted SDK HTTP timeout - Bump promptetheus SDK version to 0.1.0 - Default SDK to hosted Promptetheus API - Add Promptetheus Codex skill ## Key source files (fetched from GitHub, selected and truncated for size) ### start.md ```markdown # Promptetheus Promptetheus is debugging infrastructure for AI agents. It is not a LangChain or LangGraph replacement. It is a Python-first SDK, local replay console, and hosted team workspace that instruments whatever agent stack developers already use, captures rich traces, detects likely failures, replays the exact bad step, and packages the fix for a coding agent. ## Core Loop 1. Observe the agent run. 2. Detect suspicious behavior. 3. Replay the exact session. 4. Attribute the critical failure step. 5. Generate a fix bundle. 6. Hand the fix to a coding agent. 7. Replay regression cases to prove the issue is prevented. ## Product Model Promptetheus has two modes: - **Local/open-source:** the SDK writes traces and replay artifacts to local `.promptetheus/` files, and `promptetheus dev` serves a local replay console. - **Cloud/team:** the SDK sends authenticated events to Promptetheus Cloud, where teams get shared trace storage, incident clustering, alerts, repo integrations, fix-agent PRs, regression replay, RBAC, audit logs, retention controls, PII redaction, and Slack digests. ## Flagship Demo The hackathon demo uses a browser agent because browser failures are visual, traceable, and painful. The demo shows a browser agent booking a demo for Tuesday at 2pm Pacific. The agent selects 2:00 AM, ignores a timezone warning, and claims success. Promptetheus records the screen, streams trace events, detects the goal mismatch, replays the failure, generates a fix brief, and shows a PR preview plus regression replay. ## Docs - [Docs Index](docs/README.md) - [Product Strategy](docs/product-strategy.md) - [Demo Plan](docs/demo-plan.md) - [SDK Architecture](docs/sdk-architecture.md) - [Technical Architecture](docs/architecture/technical-architecture.md) - [Components](docs/architecture/components.md) - [Implementation Plan](docs/architecture/implementation-plan.md) - [Staged Scope](docs/architecture/staged-scope.md) - [Build Plan](docs/build-plan.md) - [Demo Data Plan](docs/demo-data-plan.md) - [Linear Execution Plan](docs/linear-execution-plan.md) ## Current Decision Build the hackathon submission as: - `promptetheus` Python SDK - `promptetheus dev` - Local `.promptetheus/` trace and artifact store - Browser-agent / Playwright adapter - Side-by-side demo console - Screen-recording replay artifact - Failure detector and critical-step attribution - Fix-agent PR handoff - Before/after regression replay ## Business Shape Open source gets adoption: - Python SDK - Local replay console - Basic failure detectors - Browser-agent adapter Promptetheus Cloud is the paid product: - Team workspaces - Production trace storage - Search across sessions - Incident clustering over time - Alerts when agent failures spike - CI regression replay - GitHub/Linear/Jira integrations - Connected repo onboarding - Agent-generated fix plans and PRs - RBAC, audit logs, retention controls - PII redaction - SOC2-friendly deployment story - Slack incident digests ``` ### docs/mcp.md ```markdown # Promptetheus MCP Promptetheus MCP exposes hosted incident and trace evidence to coding agents. The hosted server is pull-based: the client connects to Promptetheus, then tools read evidence for the configured Promptetheus project on demand. ## Install Config Generate a client-specific config snippet: ```bash promptetheus mcp install \ --client codex \ --workspace acme \ --project-ref abcdefghijklmnopqrst ``` Supported clients are `codex`, `claude`, and `cursor`. The command prints: - the hosted MCP URL for the workspace/project pair - a stdio bridge config using `npx -y mcp-remote ` - the Promptetheus access scope for the server It does not write global client config files. Paste the printed snippet into the client or workspace-local config you choose to manage. ## Project Scope The hosted MCP server should use read-only, project-scoped access by default. A generated URL includes the Promptetheus workspace and `project_ref`, so evidence reads are scoped to that project. The SDK and MCP client config must not receive database service-role keys. ## Local Stdio Server The existing local command is unchanged: ```bash promptetheus mcp ``` That path boots the local stdio MCP server when the optional `promptetheus[mcp]` dependencies and server implementation are available. ``` ### packages/promptetheus/pyproject.toml ``` [build-system] requires = ["hatchling"] build-backend = "hatchling.build" [project] name = "promptetheus" version = "2.0.1" description = "Debugging infrastructure for AI agents — traces, replay, failure detection, and fix handoff." readme = "README.md" license = "MIT" requires-python = ">=3.12" authors = [{ name = "Owen Fisher", email = "owenfisher46@gmail.com" }] keywords = ["ai", "agents", "observability", "debugging", "tracing", "playwright"] classifiers = [ "Development Status :: 5 - Production/Stable", "Intended Audience :: Developers", "License :: OSI Approved :: MIT License", "Programming Language :: Python :: 3", "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", "Programming Language :: Python :: 3.14", "Topic :: Software Development :: Debuggers", ] dependencies = [] [project.optional-dependencies] server = [ "fastapi>=0.110", "cryptography>=42", "httpx>=0.27", "PyJWT>=2.8", "uvicorn>=0.29", "psycopg[binary]>=3.2", ] mcp = [ "mcp>=1.0", "httpx>=0.27", ] dev = [ "fastapi>=0.110", "cryptography>=42", "PyJWT>=2.8", "uvicorn>=0.29", "httpx>=0.27", "psycopg[binary]>=3.2", "pytest>=8.0", "hypothesis>=6.0", "mypy>=1.8", ] playwright = [ "playwright>=1.40", ] openai = [ "openai>=1.0", ] anthropic = [ "anthropic>=0.25", ] langchain = [ "langchain-core>=0.2", ] llamaindex = [ "llama-index-core>=0.10", ] crewai = [ "crewai>=0.40", ] otel = [ "opentelemetry-api>=1.20", "opentelemetry-sdk>=1.20", ] otlp = [ "opentelemetry-sdk>=1.20", "opentelemetry-exporter-otlp-proto-http>=1.20", ] langgraph = [ "langgraph>=0.2", "langchain-core>=0.2", ] litellm = [ "litellm>=1.40", ] autogen = [ "autogen-agentchat>=0.2", ] dspy = [ "dspy-ai>=2.4", ] haystack = [ "haystack-ai>=2.0", ] pydantic-ai = [ "pydantic-ai>=0.0.13", ] tiktoken = [ "tiktoken>=0.7", ] [project.urls] Homepage = "https://github.com/obro79/promptetheus" Repository = "https://github.com/obro79/promptetheus" Documentation = "https://github.com/obro79/promptetheus#readme" [project.scripts] promptetheus = "promptetheus.cli:main" [tool.hatch.build.targets.wheel] packages = ["promptetheus"] # py.typed lives under the package dir so it is included in the wheel by # default; listed explicitly here as a guard against future include filters # stripping it (PEP 561 typed-package marker must ship). force-include = { "promptetheus/py.typed" = "promptetheus/py.typed" } [tool.mypy] # Strict-ish baseline for the SDK package. The whole package is type-hinted and # ships py.typed, so untyped defs are a regression and unused ignores are noise. python_version = "3.12" files = ["promptetheus"] disallow_untyped_defs = true warn_unused_ignores = true warn_redundant_casts = true warn_unused_configs = true no_implicit_optional = true [[tool.mypy.overrides]] # Optional integration adapters import third-party libs that may be absent in a # bare dev env; do not fail type-checking on their missing stubs. module = [ "playwright.*", "openai.*", "anthropic.*", "langchain_core.*", "llama_index.*", "crewai.*", "opentelemetry.*", "mcp.*", "httpx", "jwt", "psycopg", "psycopg.*", "cryptography.*", "tiktoken", "tiktoken.*", ] ignore_missing_imports = true [[tool.mypy.overrides]] # The per-event TypedDicts intentionally narrow the shared "type" field to a # Literal to form a discriminated union; mypy disallows TypedDict field override # (misc), which is a known limitation here, not a real defect. module = ["promptetheus.schema"] disable_error_code = ["misc"] [[tool.mypy.overrides]] # Integration adapters are thin wrappers over untyped third-party SDKs and use # dynamic lazy subclassing (callback handlers built at call time). Strict typing # fights that dynamism with no safety benefit, so they are checked leniently. module = ["promptetheus.adapters.*"] ignore_errors = true ``` ### packages/promptetheus/promptetheus/cli.py ```python """CLI entry point for Promptetheus. Commands: promptetheus dev Boot the local FastAPI ingestion gateway on :4318. promptetheus version Print the installed version. promptetheus doctor Show resolved config, server reachability, spool backlog. promptetheus spool ... Inspect / replay / purge the local delivery spool. The spool commands operate on the durable transport's local buffer (.promptetheus/spool/.jsonl for pending deliveries and dead-letter/.jsonl for permanently-rejected events). They never crash with a traceback: missing dirs and unconfigured endpoints produce clear messages and a nonzero exit where appropriate. """ from __future__ import annotations import argparse import json import os from pathlib import Path from typing import Any from urllib import error, request from urllib.parse import quote # Host/port for the local FastAPI ingestion gateway (see CLAUDE.md "Ports"). DEV_HOST = "0.0.0.0" DEV_PORT = 4318 DEFAULT_SPOOL_DIR = ".promptetheus/spool" DEFAULT_MCP_BASE_URL = "https://mcp.promptetheus.dev/promptetheus" _DEAD_LETTER_DIR = "dead-letter" def _run_dev() -> None: """Boot the FastAPI ingestion gateway on :4318 via uvicorn. Never raises: if uvicorn (or the server app) cannot be imported, print clear guidance instead of crashing, so promptetheus dev always exits cleanly. """ try: import uvicorn except ImportError: print("promptetheus dev needs uvicorn to boot the FastAPI ingestion gateway.") print( "Install the server dependencies, e.g. pip install 'promptetheus[server]'." ) print(f"Once installed, the gateway listens on http://{DEV_HOST}:{DEV_PORT}") return try: from .server.app import create_app except Exception as exc: # pragma: no cover - defensive: never crash the CLI print("promptetheus dev could not import the FastAPI server app.") print(f"Reason: {exc}") print( "Install the server dependencies, e.g. pip install 'promptetheus[server]'." ) return app = create_app() print(f"Starting Promptetheus ingestion gateway on http://{DEV_HOST}:{DEV_PORT}") uvicorn.run(app, host=DEV_HOST, port=DEV_PORT) def _run_mcp() -> None: """Boot the incident-context MCP server over stdio. Never raises: a missing 'mcp' extra or an unset PROMPTETHEUS_API_KEY produce clear guidance instead of a traceback, so promptetheus mcp always exits cleanly. """ try: from .server.mcp import run as run_mcp except Exception as exc: # pragma: no cover - defensive: never crash the CLI print("promptetheus mcp could not import the MCP server module.") print(f"Reason: {exc}") print("Install the MCP dependencies, e.g. pip install 'promptetheus[mcp]'.") return try: run_mcp() except RuntimeError as exc: # Missing 'mcp' extra or missing PROMPTETHEUS_API_KEY surface here. print(str(exc)) def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(prog="promptetheus") subparsers = parser.add_subparsers(dest="command") subparsers.add_parser( "dev", help="Boot the local FastAPI ingestion gateway on :4318" ) mcp_p = subparsers.add_parser( "mcp", help="Boot the incident-context MCP server over stdio" ) mcp_sub = mcp_p.add_subparsers(dest="mcp_command") mcp_install_p = mcp_sub.add_parser( "install", help="Print hosted Promptetheus MCP client config for a Promptetheus project", ) mcp_install_p.add_argument( "--client", choices=("codex", "claude", "cursor"), required=True, help="MCP client config format to print", ) mcp_install_p.add_argument( "--workspace", required=True, help="Promptetheus workspace slug or id", ) mcp_install_p.add_argument( "--project-ref", required=True, help="Promptetheus project ref to scope evidence reads", ) mcp_install_p.add_argument( "--server-name", default="promptetheus", help="MCP server name in the generated client config", ) mcp_install_p.add_argument( "--hosted-url", default=DEFAULT_MCP_BASE_URL, help=f"hosted MCP base URL (default {DEFAULT_MCP_BASE_URL})", ) subparsers.add_parser("version", help="Print the installed Promptetheus version") init_p = subparsers.add_parser( "init", help="Bootstrap a Promptetheus project and print a generated API key", ) init_p.add_argument( "--api-url", default=None, help="Promptetheus API URL (default: hosted API, or PROMPTETHEUS_API_URL)", ) init_p.add_argument( "--console-token", default=None, help="Console auth token (default: PROMPTETHEUS_CONSOLE_TOKEN)", ) init_p.add_argument( "--workspace-name", default="Promptetheus Workspace", help="Workspace name to create or reuse", ) init_p.add_argument( "--project-name", default="Default Project", help="Project name to create or reuse", ) init_p.add_argument( "--agent-name", default=None, help="Optional first agent name to associate with the project", ) init_p.add_argument( "--write-env", nargs="?", const=".env", default=None, help="Write PROMPTETHEUS_API_KEY and PROMPTETHEUS_API_URL to an env file", ) init_p.add_argument( "--write-config", action="store_true", help="Write api_key and api_url to ~/.promptetheus/config.toml", ) subparsers.add_parser( "doctor", help="Show resolved config, server reachability, spool backlog" ) spool = subparsers.add_parser( "spool", help="Inspect/replay/purge the local delivery spool" ) spool.add_argument( [truncated — 27218 more characters] ``` ### conftest.py ```python """Pytest bootstrap: make the ``promptetheus`` package importable. The package lives under ``packages/promptetheus/`` and is intended to be used via an editable install. In environments where the editable ``.pth`` is not applied (some uv-managed venvs do not process bare-path ``.pth`` files at interpreter startup), tests would fail to import ``promptetheus``. Prepending the package root here makes the whole suite resolve regardless of install state. It is a no-op when the package is already importable. """ from __future__ import annotations import sys from pathlib import Path _PACKAGE_ROOT = Path(__file__).resolve().parent / "packages" / "promptetheus" if _PACKAGE_ROOT.is_dir(): path_str = str(_PACKAGE_ROOT) if path_str not in sys.path: sys.path.insert(0, path_str) ``` ### examples/langchain_minimal.py ```python #!/usr/bin/env python3 """Minimal LangChain callback example — produces a Promptetheus session.""" from __future__ import annotations import sys import types import uuid def _install_fake_langchain() -> None: class BaseCallbackHandler: def __init__(self, *args, **kwargs): pass root = types.ModuleType("langchain_core") callbacks = types.ModuleType("langchain_core.callbacks") callbacks.BaseCallbackHandler = BaseCallbackHandler root.callbacks = callbacks sys.modules["langchain_core"] = root sys.modules["langchain_core.callbacks"] = callbacks def main() -> None: _install_fake_langchain() from promptetheus.adapters.langchain import PromptetheusCallbackHandler from promptetheus.trace import start transport_events: list[dict] = [] class CaptureTransport: def create_trace(self, metadata): pass def send_event(self, event): transport_events.append(dict(event)) def flush(self, timeout=None): pass transport = CaptureTransport() with start(agent="langchain-demo", user_goal="Summarize the doc", transport=transport) as session: handler = PromptetheusCallbackHandler(session) run_id = uuid.uuid4() handler.on_llm_start({"name": "FakeLLM"}, ["hello"], run_id=run_id) class Result: llm_output = {"token_usage": {"prompt_tokens": 3, "completion_tokens": 2}} generations: list = [] handler.on_llm_end(Result(), run_id=run_id) session_id = session.session_id types_seen = {event["type"] for event in transport_events} print("session_id:", session_id) print("event_types:", sorted(types_seen)) assert "llm_call" in types_seen assert "session_end" in types_seen if __name__ == "__main__": main() ``` ### .github/workflows/publish.yml ```yaml name: Publish to PyPI on: release: types: [published] workflow_dispatch: permissions: id-token: write jobs: publish: runs-on: ubuntu-latest environment: pypi steps: - uses: actions/checkout@v4 - uses: actions/setup-python@v5 with: python-version: "3.14" - name: Install build backend run: python -m pip install --upgrade pip hatchling build - name: Build package working-directory: packages/promptetheus run: python -m build - name: Publish to PyPI uses: pypa/gh-action-pypi-publish@release/v1 with: packages-dir: packages/promptetheus/dist/ ``` ### tests/sdk/test_propagation.py ```python from __future__ import annotations import sys from pathlib import Path PACKAGE_ROOT = Path(__file__).resolve().parents[2] / "packages" / "promptetheus" sys.path.insert(0, str(PACKAGE_ROOT)) from promptetheus.propagation import ( # noqa: E402 TraceContext, extract, inject, new_trace_context, session_kwargs_from_context, ) def test_new_trace_context_is_valid(): ctx = new_trace_context() assert len(ctx.trace_id) == 32 and all(c in "0123456789abcdef" for c in ctx.trace_id) assert len(ctx.parent_id) == 16 assert ctx.trace_id != "0" * 32 and ctx.parent_id != "0" * 16 def test_inject_extract_round_trip(): ctx = new_trace_context() headers = inject(ctx, {"content-type": "application/json"}) assert headers["content-type"] == "application/json" # preserves other headers assert headers["traceparent"].startswith("00-") got = extract(headers) assert got is not None assert got.trace_id == ctx.trace_id assert got.parent_id == ctx.parent_id def test_extract_case_insensitive_header(): ctx = new_trace_context() got = extract({"TraceParent": ctx.to_traceparent()}) assert got is not None and got.trace_id == ctx.trace_id def test_extract_tolerates_missing_and_malformed(): assert extract(None) is None assert extract({}) is None assert extract({"traceparent": "garbage"}) is None assert extract({"traceparent": "00-xyz-abc-01"}) is None # all-zero ids are invalid assert extract({"traceparent": f"00-{'0'*32}-{'0'*16}-01"}) is None def test_traceparent_format(): ctx = TraceContext(trace_id="a" * 32, parent_id="b" * 16) assert ctx.to_traceparent() == f"00-{'a'*32}-{'b'*16}-01" def test_session_kwargs_from_context(): ctx = new_trace_context() kwargs = session_kwargs_from_context(ctx) assert kwargs["metadata"]["trace_id"] == ctx.trace_id assert kwargs["metadata"]["parent_span_id"] == ctx.parent_id ``` ### tests/adapters/conftest.py ```python """Import-isolation for the adapter test suite. Some adapter tests are lib-verified: they import the real third-party framework (for example dspy) to prove the adapter is a genuine subclass of the documented base and that driving the real hook surface stays thin. Importing a real framework transitively pulls in heavy provider SDKs (dspy imports openai), and those leak into sys.modules for the remainder of the process. That leak silently breaks the import-safety contract that other adapter tests assert, namely that importing a promptetheus adapter never imports its provider SDK at load time. test_openai, for instance, asserts openai is absent before and after importing promptetheus.adapters.openai; if an earlier dspy test already loaded openai, that precondition misfires through no fault of the adapter. This autouse fixture undoes that leak. It targets only the provider SDK top-level packages that other adapter tests assert are absent (openai and anthropic), and only removes them if they were not already imported when the test began. It deliberately does not touch any other module, so frameworks that cache partially-initialized submodules during import are left intact, and the import-safety preconditions read true regardless of test order without weakening any assertion. """ from __future__ import annotations import sys from collections.abc import Iterator import pytest # Provider SDKs whose absence other adapter tests assert as their import-safety # precondition. Lib-verified framework tests transitively import these, so we # unload any that a test newly introduced to keep that precondition faithful. _PROVIDER_SDKS = ("openai", "anthropic") @pytest.fixture(autouse=True) def _isolate_provider_sdks() -> Iterator[None]: present_before = {name for name in _PROVIDER_SDKS if name in sys.modules} try: yield finally: for name in _PROVIDER_SDKS: if name in present_before: continue for mod in [m for m in sys.modules if m == name or m.startswith(name + ".")]: del sys.modules[mod] ``` ### tests/sdk/test_sampling.py ```python from __future__ import annotations import asyncio import sys from pathlib import Path PACKAGE_ROOT = Path(__file__).resolve().parents[2] / "packages" / "promptetheus" sys.path.insert(0, str(PACKAGE_ROOT)) import promptetheus as pt # noqa: E402 from promptetheus.session import Session, _should_record # noqa: E402 class RecordingTransport: def __init__(self): self.events = [] self.flushed = False def send_event(self, event): self.events.append(event) def flush(self, timeout=None): self.flushed = True def test_should_record_extremes(): assert _should_record("any", 1.0) is True assert _should_record("any", 0.0) is False def test_should_record_is_deterministic_per_session(): assert _should_record("sess_abc", 0.5) == _should_record("sess_abc", 0.5) assert _should_record("sess_xyz", 0.5) == _should_record("sess_xyz", 0.5) def test_sampled_out_session_emits_nothing_but_runs(): transport = RecordingTransport() session = Session( agent="a", user_goal="g", session_id="s1", transport=transport, sample_rate=0.0 ) session.agent_message("hello") session.tool_call("t") assert transport.events == [] def test_sampled_in_session_emits(): transport = RecordingTransport() session = Session( agent="a", user_goal="g", session_id="s1", transport=transport, sample_rate=1.0 ) session.agent_message("hello") assert [e["type"] for e in transport.events] == ["agent_message"] def test_observe_respects_sample_rate_zero(): transport = RecordingTransport() @pt.observe(agent="x", user_goal="g", transport=transport, sample_rate=0.0) def run(): pt.current().agent_message("inside") return 7 assert run() == 7 # user code still runs assert transport.events == [] # but nothing recorded def test_async_observe_records_when_sampled_in(): transport = RecordingTransport() @pt.observe(agent="x", user_goal="g", transport=transport, sample_rate=1.0) async def run(): pt.current().agent_message("inside") return 5 assert asyncio.run(run()) == 5 types = [e["type"] for e in transport.events] assert types[0] == "state_change" assert "agent_message" in types assert types[-1] == "session_end" ``` [106 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]