# Project export: Open Door 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: **Care, made legible.** An AI agent that does the bureaucratic and physical errands of daily life for people who can't, and never does anything costly or irreversible without asking first. - Devpost: https://devpost.com/software/open-door-dojp02 - GitHub: https://github.com/Jeffrey-Le/open-door - Video: https://www.youtube.com/embed/ohXRDbgQunM?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent - Team: 2 GitHub contributor(s) — Jeffrey-Le (5 commits), Claude Opus 4.8 (1 commits) ## Devpost submission (written by the team) ### Inspiration Daily life now runs on phone trees and web portals. If you're homebound, low-vision, low-literacy, or elderly, that quietly locks you out of basic dignity-level tasks: refilling a prescription, renewing a registration, paying a bill. Existing tools assume you can see a screen and supervise an agent the whole way. The people who most need help are the ones those tools were never built for. Open Door is our swing at that gap: an agent that talks to you, works the web for you, and stops to ask out loud before it ever spends your money or does something it can't undo. ### What it does You say what you need ("I need to refill my metformin, I can't get to the pharmacy") Open Door: Plans it out loud: One Claude call turns your spoken goal into a visible, ordered plan. Each step labeled with which "body" handles it and why Works the portal for you: A real browser navigates the pharmacy site, finds the prescription, and reads the actual out-of-pocket cost off the page Stops at the gate: Before the one irreversible step — submitting the refill — it speaks the cost aloud ("This will charge you $14 and submit your refill. Should I go ahead?") and waits. You answer by voice -> "yes" or "no" Only then acts: On "yes," it submits and reads back the confirmation. On "no," it stops cold and tells you nothing was charged The plan is the hero of the screen: steps light up as they run, the gated step pauses red, and the question is spoken, so a person who can't see or touch the screen can complete a real, costly action entirely by voice. ### How we built it The architecture is built around seams — every external service is swappable, so the whole thing runs offline against mocks and flips to live with one env var. Planner (Anthropic, Claude Opus 4.8) - The centerpiece: A strict-JSON planner with adaptive thinking, defensive parsing, and a content-guard that retries if the spoken wording reads like a stage direction instead of real speech. It generalizes: give it a DMV renewal or a utility bill and it produces a correct, gated plan Browse leg (Browserbase + Playwright) - Drives a real Chromium through the portal: Local by default (free), Browserbase cloud with one env flag — same Playwright code over CDP. Self-healing navigation so the run survives messy pages. Stop-before-submit is structural: the effector never decides to submit; it only clicks the button when dispatch hands it the gated step, which only happens after a human "yes" Speak leg (Deepgram) - Both directions: TTS (Aura) voices the gate question and every spoken line; STT (Nova) hears your goal at intake and your yes/no at the gate. The spoken confirmation is load-bearing, not decoration Dispatch + the human gate - A pausable state machine: it physically parks the browser on the irreversible button and refuses to proceed until a human decides. Negative answers win on ambiguity. It never proceeds unless it clearly heard "yes." Declining skips the rest of the plan so it can never falsely report success Observability (Sentry) - Every effector is instrumented: every gate leaves a breadcrumb of exactly what the human approved. We exercised it for real (killed the portal mid-browse and watched the capture land), because un-triggered observability doesn't count Frontend - A single-page hero: UI streaming live state over Server-Sent Events (push on change, not polling), plus a connected-services landing that frames Open Door as a platform for all of daily life's errands 22 regression tests, an 8/8 live health check, and an offline-first build kept it honest. ### Challenges we ran into Making spoken output sound human: The planner kept reading step descriptions aloud ("Confirm which pharmacy holds their prescription"). We fixed it with a sharper prompt plus a deterministic guard that detects stage-direction phrasing and retries for real second-person speech Generalist planner vs. scripted hands: The planner imagines portal features (home delivery) the mock fixture doesn't have. We made the browse leg self-heal and tolerant so any goal completes rather than timing out A "no" that still said yes: Early on, declining the gate still ran the downstream "all done" steps. We made declining halt the plan — the bug that most violated our own thesis, and the one we're proudest to have caught ### Accomplishments we're proud of The spoken safety gate. The agent parks on the irreversible button and asks aloud, and you answer aloud. It's a small thing that makes a powerful agent safe to hand to someone who can't supervise it. That's the whole point. ### What we learned Building a careful agent is mostly about designing where it stops, not where it acts. The seams and the gate were more engineering than the "doing," and that's the right ratio for something that spends a vulnerable person's money. ### What's next Generalize the browse leg to natural-language web navigation so it handles any real portal (the planner already generalizes); add real account connections per service; a pending-errands queue and an in-app action history. ## README (from the GitHub repository) # Open Door **An AI agent that does the bureaucratic errands of daily life for people who can't — and never spends their money or does anything irreversible without asking first, out loud.** Daily life now runs on phone trees and web portals. If you're homebound, low-vision, low-literacy, or elderly, that quietly locks you out of basic, dignity-level tasks: refilling a prescription, renewing a registration, paying a bill. Tools built for general users assume you can see a screen and supervise an agent the whole way — the people who most need help are the ones they were never built for. Open Door takes a spoken goal, decomposes it into a visible plan, works the web on your behalf, and **stops to ask — by voice — before anything costly or irreversible.** Care made legible: powerful enough to act, careful enough to ask first. > **Flagship demo:** a homebound patient refills a prescription by voice. The agent confirms details aloud, navigates the pharmacy portal, reads the real out-of-pocket cost off the page, and pauses for a spoken "yes" before submitting. --- ## How it works A single Claude planner turns a spoken goal into a structured `Plan`. Each step is dispatched to an **effector** — a swappable "body" — and the run pauses at a human gate before any costly or irreversible action. | Leg | What it does | Powered by | |---|---|---| | **Plan** | Decomposes the goal into a visible, ordered plan; tags each step's risk; writes the spoken lines | Claude (Anthropic) | | **Browse** | Drives a real browser through the portal: read the page, find the item, surface the real cost, submit | Playwright (local) / Browserbase (cloud) | | **Speak** | Voices the gate question and reads results aloud (TTS); hears the goal and the yes/no answer (STT) | Deepgram (Aura + Nova) | | **Gate** | Pauses at any costly/irreversible step and refuses to proceed without an explicit human "yes" | dispatch state machine | | **Observe** | Captures every effector failure and records what the human approved at each gate | Sentry | Two design rules hold the whole thing together: 1. **The Plan is a first-class, serializable object** — the frontend renders it directly, so you watch the agent's reasoning as live data. 2. **Effectors share one Protocol** and are resolved by a registry lookup (never an `if/else`), so mock and live backends are interchangeable and a new leg is one `register()` call. Going live is configuration, not a refactor. The gate is **structural, not cosmetic**: the browse effector never decides to submit — it only clicks the irreversible button when dispatch hands it the gated step, which dispatch does only after a human yes. Declining halts the rest of the plan, so it can never falsely report success. --- ## Try it out ### Prerequisites - Python 3.11 (conda recommended) ### 1. Set up the environment ```bash conda env create -f environment.yml conda activate opendoor python -m playwright install chromium # for the real browse leg ```
Prefer a plain venv? ```bash python -m venv .venv && source .venv/bin/activate pip install -r requirements.txt python -m playwright install chromium ```
### 2. Run it — no API keys needed ```bash python -m frontend.server ``` Open **http://127.0.0.1:8000** and click **Plan & run**. Out of the box it runs **fully offline**: a deterministic golden plan, mock effectors, and the browser's built-in voice. To use a **real local browser** against the bundled mock pharmacy portal (still no key required), set `OPENDOOR_BROWSE=real`. ### 3. Go live (optional) Copy the template and fill in whichever keys you have: ```bash cp .env.example .env ``` ```ini ANTHROPIC_API_KEY=... # live planner (real goal -> real plan) DEEPGRAM_API_KEY=... # real voice in + out OPENDOOR_SPEAK=real BROWSERBASE_API_KEY=... # cloud browser (optional; local needs no key) BROWSERBASE_PROJECT_ID=... OPENDOOR_SENTRY=1 # observability (optional) SENTRY_DSN=... ``` Restart the server. `.env` is loaded automatically and is gitignored — keys never get committed. Each leg degrades gracefully: with no key, it falls back to the offline path for that leg. ### 4. Verify ```bash pytest # 26 offline tests (no keys, no network) python scripts/healthcheck.py # live end-to-end probe of every leg (needs keys) ``` --- ## Configuration All optional; set in `.env` or the environment. | Variable | Effect | |---|---| | `OPENDOOR_BROWSE` | `real` = local Chromium · `cloud` = Browserbase · unset = offline mock | | `OPENDOOR_HEADED=1` | Show the Chromium window during a run (default headless) | | `OPENDOOR_SPEAK=real` | Use Deepgram voice (TTS+STT); otherwise the browser's Web Speech | | `OPENDOOR_SENTRY=1` | Enable Sentry (requires `SENTRY_DSN`) | | `OPENDOOR_PLANNER_MODEL` | Override the planner model (default `claude-opus-4-8`) | --- ## Project structure ``` agent/ contracts.py Plan / Step / Risk / Effector / EffectorBackend planner.py the Claude planner: prompt, defensive JSON parsing, retries dispatch.py execution loop + the pausable human gate observability.py Sentry seam (instrument effectors + record gate decisions) store.py JSON-backed persistence (services, errands, history) demo.py the golden plan + env-selected effector registry effectors/ base.py the effector registry (the open seam, in code) browse.py Playwright/Browserbase web navigation speak.py Deepgram TTS + STT mock.py offline mock backends frontend/ server.py FastAPI host: drives runs, streams state over SSE index.html the hero UI: live plan, spoken gate, errands, history portal/ mock_pharmacy.html deterministic test fixture for the browse leg tests/ 26 tests, all offline scripts/healthcheck.py ``` --- ## Status & what's next All four legs are live and verified, with persistence, an urgency-ordered errands queue, a connected-services view, and an in-app action history. The planner already generalizes to new errands (e.g. a DMV renewal); the browse leg is scripted to the demo portal today. The clear next steps are generalizing the browse leg to natural-language navigation of arbitrary real portals, real per-service account connections, and an interactive "clarify" step that collects missing details mid-plan. Built for the AI Hackathon 2026 with Claude, Deepgram, Browserbase, and Sentry. ## Detected evidence (automated analysis) Indexed codebase: 26 recognized source files, 157 KB. - Anthropic (technology) — detected in the code - FastAPI (technology) — detected in the code - HTML (language) — detected in the code - Python (language) — detected in the code - AI coding agent: Claude Code — evidence: commit authorship or trailers ## Codebase structure (from repository index) ### Files (29 of 29) ``` .env.example .github/workflows/deploy.yml .gitignore agent/__init__.py agent/contracts.py agent/demo.py agent/dispatch.py agent/observability.py agent/planner.py agent/store.py DEVPOST.md effectors/__init__.py effectors/base.py effectors/browse.py effectors/mock.py effectors/speak.py environment.yml frontend/index.html frontend/server.py PITCH.md portal/mock_pharmacy.html README.md requirements.txt scripts/healthcheck.py tests/test_browse.py tests/test_dispatch.py tests/test_planner.py tests/test_speak.py tests/test_store.py ``` ### Dependencies - requirements.txt: anthropic, browserbase, deepgram-sdk, fastapi, httpx, jinja2, playwright, pytest, python-multipart, sentry-sdk, uvicorn[standard] ### Recent commits (newest first) - Update server.py - Fixing Bugs - Merge branch 'main' of https://github.com/Jeffrey-Le/open-door - deploy: ready for persistent hosting - Create deploy.yml - Rewrite README as project documentation - Open Door: voice-driven agent for daily-life errands with a spoken human gate ## Key source files (fetched from GitHub, selected and truncated for size) ### PITCH.md ```markdown # Open Door — recording script & framing > DEVPOST.md was locked when this was written (likely open in an editor / iCloud sync). > This is the up-to-date version — record from here, and paste into DevPost / merge into DEVPOST.md. ## The one-liner (lead with the person, not the feature) > **Open Door is an AI agent that does the bureaucratic errands of daily life for people who can't — homebound, low-vision, elderly — and never spends their money or does anything irreversible without asking, out loud.** Don't frame it as a to-do list or expense hub: a to-do app makes *you* do the task; Open Door *does it for you, by voice, for someone who can't.* The errands queue / connections / history are proof it's a **platform**, not a one-off — but the hero is **agent + voice + the gate that stops to ask.** --- ## 3-minute pitch script **(0:00–0:30) The problem.** "Daily life runs on phone trees and web portals. If you're homebound or can't see a screen well, that locks you out of basic things — refilling a prescription, paying a bill. The agents being built today assume you can watch them work. The people who need help most can't." **(0:30–0:55) The idea.** "Open Door is an agent for the bureaucratic errands of daily life — it keeps your to-do list, talks to you, works the web for you, and never does anything costly or irreversible without asking first, out loud. Let me show you." **(0:55–2:15) Live demo — open on the Errands tab.** - "Open Door keeps a to-do list of your daily chores — most urgent first." *(Errands queue visible)* - "I'll have it pay this pharmacy refill — I can just talk to it." *(Run now, or speak the goal)* - "It reasons about it — you watch the plan — and it's working the real portal now." *(steps light up; headed Chromium visible)* - "Here's the moment that matters: it found the real cost, and it stops." *(red gate; the $14 question is spoken aloud)* - "I answer out loud — yes." *(say "yes")* — "and only now does it submit. And it's logged in History." *(History tab — the record)* - *(If time: run once and say "no" — "it stops, and tells me nothing was charged.")* **(2:15–2:45) Why it's more than a demo.** "The gate is structural — the browser is physically parked on the submit button and can't proceed without a spoken yes. Every action is audited in Sentry. And the planner generalizes — here it is reasoning about a DMV renewal." *(DMV card → plan renders)* **(2:45–3:00) The aspiration.** "This is a hard, unsolved, dignity-level problem. Open Door is our swing at making an agent powerful enough to act — and careful enough to ask first." ### Q&A prep (2 min) - *"Is the gate real or theater?"* — Structural. The effector can't submit; dispatch only hands it the gated step after a human yes. Declining halts the rest of the plan. - *"How does it generalize?"* — The planner is general today (shown on DMV); the browse leg is scripted to the demo portal and generalizes via natural-language web navigation next. - *"Login/credential [truncated — 1477 more characters] ``` ### DEVPOST.md ```markdown # Open Door **Care, made legible. An AI agent that does the bureaucratic and physical errands of daily life for people who can't — and never does anything costly or irreversible without asking first.** --- ## Inspiration Daily life now runs on phone trees and web portals. If you're homebound, low-vision, low-literacy, or elderly, that quietly locks you out of basic dignity-level tasks: refilling a prescription, renewing a registration, paying a bill. Existing tools assume you can see a screen and supervise an agent the whole way. The people who most need help are the ones those tools were never built for. Open Door is our swing at that gap: an agent that **talks to you, works the web for you, and stops to ask — out loud — before it ever spends your money or does something it can't undo.** ## What it does You say what you need ("I need to refill my metformin, I can't get to the pharmacy"). Open Door: 1. **Plans it out loud.** One Claude call turns your spoken goal into a visible, ordered plan — each step labeled with which "body" handles it and why. 2. **Works the portal for you.** A real browser navigates the pharmacy site, finds the prescription, and reads the actual out-of-pocket cost off the page. 3. **Stops at the gate.** Before the one irreversible step — submitting the refill — it **speaks the cost aloud** ("This will charge you $14 and submit your refill. Should I go ahead?") and waits. You answer **by voice** — "yes" or "no." 4. **Only then acts.** On "yes," it submits and reads back the confirmation. On "no," it stops cold and tells you nothing was charged. The plan is the hero of the screen: steps light up as they run, the gated step pauses **red**, and the question is spoken — so a person who can't see or touch the screen can complete a real, costly action entirely by voice. ## How we built it The architecture is built around **seams** — every external service is swappable, so the whole thing runs offline against mocks and flips to live with one env var. - **Planner (Anthropic, Claude Opus 4.8)** — the centerpiece. A strict-JSON planner with adaptive thinking, defensive parsing, and a content-guard that retries if the spoken wording reads like a stage direction instead of real speech. It generalizes: give it a DMV renewal or a utility bill and it produces a correct, gated plan. - **Browse leg (Browserbase + Playwright)** — drives a real Chromium through the portal. Local by default (free), **Browserbase cloud with one env flag** — same Playwright code over CDP. Self-healing navigation so the run survives messy pages. *Stop-before-submit is structural*: the effector never decides to submit; it only clicks the button when dispatch hands it the gated step, which only happens after a human "yes." - **Speak leg (Deepgram)** — both directions. **TTS (Aura)** voices the gate question and every spoken line; **STT (Nova)** hears your goal at intake *and your yes/no at the gate*. The spoken confirmation is load-bearing, not decoratio [truncated — 5709 more characters] ``` ### requirements.txt ``` # Open Door — Python dependencies # Python 3.11+ recommended (contracts use 3.10+ union syntax). # --- Core: the planner (Anthropic centerpiece) --- anthropic # Claude SDK — the planner calls this # --- Server + frontend host --- fastapi # API server (reuse of the Due Process spine) uvicorn[standard] # ASGI server to run FastAPI jinja2 # if you template the single-file frontend python-multipart # form/file handling if the UI posts audio # --- Browse leg (Browserbase) --- # Browserbase is driven via their platform; install whichever harness you use. # Stagehand / Playwright are the common paths — pick one in Claude Code: playwright # if going the Playwright route (also: run `playwright install`) # stagehand # uncomment if you use Stagehand instead browserbase # Browserbase SDK/client # --- Speak leg (Deepgram) — both directions --- deepgram-sdk # STT (intake) + TTS (spoken gate confirmations) # --- Observability seam (Sentry) — optional, enabled by flag --- sentry-sdk # no-op unless OPENDOOR_SENTRY=1 + SENTRY_DSN # --- Dev / test --- pytest # regression tests against the golden scenario httpx # FastAPI TestClient dependency + general HTTP # NOTE: versions intentionally unpinned — let pip resolve latest-compatible on # first install, then freeze with `pip freeze > requirements.lock.txt` once it # works, so your demo environment is reproducible. ``` ### frontend/server.py ```python """ The demo host: FastAPI server that renders the Plan as the hero UI and drives it. Responsibilities, all thin: - Serve the single-file frontend (frontend/index.html) and the mock pharmacy portal (portal/mock_pharmacy.html) so the whole demo is self-contained. - Start a run (golden plan offline; real planner when ANTHROPIC_API_KEY is set) and drive it step-by-step in the background so steps visibly light up. - Pause at gates: the run sits in AWAITING_CONFIRM until POST /decision, and the UI lights that step RED and SPEAKS the question. State reaches the UI by Server-Sent Events: the server PUSHES a snapshot only when state actually changes (a step ran, a gate opened, the run finished), instead of the browser polling on a timer. One open stream per viewer, a handful of events per run -- not hundreds of GETs. The snapshot is exactly Plan.to_dict() plus gate context: the judge watches the agent's reasoning as data, no second source of truth. (A GET snapshot endpoint remains for tests and as a fallback.) """ from __future__ import annotations import asyncio import json import os import uuid from concurrent.futures import ThreadPoolExecutor from contextlib import asynccontextmanager from dataclasses import dataclass, field from pathlib import Path # Load secrets/config from .env into the environment BEFORE importing the agent # modules -- some (e.g. observability) read their flags at import time, so this # has to happen first. No-op if python-dotenv or the file is absent. try: from dotenv import load_dotenv load_dotenv(Path(__file__).resolve().parent.parent / ".env") except ImportError: pass from fastapi import FastAPI, HTTPException, Request from fastapi.responses import FileResponse, JSONResponse, Response, StreamingResponse from agent import store from agent.contracts import Plan from agent.demo import build_registry, golden_plan from agent.dispatch import AWAITING, DONE, FAILED, Runner from agent.observability import init_observability from effectors.speak import build_speak_backend # App-level speak backend for the audio endpoints (Deepgram when OPENDOOR_SPEAK # =real + key; else mock, and the frontend falls back to browser Web Speech). SPEAK = build_speak_backend() _TTS_AVAILABLE = hasattr(SPEAK, "synthesize") ROOT = Path(__file__).resolve().parent.parent STEP_GLOW_SECONDS = 0.9 # pause between steps so each visibly runs in the UI @asynccontextmanager async def lifespan(app: FastAPI): store.seed_defaults() # services + example errands on first run app.state.sentry_live = init_observability() # no-op unless OPENDOOR_SENTRY=1 planner = "live (ANTHROPIC_API_KEY set)" if os.environ.get("ANTHROPIC_API_KEY") else "golden-plan fallback" browse = os.environ.get("OPENDOOR_BROWSE", "mock") print(f"[open door] planner={planner} browse={browse} sentry={'live' if app.state.sentry_live else 'stub'}") yield app = FastAPI(title="Open Door", lifespan=lifespan) @dataclass class Run: """One in-flight plan execution. Holds the runner, the gate signal, the SSE subscribers to fan state out to, and a single-thread executor so all ticks for this run share one OS thread (required for the sync-Playwright browse leg: its browser/page is thread-bound and must persist across steps).""" runner: Runner decision_event: asyncio.Event = field(default_factory=asyncio.Event) finished: bool = False failed: bool = False task_id: str | None = None # set when this run was launched from an errand _subscribers: set[asyncio.Queue] = field(default_factory=set) executor: ThreadPoolExecutor = field( default_factory=lambda: ThreadPoolExecutor(max_workers=1, thread_name_prefix="run") ) @property def plan(self) -> Plan: return self.runner.plan def snapshot(self) -> dict: """The single payload the UI renders: the plan as data, plus gate context.""" awaiting = self.runner.awaiting_step return { "plan": self.plan.to_dict(), "finished": self.finished, "failed": self.failed, "awaiting_step_id": awaiting.id if awaiting else None, # The exact sentence to speak at the gate (TTS'd client-side; swap to # Deepgram audio when the key is present). "gate_question": (awaiting.args or {}).get("confirm_prompt") if awaiting else None, } def subscribe(self) -> asyncio.Queue: """Register an SSE listener. Seeds it with the current snapshot so a late joiner renders immediately rather than waiting for the next change.""" q: asyncio.Queue = asyncio.Queue() q.put_nowait(self.snapshot()) self._subscribers.add(q) return q def unsubscribe(self, q: asyncio.Queue) -> None: self._subscribers.discard(q) def publish(self) -> None: """Push the current snapshot to every listener. Called only on state change, so each run produces a handful of events, not a polling storm.""" snap = self.snapshot() for q in list(self._subscribers): q.put_nowait(snap) RUNS: dict[str, Run] = {} def _build_plan(goal: str | None) -> Plan: """Live planner when a goal AND ANTHROPIC_API_KEY are present; otherwise the offline golden plan. Planner failures fall back to the golden plan so the demo never hard-fails on a bad model response -- the Plan shape is identical.""" if goal and goal.strip() and os.environ.get("ANTHROPIC_API_KEY"): from agent.planner import make_plan # lazy: offline path needs no SDK try: return make_plan(goal.strip()) except Exception as exc: # noqa: BLE001 -- resilience over strictness for the demo print(f"[planner] live plan failed ({exc!r}); falling back to golden plan") return golden_plan() def _build_runner(goal: str | None = None) -> Runner: """A plan (live or golden) + the env-selected registry (mock, or a liv [truncated — 8180 more characters] ``` ### environment.yml ```yaml # Open Door — conda environment (alternative to requirements.txt + venv) # # Create with: conda env create -f environment.yml # Activate with: conda activate opendoor # # For a hackathon, a plain venv + requirements.txt is lighter and faster; this # is here for parity if conda is your default. All the project's real deps are # pip packages, so most of the work happens in the pip: block below. name: opendoor channels: - conda-forge - defaults dependencies: - python=3.11 - pip - pip: # Core: the planner (Anthropic centerpiece) - anthropic # Server + frontend host - fastapi - uvicorn[standard] - jinja2 - python-multipart # Browse leg (Browserbase) — pick your harness in Claude Code - playwright # then run: playwright install # - stagehand # uncomment if using Stagehand instead - browserbase # Speak leg (Deepgram) — STT intake + TTS spoken gate confirmations - deepgram-sdk # Observability seam (Sentry) — no-op unless OPENDOOR_SENTRY=1 - sentry-sdk # Dev / test - pytest - httpx # After first successful install + run, freeze for reproducibility: # pip freeze > requirements.lock.txt ``` ### tests/test_speak.py ```python """ Speak-leg tests — offline only, no Deepgram calls (no credits spent). Covers backend selection by env and the text-recording execute path. The live TTS/STT round-trip is verified manually against the real key (it costs credits), so it's deliberately not in the unit suite. """ from __future__ import annotations from agent.contracts import Effector, Risk, Step from effectors.mock import MockSpeakBackend from effectors.speak import DeepgramSpeakBackend, build_speak_backend def _say(text: str) -> Step: return Step(effector=Effector.SPEAK, intent="speak a line", args={"action": "say", "text": text}, risk=Risk.SAFE) def test_build_speak_backend_defaults_to_mock(monkeypatch): monkeypatch.delenv("OPENDOOR_SPEAK", raising=False) assert isinstance(build_speak_backend(), MockSpeakBackend) def test_build_speak_backend_real_needs_key(monkeypatch): monkeypatch.setenv("OPENDOOR_SPEAK", "real") monkeypatch.delenv("DEEPGRAM_API_KEY", raising=False) # No key -> still mock (don't construct a Deepgram client that would fail). assert isinstance(build_speak_backend(), MockSpeakBackend) def test_build_speak_backend_real_with_key(monkeypatch): monkeypatch.setenv("OPENDOOR_SPEAK", "real") monkeypatch.setenv("DEEPGRAM_API_KEY", "test-key") assert isinstance(build_speak_backend(), DeepgramSpeakBackend) def test_deepgram_execute_records_text_without_network(monkeypatch): """execute() only records what to say -- no Deepgram call -- so the dispatch loop never makes network calls; audio is rendered separately via /api/tts.""" monkeypatch.setenv("DEEPGRAM_API_KEY", "test-key") backend = DeepgramSpeakBackend() step = backend.execute(_say("Your refill is confirmed.")) assert step.result["spoken"] == "Your refill is confirmed." # The gate hook returns the question text too, no network. assert backend.speak("Proceed?")["question"] is True ``` ### tests/test_store.py ```python """ Store tests — offline, isolated to a tmp file (no API, no shared state). Locks the urgency ordering and the seed/update behavior the errands queue and history depend on. """ from __future__ import annotations import time from agent import store def _isolate(tmp_path, monkeypatch): monkeypatch.setattr(store, "_PATH", tmp_path / "state.json") def test_history_newest_first(tmp_path, monkeypatch): _isolate(tmp_path, monkeypatch) store.add_history({"goal": "a", "outcome": "done"}) store.add_history({"goal": "b", "outcome": "declined"}) hist = store.list_history() assert [h["goal"] for h in hist] == ["b", "a"] # newest first def test_tasks_sorted_by_urgency(tmp_path, monkeypatch): _isolate(tmp_path, monkeypatch) now = time.time() store.add_task({"title": "late", "due": now + 10 * 86400}) store.add_task({"title": "soon", "due": now + 1 * 86400}) store.add_task({"title": "nodue", "due": None}) titles = [t["title"] for t in store.list_tasks()] assert titles == ["soon", "late", "nodue"] # soonest due first, no-due last def test_completed_task_sorts_after_pending(tmp_path, monkeypatch): _isolate(tmp_path, monkeypatch) now = time.time() done = store.add_task({"title": "done-one", "due": now + 1 * 86400}) store.add_task({"title": "still-pending", "due": now + 5 * 86400}) store.update_task(done["id"], status="done") tasks = store.list_tasks() assert tasks[0]["title"] == "still-pending" # pending first assert tasks[-1]["status"] == "done" # finished after def test_seed_defaults_is_idempotent(tmp_path, monkeypatch): _isolate(tmp_path, monkeypatch) store.seed_defaults() n_services, n_tasks = len(store.list_services()), len(store.list_tasks()) assert n_services > 0 and n_tasks > 0 store.seed_defaults() # second call must not duplicate assert len(store.list_services()) == n_services assert len(store.list_tasks()) == n_tasks ``` ### effectors/base.py ```python """ Effector registry: the open seam, in code. The planner picks an effector *by name* (Effector enum); dispatch looks the backend up here. Adding a third body later (a robot, a phone-call leg) is one `register()` call plus one backend class -- no if/else chain to edit, nothing in the execution loop changes. That is the "swappable, extensible" promise from contracts.py made concrete. A backend is anything satisfying the EffectorBackend Protocol: it carries a `name: Effector` and an `execute(step) -> Step`. Mock backends (effectors/mock.py) and live backends (effectors/browse.py, effectors/speak.py) are interchangeable here, so the whole pipeline runs offline against mocks and flips to live by registering a different backend -- config, not refactor. """ from __future__ import annotations from agent.contracts import Effector, EffectorBackend, Step class EffectorRegistry: """A name -> backend lookup. Dispatch holds one of these.""" def __init__(self) -> None: self._backends: dict[Effector, EffectorBackend] = {} def register(self, backend: EffectorBackend) -> EffectorBackend: """Register (or replace) the backend for an effector. Returns it, so it doubles as a decorator-ish one-liner at wiring time.""" self._backends[backend.name] = backend return backend def get(self, effector: Effector) -> EffectorBackend: try: return self._backends[effector] except KeyError: raise LookupError( f"No backend registered for effector {effector.value!r}. " f"Registered: {[e.value for e in self._backends]}" ) from None def execute(self, step: Step) -> Step: """Dispatch one step to its backend. The backend is responsible for its own instrumentation (the @instrument_step decorator on execute).""" return self.get(step.effector).execute(step) def close(self) -> None: """Release any backend that holds resources (e.g. the browse leg's live browser). Backends without a close() are left alone. Idempotent.""" for backend in self._backends.values(): closer = getattr(backend, "close", None) if callable(closer): closer() def __contains__(self, effector: Effector) -> bool: return effector in self._backends ``` ### tests/test_browse.py ```python """ Integration test for the live browse leg. Drives a real Chromium through the mock pharmacy portal loaded as a file:// fixture (no server needed), exercising the same flow the demo uses: ensure signed in -> find prescription -> read the real cost -> submit. Skips cleanly if Playwright or its browser binary isn't installed, so the unit suite still runs anywhere. This is the regression anchor for the deep technical leg: if a selector in portal/mock_pharmacy.html drifts, this goes red before the demo does. """ from __future__ import annotations from pathlib import Path import pytest from agent.contracts import Effector, Risk, Step playwright = pytest.importorskip("playwright.sync_api") PORTAL = (Path(__file__).resolve().parent.parent / "portal" / "mock_pharmacy.html").as_uri() def _browser_available() -> bool: try: from playwright.sync_api import sync_playwright with sync_playwright() as pw: b = pw.chromium.launch(headless=True) b.close() return True except Exception: return False pytestmark = pytest.mark.skipif( not _browser_available(), reason="Chromium not installed (run: playwright install chromium)" ) @pytest.fixture def backend(): from effectors.browse import BrowseBackend b = BrowseBackend(portal_url=PORTAL) yield b b.close() def _step(action: str, **args) -> Step: risk = Risk.IRREVERSIBLE if action == "submit_refill" else Risk.SAFE return Step(effector=Effector.BROWSE, intent=action, args={"action": action, **args}, risk=risk) def test_browse_drives_portal_end_to_end(backend): # find prescription (self-ensures sign-in first) found = backend.execute(_step("find_prescription", name="metformin")).result assert found["found"] is True assert "Metformin" in found["name"] # read the real out-of-pocket cost from the DOM refill = backend.execute(_step("check_refill")).result assert refill["copay"] == "$14.00" assert "mail-order" in refill["cost_note"].lower() # submit (the irreversible action) and read the confirmation back confirm = backend.execute(_step("submit_refill")).result assert confirm["confirmation_no"].startswith("BM") assert confirm["cost"] == "$14.00" def test_find_prescription_self_ensures_signin(backend): """The browse leg must work even though the golden plan has no explicit open_portal step -- find_prescription signs in on its own.""" result = backend.execute(_step("find_prescription")).result assert result["found"] is True ``` ### agent/contracts.py ```python """ Core contracts for Open Door. One Claude planner takes a spoken goal, emits a structured Plan, and dispatches each Step to one of two effectors: speak (Deepgram) or browse (Browserbase). The planner is the Anthropic centerpiece; the Plan is a first-class object the frontend renders directly, so the judge watches the reasoning as data. Two design rules: 1. The Plan is serializable and IS the hero UI element. 2. Effectors share one Protocol; nothing costly/irreversible runs without an explicit human yes at the gate -- the care signal, made literal. """ from __future__ import annotations from dataclasses import dataclass, field from enum import Enum from typing import Any, Protocol import time import uuid class Effector(str, Enum): SPEAK = "speak" # Deepgram: voice in/out -- ESSENTIAL because the user can't read a screen. # Built deep on purpose: the spoken gate confirmations are the critical beat, # which is what clears Deepgram's "essential, not tacked on" bar honestly. BROWSE = "browse" # Browserbase: real portal navigation, recovery, stop-before-submit. # The deep technical leg + the Anthropic technical-depth showcase. class Risk(str, Enum): SAFE = "safe" # read-only / reversible: read a page, speak a sentence COSTLY = "costly" # spends money / commits the user: authorize a copay IRREVERSIBLE = "irreversible" # cannot be undone: submit the refill request class StepState(str, Enum): PENDING = "pending" RUNNING = "running" AWAITING_CONFIRM = "awaiting_confirm" # gated -- UI shows red, waits + SPEAKS the question DONE = "done" FAILED = "failed" SKIPPED = "skipped" @dataclass class Step: effector: Effector intent: str # "Request a 90-day refill of metformin on the portal" args: dict[str, Any] = field(default_factory=dict) risk: Risk = Risk.SAFE rationale: str = "" # WHY this effector -- shown to the judge id: str = field(default_factory=lambda: uuid.uuid4().hex[:8]) state: StepState = StepState.PENDING result: Any = None error: str | None = None @property def gated(self) -> bool: return self.risk in (Risk.COSTLY, Risk.IRREVERSIBLE) def to_dict(self) -> dict[str, Any]: return { "id": self.id, "effector": self.effector.value, "intent": self.intent, "args": self.args, "risk": self.risk.value, "rationale": self.rationale, "state": self.state.value, "result": self.result, "error": self.error, "gated": self.gated, } @dataclass class Plan: goal: str steps: list[Step] = field(default_factory=list) id: str = field(default_factory=lambda: uuid.uuid4().hex[:8]) created_at: float = field(default_factory=time.time) def current(self) -> Step | None: for s in self.steps: if s.state not in (StepState.DONE, StepState.SKIPPED, StepState.FAILED): return s return None def to_dict(self) -> dict[str, Any]: return { "id": self.id, "goal": self.goal, "created_at": self.created_at, "steps": [s.to_dict() for s in self.steps], } class EffectorBackend(Protocol): name: Effector def execute(self, step: Step) -> Step: ... ``` [14 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]