# Project export: Pulse AI
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: Real verified people answer your medical questions honestly, before you decide. Built so an AI agent never guesses, it waits for a real person, then tells you how well their experience fits yours.
- Devpost: https://devpost.com/software/pulse-ai-ub9lxt
- GitHub: https://github.com/2006-sk/AIhack
- Video: https://www.youtube.com/embed/-EC0xCrg5H8?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent
- Result: winner (Best Use of Orkes; Best Use of TokenRouter by PaleBlueDot AI)
- Team: 2 GitHub contributor(s) — 2006-sk (1 commits), Cursor (1 commits)
## Devpost submission (written by the team)
### Inspiration
Our teammate Kenil got LASIK. He came out of it with a permanent dark spot in his vision that nobody had told him to expect. He was given statistics beforehand, the way every patient is, but no one who had actually lived through that specific outcome ever got the chance to tell him what it was really like, or whether it fades, or how to think about it. That's the gap we wanted to close. Before an elective or irreversible procedure, people are told probabilities. What they actually want is a real person who's been through it, telling them the truth.
### What it does
A patient uploads their pre-surgery consult notes, the actual document their doctor gave them. Pulse AI reads it, pulls out the procedure and the patient's specific concern, and launches a real study on Terac targeting people who've genuinely had that exact procedure. While it waits for a real, verified respondent, the agent doesn't guess or generate a placeholder answer. It pauses, durably, for as long as it takes, then resumes the moment a real account comes in. The response a patient sees is built from that real person's actual words, clearly scoped to how closely their situation matches the patient's own, and never inflated into more confidence than the sample size actually supports. If \( n \) is the number of verified respondents behind a given answer, Pulse AI always shows \( n \) explicitly rather than letting a single account read as a general outcome. Right now \( n = 1 \) for every answer, and the product says so.
### How we built it
The core is an agent built on Agentspan, Orkes' durable execution SDK, structured as a sequence of tool calls: parse the intake document, launch a Terac study, pause and wait for a real respondent, check how well that respondent's situation actually matches the patient's question, synthesize an honest answer, then log the run for evaluation. The pause step is the part we cared about most. Real human response time can't be forced into a few seconds, so the workflow has to survive that gap without losing the original question or any prior step's output, which is exactly what Agentspan's approval-gated tools are built for. Every model call in the pipeline runs through PaleBlueDot's TokenRouter rather than a single hardcoded provider, so cheap extraction steps and careful patient-facing writing each get routed to an appropriately sized model, with automatic failover if a provider has an issue. Every completed run is traced through Arize Phoenix and logged for evaluation, with the explicit goal of catching one specific failure mode: the AI claiming something a real respondent never actually said.
### Challenges we ran into
The honest one is selection and matching, not infrastructure. Two people who've had the exact same procedure can have completely different experiences, so a system that just hands back any matching respondent's story risks implying relevance that isn't there. We built an explicit match-quality check for this rather than papering over it, and the first time we tested it end to end with a real respondent, the system correctly flagged a partial match. The respondent had the same procedure but a meaningfully different complication than the patient asked about, and it said so honestly instead of presenting it as a clean answer. That moment is the actual proof of the thing we set out to build. We also went through a real architecture correction mid-build. We initially assumed Terac would push completion data to us as a webhook, then learned there were no webhooks available yet, which meant rebuilding the resume mechanism around polling instead of an inbound callback. Catching that early, rather than discovering it the night before judging, mattered.
### What we learned
Retrieval is safer than generation when the subject is someone's real medical decision. Early on we considered letting the model infer or extrapolate an answer when no real respondent was available, and we deliberately ruled that out. An AI inventing a plausible-sounding account of a stranger's medical experience is a worse outcome than telling a patient honestly that no one has answered yet.
### What's next
A real outcome-distribution view rather than individual anecdotes alone, since showing only verified stories risks skewing toward people motivated to share strong experiences. We'd also want broader procedure coverage and faster respondent recruitment paths so the wait between a question and a real answer keeps shrinking.
## README (from the GitHub repository)
# Afterward
*A backend AI agent that connects patients considering a medical procedure with real people who have actually had it — and never fabricates the answer.*
## 📖 Overview
**Afterward** is a hackathon submission. It is a backend agent that answers a patient's pre-procedure question not with generic model output, but with the experience of a **real human** who has been through the same procedure.
A patient pastes their pre-procedure consult note. The agent extracts the procedure and the patient's real concern, launches a study to recruit someone who has had that procedure, **durably pauses** until that person responds, and only then writes a patient-facing answer grounded strictly in the real respondent's words.
The orchestration is built on [Agentspan](https://pypi.org/project/agentspan/), whose durable execution runtime appears to be Conductor/Orkes-backed — the code guards tool state against "breaking Conductor serialization" and runs against a Java runtime server on `localhost:6767` (this is also why the default branch is named `orkes`).
> ⚠️ This is a hackathon build. It ships with **stubs enabled by default** so the full flow runs without live external services — the Terac endpoint is an explicit placeholder, and the LLM tools fall back to canned output unless real API keys are provided.
## ✨ Features
- **Never fabricates an answer.** The agent must complete a fixed four-step tool sequence — `parse_intake → launch_terac_study → wait_for_real_response → synthesize_answer` — and is instructed never to skip a step or invent a respondent's answer.
- **Human-in-the-loop durable pause.** `wait_for_real_response` is an approval-gated tool: the execution durably pauses until a real respondent's answer arrives, then resumes exactly where it left off.
- **Webhook-driven resume.** A FastAPI endpoint (`/terac-webhook`) receives study-completion callbacks and resumes the paused execution by mapping `study_id → execution_id`.
- **Multi-model routing by responsibility.** Anthropic Claude handles orchestration / tool-calling only; a separate OpenAI-compatible router (TokenRouter) handles mechanical intake parsing and patient-facing writing, each with its own model.
- **Failover demo.** `SIMULATE_PROVIDER_OUTAGE=1` sends a broken model and invalid auth to the router to exercise the provider-failover path.
- **Stub modes for offline demos.** Runs end-to-end without a live Terac API or LLM keys via built-in stubs and fallback respondent answers.
- **Structured, secret-redacting logging.** Every log line carries a phase and `execution_id`; API keys and tokens are redacted, and `scripts/tail_execution.sh` filters logs down to a single execution.
## 🛠️ Tech Stack
| Area | Technology |
| --- | --- |
| Language | Python |
| Agent orchestration | Agentspan (durable executions, approval-gated tools; Conductor/Orkes-backed runtime) |
| Orchestrator LLM | Anthropic Claude via Agentspan (default `anthropic/claude-sonnet-4-6`) |
| Intake / synthesis LLMs | TokenRouter (OpenAI-compatible), e.g. `qwen3.6-flash` and `deepseek/deepseek-v4-pro` |
| Respondent recruiting | Terac studies API (with webhook callback) |
| Webhook server | FastAPI + Uvicorn |
| Utilities | `requests`, `python-dotenv` |
## 🚀 Getting Started
### Prerequisites
- Python 3.9+
- The **Agentspan runtime server** (a Java daemon reachable at `http://localhost:6767`). It can be started for you by the helper script below (`agentspan server start`).
- API keys (only required to leave stub mode):
- `ANTHROPIC_API_KEY` — required for the agent's tool-calling orchestrator.
- `TOKENROUTER_API_KEY` — required for real intake parsing and answer synthesis.
### Installation
```bash
# Install Python dependencies
pip install -r requirements.txt
# Configure environment
cp .env.example .env
# then edit .env and fill in the keys you need
```
### Usage
The project is built in **phases**, each a runnable smoke/end-to-end test.
```bash
# 0. Ensure the Agentspan runtime server is up (starts it if needed)
python scripts/ensure_server.py
# 1. Agentspan hello-world smoke test (weatherbot)
python phase1/weatherbot.py
# 2. Start the Afterward agent and poll until it pauses for approval
python phase2/run_agent.py
# then, in another terminal, approve with a respondent answer:
python phase2/approve_response.py "I had this procedure. Here's my experience..."
# and poll to completion:
python phase2/wait_complete.py
# 3. Test the LLM-backed tools against the sample consult notes
# (requires TOKENROUTER_API_KEY, or SIMULATE_PROVIDER_OUTAGE=1 to demo failover)
python phase3/test_llm_tools.py
# 4. Full end-to-end via the webhook path
python phase4/run_webhook.py # terminal A: FastAPI webhook receiver
python phase4/run_e2e.py # terminal B: run agent → fire webhook → complete
python phase4/test_webhook.py "respondent answer" # post a webhook manually
```
Key configuration flags (see `.env.example`):
| Variable | Purpose |
| --- | --- |
| `AFTERWARD_LLM_TOOLS` | `1` to use real TokenRouter tools; otherwise stubbed output |
| `TERAC_USE_STUB` | `1` to return fake study IDs instead of calling the Terac API |
| `SIMULATE_PROVIDER_OUTAGE` | `1` to exercise the provider-failover path |
| `AGENTSPAN_MODEL` | Orchestrator model (default `anthropic/claude-sonnet-4-6`) |
| `AFTERWARD_WEBHOOK_BASE_URL` / `WEBHOOK_PORT` | Webhook receiver location |
## 📁 Project Structure
```
.
├── afterward/ # Core agent package
│ ├── agent.py # Agent definition + enforced 4-step tool sequence
│ ├── tools/ # parse_intake, launch_terac_study, wait_for_real_response, synthesize_answer
│ ├── terac_client.py # Terac studies API client (+ stub)
│ ├── tokenrouter.py # OpenAI-compatible TokenRouter client
│ ├── webhook_app.py # FastAPI receiver for Terac completion callbacks
│ ├── resume_execution.py # Resume a paused execution with a respondent answer
│ ├── study_mapping.py # study_id → execution_id mapping
│ ├── pending_responses.py # respondent-answer cache keyed by execution_id
│ └── logging_config.py # structured, secret-redacting logging
├── phase1/ … phase4/ # Phased smoke / end-to-end test scripts
├── samples/ # Example pre-procedure consult notes
├── scripts/ # ensure_server.py, tail_execution.sh
├── requirements.txt
└── .env.example
```
## Detected evidence (automated analysis)
Indexed codebase: 25 recognized source files, 55 KB.
- FastAPI (technology) — detected in the code
- Python (language) — detected in the code
- React (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
- AI coding agent: Cursor — evidence: commit authorship or trailers
## Codebase structure (from repository index)
### Files (31 of 31)
```
.env.example
.gitignore
afterward/__init__.py
afterward/agent.py
afterward/logging_config.py
afterward/pending_responses.py
afterward/resume_execution.py
afterward/study_mapping.py
afterward/terac_client.py
afterward/tokenrouter.py
afterward/tool_state.py
afterward/tools/__init__.py
afterward/tools/intake.py
afterward/tools/synthesize.py
afterward/tools/terac.py
afterward/tools/wait.py
afterward/webhook_app.py
phase1/weatherbot.py
phase2/approve_response.py
phase2/run_agent.py
phase2/wait_complete.py
phase3/test_llm_tools.py
phase4/run_e2e.py
phase4/run_webhook.py
phase4/test_webhook.py
requirements.txt
samples/lasik_consult.txt
samples/vasectomy_consult.txt
samples/wisdom_teeth_consult.txt
scripts/ensure_server.py
scripts/tail_execution.sh
```
### Dependencies
- requirements.txt: agentspan@>=0.1.10, fastapi@>=0.100.0, python-dotenv@>=1.0.0, requests@>=2.31.0, uvicorn@>=0.24.0
### Recent commits (newest first)
- Add Afterward durable agent backend (Agentspan + TokenRouter).
## Key source files (fetched from GitHub, selected and truncated for size)
### requirements.txt
```
agentspan>=0.1.10
requests>=2.31.0
python-dotenv>=1.0.0
fastapi>=0.100.0
uvicorn>=0.24.0
```
### afterward/__init__.py
```python
"""Afterward — durable medical procedure Q&A agent (hackathon build)."""
```
### scripts/tail_execution.sh
```shell
#!/usr/bin/env bash
# Filter afterward.log for a single execution_id.
# Usage: ./scripts/tail_execution.sh
# ./scripts/tail_execution.sh -f # follow
set -euo pipefail
EXEC_ID="${1:?usage: tail_execution.sh [-f]}"
shift || true
LOG_FILE="${LOG_FILE:-afterward.log}"
if [[ "${1:-}" == "-f" ]]; then
tail -f "$LOG_FILE" | grep --line-buffered "$EXEC_ID"
else
grep "$EXEC_ID" "$LOG_FILE"
fi
```
### phase4/run_webhook.py
```python
#!/usr/bin/env python3
"""Run the Terac webhook receiver (FastAPI + uvicorn)."""
from __future__ import annotations
import os
import sys
from pathlib import Path
from dotenv import load_dotenv
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
load_dotenv()
import uvicorn
from afterward.logging_config import setup_logging
logger = setup_logging("run_webhook")
if __name__ == "__main__":
host = os.environ.get("WEBHOOK_HOST", "0.0.0.0")
port = int(os.environ.get("WEBHOOK_PORT", "8787"))
logger.info("Starting Terac webhook server on %s:%s", host, port)
uvicorn.run("afterward.webhook_app:app", host=host, port=port, log_level="info")
```
### afterward/tool_state.py
```python
"""Safe context.state updates for Agentspan tool persistence."""
from __future__ import annotations
from agentspan.agents import ToolContext
ALLOWED_STATE_KEYS = frozenset({
"procedure",
"question_focus",
"screening_criteria",
"terac_study_id",
"respondent_answer",
"patient_response",
})
def patch_tool_state(context: ToolContext, **fields: str) -> None:
"""Keep only flat string state keys — nested dicts break Conductor serialization."""
kept: dict[str, str] = {}
for key, value in context.state.items():
if key in ALLOWED_STATE_KEYS and isinstance(value, (str, int, float, bool)):
kept[key] = str(value)
for key, value in fields.items():
if key in ALLOWED_STATE_KEYS:
kept[key] = str(value)
context.state.clear()
context.state.update(kept)
```
### afterward/resume_execution.py
```python
"""Resume a paused Afterward execution after an external respondent answer arrives."""
from __future__ import annotations
import logging
from agentspan.agents import AgentRuntime
from afterward.agent import build_afterward_agent
from afterward.pending_responses import set_pending_response
logger = logging.getLogger("afterward.resume_execution")
PHASE = "resume_execution"
def resume_with_respondent_answer(execution_id: str, respondent_answer: str) -> None:
"""Store answer in cache, resume workers, and approve the pending wait tool."""
logger.info("STATE transition=resume_start execution_id=%s", execution_id)
set_pending_response(execution_id, respondent_answer)
agent = build_afterward_agent()
with AgentRuntime() as runtime:
handle = runtime.resume(execution_id, agent)
logger.info("Calling handle.approve() for execution_id=%s", execution_id)
handle.approve()
logger.info("STATE transition=resume_success execution_id=%s", execution_id)
```
### phase2/approve_response.py
```python
#!/usr/bin/env python3
"""Approve a paused Afterward execution and inject a fake respondent answer."""
from __future__ import annotations
import sys
from pathlib import Path
from dotenv import load_dotenv
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
load_dotenv()
from afterward.logging_config import set_execution_id, setup_logging
from afterward.resume_execution import resume_with_respondent_answer
PHASE = "phase2_approve_response"
logger = setup_logging(PHASE)
def main() -> None:
if len(sys.argv) < 3:
logger.error("Usage: python phase2/approve_response.py ")
sys.exit(1)
execution_id = sys.argv[1]
respondent_answer = " ".join(sys.argv[2:])
token = set_execution_id(execution_id)
try:
logger.info("STATE transition=approve_requested execution_id=%s", execution_id)
resume_with_respondent_answer(execution_id, respondent_answer)
logger.info("Approve succeeded for execution_id=%s", execution_id)
except Exception:
logger.exception("Approve FAILED for execution_id=%s", execution_id)
sys.exit(1)
finally:
from afterward.logging_config import reset_execution_id
reset_execution_id(token)
if __name__ == "__main__":
main()
```
### phase4/test_webhook.py
```python
#!/usr/bin/env python3
"""Post a fake Terac webhook payload to test the resume path without a real study."""
from __future__ import annotations
import json
import os
import sys
from pathlib import Path
import requests
from dotenv import load_dotenv
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
load_dotenv()
from afterward.logging_config import setup_logging
logger = setup_logging("phase4_test_webhook")
def main() -> None:
if len(sys.argv) < 3:
logger.error(
"Usage: python phase4/test_webhook.py \n"
" study_id comes from launch_terac_study logs or afterward_study_mapping.json"
)
sys.exit(1)
study_id = sys.argv[1]
respondent_answer = " ".join(sys.argv[2:])
base = os.environ.get("AFTERWARD_WEBHOOK_BASE_URL", "http://localhost:8787")
url = f"{base.rstrip('/')}/terac-webhook"
payload = {
"study_id": study_id,
"respondent_answer": respondent_answer,
"status": "completed",
"source": "phase4_test_webhook",
}
logger.info("POST %s payload=%s", url, json.dumps(payload)[:500])
try:
resp = requests.post(url, json=payload, timeout=30)
logger.info("Response status=%s body=%s", resp.status_code, resp.text[:2000])
if resp.status_code >= 400:
sys.exit(1)
except requests.RequestException as exc:
logger.error("Webhook POST failed: %s", exc)
sys.exit(1)
if __name__ == "__main__":
main()
```
### afterward/study_mapping.py
```python
"""study_id → execution_id mapping for Terac webhook resume."""
from __future__ import annotations
import json
import logging
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Optional
logger = logging.getLogger("afterward.study_mapping")
MAPPING_FILE = Path("afterward_study_mapping.json")
def _load() -> dict[str, dict[str, Any]]:
if not MAPPING_FILE.exists():
return {}
try:
return json.loads(MAPPING_FILE.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError) as exc:
logger.error("Failed to load study mapping: %s", exc)
return {}
def _save(data: dict[str, dict[str, Any]]) -> None:
MAPPING_FILE.write_text(json.dumps(data, indent=2), encoding="utf-8")
def register_study(
study_id: str,
execution_id: str,
*,
procedure: str = "",
screening_criteria: str = "",
) -> None:
data = _load()
data[study_id] = {
"execution_id": execution_id,
"procedure": procedure,
"screening_criteria": screening_criteria,
"registered_at": datetime.now(timezone.utc).isoformat(),
}
_save(data)
logger.info(
"STATE transition=study_mapped study_id=%s execution_id=%s",
study_id,
execution_id,
)
def lookup_execution_id(study_id: str) -> Optional[str]:
entry = _load().get(study_id)
if not entry:
logger.warning("No execution_id mapped for study_id=%s", study_id)
return None
return entry.get("execution_id")
def lookup_entry(study_id: str) -> Optional[dict[str, Any]]:
return _load().get(study_id)
```
### phase2/wait_complete.py
```python
#!/usr/bin/env python3
"""Poll an Afterward execution until complete and print the final output."""
from __future__ import annotations
import sys
import time
from pathlib import Path
from dotenv import load_dotenv
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
load_dotenv()
from agentspan.agents import AgentRuntime
from afterward.agent import build_afterward_agent
from afterward.logging_config import set_execution_id, setup_logging
PHASE = "phase2_wait_complete"
logger = setup_logging(PHASE)
POLL_SECONDS = 2.0
MAX_WAIT_SECONDS = 300
def main() -> None:
if len(sys.argv) < 2:
logger.error("Usage: python phase2/wait_complete.py ")
sys.exit(1)
execution_id = sys.argv[1]
token = set_execution_id(execution_id)
try:
agent = build_afterward_agent()
with AgentRuntime() as runtime:
handle = runtime.resume(execution_id, agent)
deadline = time.monotonic() + MAX_WAIT_SECONDS
while time.monotonic() < deadline:
status = handle.get_status()
logger.info(
"STATUS poll is_complete=%s is_waiting=%s status=%s",
status.is_complete,
status.is_waiting,
status.status,
)
if status.is_complete:
logger.info("STATE transition=agent_complete output=%s", status.output)
print("\n--- FINAL OUTPUT ---")
print(status.output)
return
time.sleep(POLL_SECONDS)
logger.error("Timed out waiting for completion execution_id=%s", execution_id)
sys.exit(1)
finally:
from afterward.logging_config import reset_execution_id
reset_execution_id(token)
if __name__ == "__main__":
main()
```
[16 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]