vasceannie/guide
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

.

Browse Source
This commit is contained in:
Travis Vasceannie
2025-11-22 00:54:25 +00:00
parent abca5b612a
commit d54eeeaeeb
29 changed files with 919 additions and 609 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
config/hosts.yaml
View File

@@ -1,7 +1,8 @@
hosts:
- id: desktop
demo-cdp:
kind: cdp
host: 192.168.50.185
cdp_port: 9223
- id: laptop
host: 192.168.50.152
cdp_port: 9222
port: 9223
headless-local:
kind: headless
browser: chromium

879
docs/spec.md
View File

@@ -1,498 +1,595 @@
Here’s a PRD you can hand to Future‑You (or a code assistant) and not want to scream at it later.
Absolutely — this is a great moment to “lock in” your error + utilities + GraphQL story while the project is still small.
Below is an **updated spec** you can paste straight into your code assistant. It *extends* your current project state to cover:
* Centralized error handling
* Shared utilities
* Direct GraphQL operations for background tasks
---
# PRD: Raindrop Demo Automation Service (FastAPI + Playwright/CDP)
# Spec Update: Errors, Utilities, and Direct GraphQL Operations
* **Owner:** You
* **Service Name (working):** `raindrop-demo-automation`
* **Primary Host:** `192.168.50.151` (homelab server)
* **Clients:** Stream Deck (HTTP actions), CLI tools, or other automation
## 0. Current Layout (for context)
---
## 1. Background & Problem
You run live Raindrop demos and already:
* Launch Chrome with **CDP (remote debugging)** enabled.
* Manually drive the UI, assisted by **Stream Deck** macros today.
* Have **Python/Playwright** scripts that can attach to that browser and perform actions.
Limitations right now:
* Scripts run **locally**, tied to one machine.
* There isn’t a **single, extensible service** orchestrating “demo actions” (fill forms, advance steps, etc.).
* Strings, selectors, and values are **hardcoded and scattered** across scripts, making it easy for a code assistant to duplicate or diverge.
You want:
* A **FastAPI service** on your homelab (`192.168.50.151`) that exposes HTTP endpoints for demo actions.
* Config that supports **multiple demo machines** (desktop, laptop via VPN) and is easy to switch.
* A **modular architecture**: small packages, low line length, clear facades, zero magic values in-line.
* All **input strings / UI texts / selectors** in a **single package** to avoid duplication and make future tooling straightforward.
---
## 2. Goals
### Functional Goals
1. **Trigger demo actions via HTTP**
* Stream Deck calls endpoints like `POST /actions/fill-intake` or `POST /actions/add-suppliers`.
* Actions attach to an existing CDP-enabled browser and run against the current Raindrop tab.
2. **Support multiple browser hosts**
* Each demo machine (desktop, laptop via VPN) is modeled as a **Browser Host** with:
* ID (e.g., `desktop`, `laptop`, `lab-vm`)
* Host/IP (e.g., `192.168.50.100`)
* CDP port (e.g., `9222`)
* The API caller can specify which host to target per request (query param or payload).
3. **Encapsulate all demo strings and selectors**
* No “free-floating” strings in action code for:
* Input text (demo narratives, supplier names, event names).
* UI labels / button text / placeholder strings.
* CSS/XPath selectors.
* These live in a central `strings` package.
4. **Extensible actions architecture**
* Easy to add new actions (e.g., “create 3-bids-and-buy event”, “run 3-way match”) without copy-paste.
* A registry/facade manages all actions by ID.
### Non‑Goals
* No UI or dashboard for now (everything via HTTP / Stream Deck).
* No multi-tenant security model beyond basic network trust.
* No scheduling / long-running workflows (actions are short-lived scripts).
---
## 3. Users & Usage
### Primary User
* **Demo Host (you)**
Use Stream Deck to trigger HTTP calls to the service while driving a Raindrop demo.
### Usage Scenarios
1. **Simple Intake Fill**
* You navigate to the Raindrop intake screen.
* Hit a Stream Deck button that calls `POST /actions/fill-intake-basic` targeting your current machine.
* The service attaches to your browser and fills out description + moves to the next step.
2. **Three Suppliers Auto-Add**
* On the “Suppliers” step of a sourcing event, you hit a Stream Deck button.
* The service adds three predefined suppliers.
3. **Multi-host Setup**
* Some days you demo from desktop; some days from laptop via VPN.
* You switch the Stream Deck config or action payload to target `browser_host_id=laptop`.
---
## 4. High-Level Architecture
### Components
1. **FastAPI Application (`app`)**
* Exposes REST endpoints.
* Does auth (if/when needed), routing, validation, and returns structured responses.
2. **Config & Settings (`app.core.config`)**
* Uses Pydantic `BaseSettings` + optional YAML for:
* Known browser hosts (id, host, port).
* Default browser host.
* Raindrop tenant URL.
* No hardcoded values in code; read from config/env.
3. **Strings Package (`app.strings`)**
* Contains **all**:
* Demo text (descriptions, comments).
* UI labels & button names.
* CSS/Playwright selectors.
* Structured by domain (intake, sourcing, payables).
4. **Browser Connector (`app.browser`)**
* Encapsulates Playwright CDP connections.
* Provides a `BrowserClient` or `BrowserFacade`:
* `connect(browser_host)` → returns a connection.
* `get_raindrop_page()` → returns a `Page` to act on (e.g., the tab whose URL matches Raindrop host).
* This is the only layer that knows about CDP endpoints.
5. **Action Framework (`app.actions`)**
* Base `DemoAction` protocol/class:
* `id: str`
* `run(page, context) -> ActionResult`
* Each domain package implements focused actions:
* `intake.py` (fill forms, advance steps).
* `sourcing.py` (add suppliers, configure event).
* `navigation.py` (jump to certain pages).
* An `ActionRegistry` maps action IDs → action objects.
6. **Domain Models (`app.domain`)**
* Typed models with Pydantic for:
* `ActionRequest`, `ActionResponse`.
* `BrowserHost` and config structures.
* `ActionContext` (host id, session info, optional parameters).
---
## 5. Detailed Design
### 5.1 Directory Layout (Python Package)
Example layout (you can tweak):
You currently have:
```text
app/
__init__.py
main.py # FastAPI app instance
api/
__init__.py
routes_actions.py # /actions endpoints
routes_config.py # optional: expose config/browser hosts
routes_health.py # /healthz
core/
__init__.py
config.py # Pydantic settings and config loading
logging.py # logging setup
domain/
__init__.py
models.py # ActionRequest, ActionResponse, BrowserHost, ActionContext
enums.py # Action IDs, maybe host status enums
browser/
__init__.py
client.py # BrowserFacade/BrowerClient implementation
page_selector.py # logic to pick the correct Raindrop tab
actions/
__init__.py
base.py # DemoAction interface, ActionRegistry
intake.py # intake-related actions
sourcing.py # sourcing-related actions
navigation.py # navigation actions
strings/
__init__.py
selectors.py # all CSS/xpath selectors
labels.py # visible UI labels/button names
demo_texts.py # all pre-baked text content
config/
strings.yaml # optional external strings source
hosts.yaml # browser host definitions (desktop, laptop, etc.)
.
├── config
│ └── hosts.yaml
└── src
└── guide
├── app
│ ├── actions
│ ├── api
│ ├── browser
│ ├── core
│ ├── domain
│ ├── strings
│ ├── __init__.py
│ ├── main.py
├── __init__.py
└── main.py
```
**Guideline:**
Each module should be small, focused, and under ~200–300 lines. If a module grows, split it further (e.g., `intake_create.py`, `intake_approve.py`).
Plus the more detailed subpackages already defined in the prior spec (auth, personas, etc.).
---
### 5.2 Config & Settings
## 1. High-Level Requirements to Append to Project State
Use Pydantic `BaseSettings` to load:
Append the following to the project’s PRD / requirements:
* Environment variables (for secrets, host IP).
* YAML/JSON for structured config (hosts, string groups).
1. **Centralized Error Model**
Example conceptual model:
* All non-trivial errors should be represented by a small set of **typed exceptions** rooted in a common base `GuideError`.
* Error mapping to HTTP responses and `ActionResponse` objects must be centralized; individual actions should raise `GuideError` subclasses, not build HTTP responses themselves.
* Error codes must be **stable, string-based enums** (e.g., `BROWSER_CONNECT_FAILED`, `GRAPHQL_ERROR`, `AUTH_MFA_FAILED`) to make it easy for clients and MCP tools to reason about failures.
2. **Utilities Layer**
* Introduce a `utils` package for common cross-cutting concerns:
* Correlation ID generation.
* Timed execution & simple metrics.
* Retry/backoff helpers for flaky operations (CDP attachment, GraphQL, MFA fetch).
* Small helpers for safe env access, JSON handling, etc.
* Actions, auth, browser, and GraphQL code must use these utilities rather than re-implementing ad-hoc logic.
3. **Direct GraphQL Operations**
* For some **background tasks**, it must be possible to operate directly against Raindrop’s **GraphQL API** instead of the browser:
* Example: fetch/update request status, add suppliers, update metadata, etc.
* GraphQL interaction must be encapsulated in a dedicated `raindrop` service layer:
* A low-level GraphQL client abstraction.
* Domain-specific operations (intake, sourcing, etc.) calling that client.
* No raw HTTP or inline GraphQL queries are allowed in actions; they must go through this service layer (and queries/operation names must live in `strings`).
4. **Persona-Aware Backend Operations**
* GraphQL operations should be able to run **as a persona** when needed (e.g., Supplier vs Buyer), using persona-specific auth (tokens/cookies/headers).
* If GraphQL uses a service account instead of persona-based auth, that must still be expressed explicitly in the persona or configuration (e.g., a special `SERVICE` persona or a `service_auth` section).
5. **Consistent Error Surfacing**
* Both browser-based actions and GraphQL-based background tasks must surface errors in a consistent structure:
* Machine-readable `error_code`.
* Human-readable `message`.
* Optional `details` for debugging.
* FastAPI routes must convert exceptions to this structure consistently.
---
## 2. New Packages & Files to Add
### 2.1 `errors` package
Create:
```text
src/guide/app/errors/
__init__.py
exceptions.py
handlers.py
```
#### `exceptions.py`
Define a common base and key subclasses:
```python
# app/core/config.py
from pydantic import BaseSettings
from typing import Dict
from typing import Any, Optional
class BrowserHostConfig(BaseSettings):
id: str
host: str
cdp_port: int
class GuideError(Exception):
code: str = "UNKNOWN_ERROR"
class AppSettings(BaseSettings):
raindrop_base_url: str
default_browser_host_id: str
browser_hosts: Dict[str, BrowserHostConfig] # keyed by id
def __init__(self, message: str, *, details: Optional[dict[str, Any]] = None):
super().__init__(message)
self.message = message
self.details = details or {}
class Config:
env_prefix = "RAINDROP_DEMO_"
# Optionally load from config/hosts.yaml
class ConfigError(GuideError):
code = "CONFIG_ERROR"
class BrowserConnectionError(GuideError):
code = "BROWSER_CONNECT_FAILED"
class PersonaError(GuideError):
code = "PERSONA_ERROR"
class AuthError(GuideError):
code = "AUTH_ERROR"
class MfaError(GuideError):
code = "AUTH_MFA_FAILED"
class ActionExecutionError(GuideError):
code = "ACTION_EXECUTION_FAILED"
class GraphQLTransportError(GuideError):
code = "GRAPHQL_TRANSPORT_ERROR"
class GraphQLOperationError(GuideError):
code = "GRAPHQL_OPERATION_ERROR"
```
**Requirements:**
* No raw IPs or ports in code. IP `192.168.50.151` is used at deployment level (e.g., uvicorn bind host), not in business logic.
* Changing default host or adding a laptop host should mean:
* Any domain-specific error should subclass `GuideError` and set a stable `code`.
* Lower layers (browser, auth, GraphQL, actions) raise these exceptions; they **never** construct FastAPI responses themselves.
* Update `hosts.yaml` and/or env var.
* Restart service, no code changes.
#### `handlers.py`
Implement functions to integrate with FastAPI:
```python
from fastapi import Request
from fastapi.responses import JSONResponse
from .exceptions import GuideError
def guide_error_handler(request: Request, exc: GuideError) -> JSONResponse:
return JSONResponse(
status_code=400, # can refine based on exc.code later
content={
"status": "error",
"error_code": exc.code,
"message": exc.message,
"details": exc.details or {},
},
)
def unhandled_error_handler(request: Request, exc: Exception) -> JSONResponse:
return JSONResponse(
status_code=500,
content={
"status": "error",
"error_code": "UNHANDLED_EXCEPTION",
"message": "An unexpected error occurred",
},
)
```
Then, in `src/guide/app/main.py`, register these handlers with the FastAPI app.
---
### 5.3 Strings Package
### 2.2 `utils` package
**Objective:** Any textual thing that might be typed into or read from the UI lives here.
Create:
Submodules:
```text
src/guide/app/utils/
__init__.py
ids.py
timing.py
retry.py
env.py
```
1. `selectors.py`
#### `ids.py`
* All selectors used by Playwright:
* Provide correlation ID / job ID helpers:
* e.g., `INTAKE_DESCRIPTION_FIELD`, `BUTTON_NEXT`, `SUPPLIER_SEARCH_INPUT`.
* Prefer centrally named constants:
```python
import uuid
* `selectors.INTAKE.DESCRIPTION_FIELD`
* `selectors.SOURCING.SUPPLIER_SEARCH_INPUT`
* Keep selectors DRY, referenced by actions.
def new_correlation_id() -> str:
return str(uuid.uuid4())
2. `labels.py`
def new_job_id(prefix: str = "job") -> str:
return f"{prefix}-{uuid.uuid4()}"
```
* Just the user-visible string labels: e.g., `"Next"`, `"Submit"`, `"Suppliers"`.
* Some selectors may be derived from labels (e.g., Playwright `get_by_role("button", name=labels.NEXT_BUTTON)`).
#### `timing.py`
3. `demo_texts.py`
* Provide a decorator/context manager for timing:
* All “scripted” text you want to appear in demos:
```python
import time
from collections.abc import Callable
from functools import wraps
from typing import TypeVar
* Intakes (“500 tons of conveyor belts…”).
* Event names.
* Supplier names.
* Grouped by scenario:
T = TypeVar("T")
* `INTAKE.CONVEYOR_BELT_REQUEST`
* `SOURCING.THREE_BIDS_EVENT_NAME`
* `SUPPLIERS.DEFAULT_TRIO = ["Demo Supplier A", "Demo Supplier B", "Demo Supplier C"]`
def timed(operation: str, logger) -> Callable[[Callable[..., T]], Callable[..., T]]:
def decorator(fn: Callable[..., T]) -> Callable[..., T]:
@wraps(fn)
def wrapper(*args, **kwargs) -> T:
start = time.perf_counter()
try:
return fn(*args, **kwargs)
finally:
duration_ms = (time.perf_counter() - start) * 1000.0
logger.info("operation=%s duration_ms=%.2f", operation, duration_ms)
return wrapper
return decorator
```
Optional: load from external `strings.yaml` but always surfaced through `app.strings` to keep a single import point.
#### `retry.py`
**Rule:**
Actions must **never** use raw strings for content or selectors directly; they import from `app.strings`.
* Provide a simple, configurable retry helper for flaky operations (CDP attach, GraphQL, MFA fetch):
```python
import time
from collections.abc import Callable
from typing import TypeVar
T = TypeVar("T")
def retry(
fn: Callable[[], T],
*,
retries: int = 3,
delay_seconds: float = 0.5,
backoff_factor: float = 2.0,
on_error: Callable[[Exception, int], None] | None = None,
) -> T:
attempt = 0
current_delay = delay_seconds
while True:
try:
return fn()
except Exception as exc:
attempt += 1
if attempt > retries:
raise
if on_error:
on_error(exc, attempt)
time.sleep(current_delay)
current_delay *= backoff_factor
```
Actions, browser client, and GraphQL client can use this instead of custom loops.
#### `env.py`
* Safe access to env vars (and central place to throw `ConfigError`):
```python
import os
from ..errors import ConfigError
def require_env(name: str) -> str:
value = os.getenv(name)
if not value:
raise ConfigError(f"Missing required environment variable {name}")
return value
```
---
### 5.4 Browser Connector
## 3. GraphQL Integration (Direct API Operations)
`app.browser.client.BrowserClient` (or `BrowserFacade`):
### 3.1 New `raindrop` service package
Create:
```text
src/guide/app/raindrop/
__init__.py
graphql.py
operations/
__init__.py
intake.py
sourcing.py
```
#### `graphql.py` – low-level client
Responsibilities:
* Resolve `BrowserHost` config by id.
* Connect to CDP endpoint:
* Build and send GraphQL HTTP requests.
* Attach appropriate auth (persona-based token, service token, etc.).
* Parse responses and raise typed `GraphQL*` errors if needed.
* `http://{host}:{cdp_port}` via `playwright.chromium.connect_over_cdp`.
* Resolve the correct **Raindrop page**:
**Key design points:**
* Prefer a page whose URL contains `settings.raindrop_base_url`.
* If multiple, pick the last active or last created (simple heuristic).
* Provide a simple interface to actions:
* Endpoint URL and any static headers come from `core.config` / env, not hardcoded.
* Query strings and operation names live in `strings` (see 3.3).
* `get_page()` returns a Playwright `Page`.
* It should handle errors gracefully: no host, cannot connect, no pages.
Actions use **only** this abstraction; they never touch raw CDP URLs.
---
### 5.5 Action Framework
**Base interface** (`app.actions.base`):
Example shape:
```python
class DemoAction(Protocol):
id: str
from typing import Any
import httpx
def run(self, page: Page, context: ActionContext) -> ActionResult:
...
from ..core.config import AppSettings
from ..errors import GraphQLTransportError, GraphQLOperationError
from ..personas.models import DemoPersona
class GraphQLClient:
def __init__(self, settings: AppSettings):
self._settings = settings
async def execute(
self,
*,
query: str,
variables: dict[str, Any] | None,
persona: DemoPersona | None,
operation_name: str | None = None,
) -> dict[str, Any]:
url = self._settings.raindrop_graphql_url # from config/env
headers = self._build_headers(persona)
async with httpx.AsyncClient(timeout=10.0) as client:
try:
resp = await client.post(
url,
json={
"query": query,
"variables": variables or {},
"operationName": operation_name,
},
headers=headers,
)
except httpx.HTTPError as exc:
raise GraphQLTransportError(
f"Transport error calling GraphQL: {exc}"
) from exc
if resp.status_code >= 400:
raise GraphQLTransportError(
f"HTTP {resp.status_code} from GraphQL",
details={"status_code": resp.status_code, "body": resp.text},
)
data = resp.json()
if "errors" in data and data["errors"]:
raise GraphQLOperationError(
"GraphQL operation failed",
details={"errors": data["errors"]},
)
return data.get("data", {})
def _build_headers(self, persona: DemoPersona | None) -> dict[str, str]:
# TODO: adapt based on persona; for now placeholder
headers: dict[str, str] = {"Content-Type": "application/json"}
# Option A: service account
# Option B: persona-specific token from config/env
return headers
```
* `ActionContext` includes:
You can add retries here using `utils.retry.retry` if desired.
* `browser_host_id: str`
* Optional parameters (e.g., override event name).
* Correlation id for logging.
#### `operations/intake.py` & `operations/sourcing.py`
**Action Registry:**
* Provide **higher-level functions** like:
* Maintains a mapping of `action_id → DemoAction` instance.
* Prevents duplication: all action IDs are declared in **one** place.
* Provides methods:
```python
async def get_intake_request(client: GraphQLClient, persona: DemoPersona, request_id: str) -> IntakeRequest: ...
async def create_sourcing_event(...): ...
```
* `get(action_id: str) -> DemoAction`
* `list() -> List[ActionMetadata]`
* They:
**Example Actions:**
* Import the raw query strings from `strings`.
* Call `GraphQLClient.execute`.
* Map the result into typed domain models (or simple dicts for now).
* `FillIntakeBasicAction`
* Uses `selectors.INTAKE.DESCRIPTION_FIELD`.
* Uses `demo_texts.INTAKE.CONVEYOR_BELT_REQUEST`.
* Calls `page.fill()` + `page.click()`.
* `AddThreeSuppliersAction`
* Uses `demo_texts.SUPPLIERS.DEFAULT_TRIO`.
Each action file should be short (one or a few related actions) and import everything from `strings` + `browser`.
* Actions then call these operations instead of crafting GraphQL queries.
---
### 5.6 FastAPI Layer & API Design
### 3.2 Config additions for GraphQL
#### Endpoints
In `src/guide/app/core/config.py`:
1. `GET /healthz`
* Extend `AppSettings` with:
* Returns `{ "status": "ok" }`.
```python
class AppSettings(BaseSettings):
# existing fields...
raindrop_graphql_url: str # e.g. https://tenant.raindropapp.com/graphql
```
2. `GET /actions`
* Returns a list of available actions:
* `id`, `description`, `category` (e.g., `intake`, `sourcing`).
3. `POST /actions/{action_id}/execute`
* **Request body:**
* `browser_host_id` (optional → use default).
* `params` (optional dict, action-specific).
* **Behavior:**
* Look up host; create `ActionContext`.
* Use `BrowserClient` to connect to host and get Raindrop page.
* Run the action; return result status + optional metadata.
* **Response body:**
* `status` (`ok` / `error`).
* `action_id`.
* `browser_host_id`.
* `details` (optional).
4. `GET /config/browser-hosts` (optional)
* Returns configured browser hosts and default host.
**Telemetry & Error Handling:**
* Every request logs:
* `action_id`, `browser_host_id`, `correlation_id`, `duration_ms`, `result`.
* In case of error, return `4xx/5xx` with structured error JSON:
* e.g., `{"error": "BROWSER_CONNECT_FAILED", "message": "Cannot connect to 192.168.50.100:9222"}`.
* This value should be set via env or a config file; **never** hardcoded.
---
## 6. Networking & Deployment
### 3.3 `strings` for GraphQL queries
* **Backend Location:** FastAPI app running on your homelab server (`192.168.50.151`).
Add a new folder:
* **Bind Address:** `0.0.0.0` (so local network & VPN clients can hit it).
```text
src/guide/app/strings/graphql/
__init__.py
intake.py
sourcing.py
```
* **Port:** configurable via env (e.g., default 8000).
Each file contains string constants for queries/mutations and (optionally) operation names, for example:
* **Demo Machines:**
```python
# intake.py
GET_INTAKE_REQUEST = """
query GetIntakeRequest($id: ID!) {
intakeRequest(id: $id) {
id
title
status
# ...
}
}
"""
* Desktop/laptop each runs Chrome with:
CREATE_INTAKE_REQUEST = """
mutation CreateIntakeRequest($input: IntakeRequestInput!) {
createIntakeRequest(input: $input) {
id
# ...
}
}
"""
```
* `--remote-debugging-port=9222`
* Known IP or DNS reachable from `192.168.50.151`.
* `hosts.yaml` defines each as a `BrowserHost`.
**Rules:**
* **Steam Deck:**
* Configured to call e.g.:
* `http://192.168.50.151:8000/actions/fill-intake-basic/execute?browser_host_id=desktop`
* or with JSON body specifying host.
* No inline GraphQL strings in actions or service code.
* All GraphQL query strings and operation names live in `strings/graphql/*`.
---
## 7. Extensibility & Code-Assistant Friendliness
## 4. Error Handling Integration Across Layers
### Extensibility
### 4.1 Domain models & ActionResponse
* To add a new action:
In `src/guide/app/domain/models.py`:
1. Add any new strings/selectors to `app.strings`.
2. Implement a small `DemoAction` in the right `actions/*` module.
3. Register it in the `ActionRegistry`.
4. Optionally expose documentation via `/actions`.
* Extend `ActionResponse` with error fields:
* To support a new demo machine:
```python
class ActionStatus(str, Enum):
SUCCESS = "success"
ERROR = "error"
* Add `BrowserHostConfig` entry in `hosts.yaml`.
* Restart service.
class ActionResponse(BaseModel):
status: ActionStatus
action_id: str
correlation_id: str
result: dict[str, Any] | None = None
error_code: str | None = None
message: str | None = None
details: dict[str, Any] | None = None
```
### Code-Assistant Guardrails
* When an action completes successfully, `status=SUCCESS` and `error_*` fields are `None`.
* When it fails, `status=ERROR` and `error_code/message/details` are populated from a `GuideError`.
* All core primitive operations (connect to browser, select Raindrop page, use selectors, pick texts) are **centralized**:
### 4.2 FastAPI route error handling
* `browser.client` for CDP.
* `strings` for text and selectors.
* `actions.base` & `ActionRegistry` for action definitions.
* PRD requirement: modules must be small and self-contained; code assistants should:
In `src/guide/app/api/routes/actions.py`:
* Prefer using existing `BrowserClient` instead of re-implementing CDP logic.
* Prefer using `strings` instead of writing raw strings.
* Use `ActionRegistry` for action lookup.
* Ensure that `ActionResponse` is the canonical shape returned for `/actions/*`.
* Either:
* Let `GuideError` bubble up to the global FastAPI exception handlers, or
* Catch `GuideError` in the route, create `ActionResponse(status=ERROR, ...)`, and return that.
Pick one approach and stick with it; do not mix ad-hoc `JSONResponse`s and `ActionResponse` everywhere.
---
## 8. Security & Safety
## 5. How Actions Decide Between Browser vs GraphQL
* Runs on trusted internal network/VPN.
* Optional enhancements:
Add these architectural guidelines (comments/docs for the assistant):
* Simple API key header for Stream Deck / other clients.
* IP allowlist (only allow local subnet/VPN).
* No credentials in code:
1. **Browser-first for interactive flows**
* Any secrets (e.g., if you *ever* login from the service) stored in env, not in repo.
* Actions should **never** perform destructive or irreversible operations in Raindrop unless explicitly designed and named that way (e.g., `submit-event`, `delete-draft`).
* Actions that are meant to be visible during demos (live UI clicking, form filling) should use the **browser layer** (CDP/Playwright).
* They:
* Use `BrowserClient.get_demo_page(...)`.
* Expect persona login to be handled by `auth.session.ensure_persona`.
2. **GraphQL-first for purely backend tasks**
* Background actions that don’t need UI (e.g., polling statuses, “prepare data before demo”, mass-updating metadata) should call the **GraphQL service** instead of automating the UI.
* These actions:
* Get a `DemoPersona` for context (or a service persona).
* Use `raindrop.graphql.GraphQLClient` + `raindrop.operations.*`.
3. **Hybrid actions**
* Some scenarios may need both:
* Use GraphQL to create or prepare data.
* Use browser actions to show the result in the demo.
* Hybrid actions must **still** route GraphQL calls through `raindrop.operations`, and browser calls through `BrowserClient`.
4. **Error consistency**
* Whether the failure happens in CDP attachment, MFA login, or GraphQL:
* Lower-level code throws the appropriate `GuideError` subclass.
* The route or job layer converts that into a consistent `ActionResponse` or job status.
---
## 9. Acceptance Criteria
## 6. Concrete TODOs for the Code Assistant
1. **End-to-end happy path:**
You can give your agent this checklist:
* From desktop, with Chrome CDP running, you trigger `fill-intake-basic` via Stream Deck and the intake form is filled and advanced.
1. **Add `src/guide/app/errors/{exceptions.py,handlers.py}` and wire handlers into `app.main`’s FastAPI app.**
2. **Add `src/guide/app/utils/{ids.py,timing.py,retry.py,env.py}` and update:**
2. **Multi-host support:**
* Logging to include correlation IDs where appropriate.
* Any flaky CDP attach logic or future GraphQL calls to optionally use `retry`.
3. **Extend `core/config.py` and `config/hosts.yaml` as previously specified AND add `raindrop_graphql_url` to `AppSettings`.**
4. **Create `src/guide/app/raindrop/graphql.py` and `src/guide/app/raindrop/operations/{intake.py,sourcing.py}` using the patterns above.**
5. **Create `src/guide/app/strings/graphql/{intake.py,sourcing.py}` to store GraphQL query strings.**
6. **Update `domain/models.py` to include error fields in `ActionResponse` and ensure the routes use it consistently.**
7. **Refactor existing (and future) code to:**
* You can successfully run the same action against `desktop` and `laptop` by only changing `browser_host_id`.
3. **Strings centralization:**
* A grep for known demo text (`"conveyor belts"`, `"Demo Supplier A"`) returns only `app/strings/*` files.
* No selectors appear directly in `actions/*`.
4. **Small module sizes:**
* No file exceeds an agreed limit (e.g. 300 LoC), except possibly `domain/models.py` if that’s acceptable.
5. **Extensibility check:**
* You can add a new action (e.g., `open-intake-dashboard`) following documented steps without touching existing actions, and it appears in `/actions`.
* Use `GuideError` subclasses instead of generic exceptions where appropriate.
* Avoid inline strings for UI selectors, demo text, or GraphQL queries.
---
You can drop this whole spec into your agent as the “next iteration instructions” and let it build out the `errors`, `utils`, and `raindrop` layers in a way that stays consistent with your existing modular architecture.
---
## 3. Engineering & Code Style Standards (add-on)
These apply to all new work in this repo.
### Python & typing
* Target Python **3.12+**; no `__future__` imports.
* Use **PEP 604 unions** (`str | None`) and **built-in generics** (`list[str]`, `dict[str, str]`); avoid `typing.List/Dict`.
* Do not use `typing.Any`; model shapes explicitly (e.g., `dict[str, str]`, `TypedDict`, `Protocol`, or data classes/Pydantic models).
* Avoid `# type: ignore`; prefer refactoring to satisfy type checkers.
* Use `Literal` for constrained status/enum-like strings.
### Imports and organization
* Import order: stdlib → third-party → internal.
* No wildcard imports. Keep line length ≤120 chars.
* Keep domain packages segmented (e.g., `app/actions/<domain>/`, `app/strings/<domain>/`); add submodules rather than expanding single files.
### Pydantic & settings
* Use **Pydantic v2** and `pydantic-settings`; no v1 compat shims.
* All configurable values flow through `AppSettings`; never hardcode hosts, ports, or URLs in business logic.
### Actions and strings
* Actions must import selectors/text from `app.strings`; no inline UI strings/selectors.
* Split actions once they exceed ~150 LoC to keep files focused.
### Error handling and logging
* Prefer structured errors (`{"error": "CODE", "message": "...", "correlation_id": "..."}`).
* Log correlation IDs on entry/exit for actions and browser connections when available.
### Testing & validation
* Bias toward unit/contract tests; when testing Playwright-dependent logic, mock the browser rather than driving real CDP in CI.
* Keep `python -m compileall src/guide` green as a minimum sanity gate.

1
pyproject.toml
View File

@@ -12,6 +12,7 @@ dependencies = [
"python-dotenv>=1.2.1",
"pyyaml>=6.0.2",
"uvicorn>=0.30.6",
"httpx>=0.27.0",
]
[dependency-groups]

0
src/guide/app/__init__.py
View File

20
src/guide/app/actions/__init__.py
View File

@@ -1,20 +0,0 @@
from guide.app.actions.base import ActionRegistry, DemoAction
from guide.app.actions.intake import FillIntakeBasicAction
from guide.app.actions.sourcing import AddThreeSuppliersAction
def default_registry() -> ActionRegistry:
actions: list[DemoAction] = [
FillIntakeBasicAction(),
AddThreeSuppliersAction(),
]
return ActionRegistry(actions)
__all__ = [
"ActionRegistry",
"DemoAction",
"default_registry",
"FillIntakeBasicAction",
"AddThreeSuppliersAction",
]

18
src/guide/app/actions/base.py
View File

@@ -1,14 +1,17 @@
from typing import Iterable, Protocol
from collections.abc import Iterable
from typing import ClassVar, Protocol, runtime_checkable
from playwright.async_api import Page
from guide.app.domain.models import ActionContext, ActionMetadata, ActionResult
from guide.app import errors
from guide.app.models.domain import ActionContext, ActionMetadata, ActionResult
@runtime_checkable
class DemoAction(Protocol):
id: str
description: str
category: str
id: ClassVar[str]
description: ClassVar[str]
category: ClassVar[str]
async def run(self, page: Page, context: ActionContext) -> ActionResult: ...
@@ -19,9 +22,12 @@ class ActionRegistry:
def __init__(self, actions: Iterable[DemoAction]):
self._actions: dict[str, DemoAction] = {action.id: action for action in actions}
def register(self, action: DemoAction) -> None:
self._actions[action.id] = action
def get(self, action_id: str) -> DemoAction:
if action_id not in self._actions:
raise KeyError(f"Unknown action '{action_id}'")
raise errors.ActionExecutionError(f"Unknown action '{action_id}'")
return self._actions[action_id]
def list_metadata(self) -> list[ActionMetadata]:

20
src/guide/app/actions/intake/basic.py
View File

@@ -1,16 +1,22 @@
from playwright.async_api import Page
from typing import ClassVar, override
from guide.app.actions.base import DemoAction
from guide.app.domain.models import ActionContext, ActionResult
from guide.app.strings import demo_texts, selectors
from guide.app.models.domain import ActionContext, ActionResult
from guide.app.strings.service import strings
class FillIntakeBasicAction(DemoAction):
id = "fill-intake-basic"
description = "Fill the intake description and advance to the next step."
category = "intake"
id: ClassVar[str] = "fill-intake-basic"
description: ClassVar[str] = "Fill the intake description and advance to the next step."
category: ClassVar[str] = "intake"
@override
async def run(self, page: Page, context: ActionContext) -> ActionResult:
await page.fill(selectors.INTAKE.DESCRIPTION_FIELD, demo_texts.INTAKE.CONVEYOR_BELT_REQUEST)
await page.click(selectors.INTAKE.NEXT_BUTTON)
description_val = strings.text("INTAKE", "CONVEYOR_BELT_REQUEST")
if not isinstance(description_val, str):
raise ValueError("INTAKE.CONVEYOR_BELT_REQUEST must be a string")
await page.fill(strings.selector("INTAKE", "DESCRIPTION_FIELD"), description_val)
await page.click(strings.selector("INTAKE", "NEXT_BUTTON"))
return ActionResult(details={"message": "Intake filled"})

25
src/guide/app/actions/sourcing/add_suppliers.py
View File

@@ -1,17 +1,24 @@
from playwright.async_api import Page
from typing import ClassVar, override
from guide.app.actions.base import DemoAction
from guide.app.domain.models import ActionContext, ActionResult
from guide.app.strings import demo_texts, selectors
from guide.app.models.domain import ActionContext, ActionResult
from guide.app.strings.service import strings
class AddThreeSuppliersAction(DemoAction):
id = "add-three-suppliers"
description = "Adds three default suppliers to the sourcing event."
category = "sourcing"
id: ClassVar[str] = "add-three-suppliers"
description: ClassVar[str] = "Adds three default suppliers to the sourcing event."
category: ClassVar[str] = "sourcing"
@override
async def run(self, page: Page, context: ActionContext) -> ActionResult:
for supplier in demo_texts.SUPPLIERS.DEFAULT_TRIO:
await page.fill(selectors.SOURCING.SUPPLIER_SEARCH_INPUT, supplier)
await page.click(selectors.SOURCING.ADD_SUPPLIER_BUTTON)
return ActionResult(details={"added_suppliers": demo_texts.SUPPLIERS.DEFAULT_TRIO})
suppliers_val = strings.text("SUPPLIERS", "DEFAULT_TRIO")
if not isinstance(suppliers_val, list):
raise ValueError("SUPPLIERS.DEFAULT_TRIO must be a list of strings")
suppliers: list[str] = list(suppliers_val)
for supplier in suppliers:
await page.fill(strings.selector("SOURCING", "SUPPLIER_SEARCH_INPUT"), supplier)
await page.click(strings.selector("SOURCING", "ADD_SUPPLIER_BUTTON"))
return ActionResult(details={"added_suppliers": list(suppliers)})

97
src/guide/app/api/routes/actions.py
View File

@@ -1,62 +1,101 @@
from fastapi import APIRouter, Depends, HTTPException, Request, status
from typing import Annotated, Protocol, cast
from guide.app.actions import ActionRegistry
from fastapi import APIRouter, Depends, Request
from fastapi import FastAPI
from guide.app.actions.registry import ActionRegistry
from guide.app.auth import DummyMfaCodeProvider, ensure_persona
from guide.app.browser.client import BrowserClient
from guide.app.domain.models import ActionContext, ActionRequest, ActionResponse
from guide.app import errors
from guide.app.core.config import AppSettings
from guide.app.models.domain import ActionContext, ActionEnvelope, ActionRequest, ActionStatus
from guide.app.models.personas import PersonaStore
router = APIRouter()
class AppStateProtocol(Protocol):
action_registry: ActionRegistry
browser_client: BrowserClient
persona_store: PersonaStore
settings: AppSettings
def _registry(request: Request) -> ActionRegistry:
return request.app.state.action_registry
app = cast(FastAPI, request.app)
state = cast(AppStateProtocol, cast(object, app.state))
return state.action_registry
def _browser_client(request: Request) -> BrowserClient:
return request.app.state.browser_client
app = cast(FastAPI, request.app)
state = cast(AppStateProtocol, cast(object, app.state))
return state.browser_client
def _personas(request: Request) -> PersonaStore:
app = cast(FastAPI, request.app)
state = cast(AppStateProtocol, cast(object, app.state))
return state.persona_store
def _settings(request: Request) -> AppSettings:
app = cast(FastAPI, request.app)
state = cast(AppStateProtocol, cast(object, app.state))
return state.settings
RegistryDep = Annotated[ActionRegistry, Depends(_registry)]
BrowserDep = Annotated[BrowserClient, Depends(_browser_client)]
PersonaDep = Annotated[PersonaStore, Depends(_personas)]
SettingsDep = Annotated[AppSettings, Depends(_settings)]
@router.get("/actions")
async def list_actions(registry: ActionRegistry = Depends(_registry)):
async def list_actions(registry: RegistryDep):
return registry.list_metadata()
@router.post("/actions/{action_id}/execute", response_model=ActionResponse)
@router.post("/actions/{action_id}/execute", response_model=ActionEnvelope)
async def execute_action(
action_id: str,
payload: ActionRequest,
registry: ActionRegistry = Depends(_registry),
browser_client: BrowserClient = Depends(_browser_client),
registry: RegistryDep,
browser_client: BrowserDep,
personas: PersonaDep,
settings: SettingsDep,
):
try:
action = registry.get(action_id)
except KeyError as exc:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Unknown action '{action_id}'",
) from exc
action = registry.get(action_id)
persona = personas.get(payload.persona_id) if payload.persona_id else None
target_host_id = payload.browser_host_id or (persona.browser_host_id if persona else None)
target_host_id = target_host_id or browser_client.settings.default_browser_host_id
context = ActionContext(
browser_host_id=payload.browser_host_id or browser_client.settings.default_browser_host_id,
params=payload.params or {},
action_id=action_id,
persona_id=persona.id if persona else None,
browser_host_id=target_host_id,
params=payload.params,
)
mfa_provider = DummyMfaCodeProvider()
try:
async with browser_client.open_page(context.browser_host_id) as page:
async with browser_client.open_page(target_host_id) as page:
if persona:
await ensure_persona(page, persona, mfa_provider, login_url=settings.raindrop_base_url)
result = await action.run(page, context)
except Exception as exc:
return ActionResponse(
status="error",
except errors.GuideError as exc:
return ActionEnvelope(
status=ActionStatus.ERROR,
action_id=action_id,
browser_host_id=context.browser_host_id,
correlation_id=context.correlation_id,
error=str(exc),
error_code=exc.code,
message=exc.message,
details=exc.details,
)
return ActionResponse(
status=result.status,
return ActionEnvelope(
status=ActionStatus.SUCCESS,
action_id=action_id,
browser_host_id=context.browser_host_id,
correlation_id=context.correlation_id,
details=result.details,
error=result.error,
result=result.details,
)

23
src/guide/app/api/routes/config.py
View File

@@ -1,16 +1,29 @@
from fastapi import APIRouter, Depends, Request
from typing import Annotated, Protocol, cast
from guide.app.domain.models import BrowserHostsResponse
from fastapi import APIRouter, Depends, Request
from fastapi import FastAPI
from guide.app.core.config import AppSettings
from guide.app.models.domain import BrowserHostsResponse
router = APIRouter()
def _settings(request: Request):
return request.app.state.settings
class AppStateProtocol(Protocol):
settings: AppSettings
def _settings(request: Request) -> AppSettings:
app = cast(FastAPI, request.app)
state = cast(AppStateProtocol, cast(object, app.state))
return state.settings
SettingsDep = Annotated[AppSettings, Depends(_settings)]
@router.get("/config/browser-hosts", response_model=BrowserHostsResponse)
async def list_browser_hosts(settings=Depends(_settings)) -> BrowserHostsResponse: # type: ignore[override]
async def list_browser_hosts(settings: SettingsDep) -> BrowserHostsResponse:
return BrowserHostsResponse(
default_browser_host_id=settings.default_browser_host_id,
browser_hosts=settings.browser_hosts,

74
src/guide/app/browser/client.py
View File

@@ -1,52 +1,88 @@
import contextlib
from typing import AsyncIterator
from collections.abc import AsyncIterator
from playwright.async_api import Browser, Page, async_playwright
from playwright.async_api import Browser, BrowserContext, Page, Playwright, async_playwright
from guide.app.core.config import AppSettings, BrowserHost
from guide.app.core.config import AppSettings, BrowserHostConfig, HostKind
from guide.app import errors
class BrowserClient:
"""CDP-aware connector that returns the active Raindrop page."""
"""Connector that yields a Playwright Page for either CDP or headless hosts."""
settings: AppSettings
def __init__(self, settings: AppSettings) -> None:
self.settings = settings
def _resolve_host(self, host_id: str | None) -> BrowserHost:
def _resolve_host(self, host_id: str | None) -> BrowserHostConfig:
resolved_id = host_id or self.settings.default_browser_host_id
host = self.settings.browser_hosts.get(resolved_id)
if not host:
known = ", ".join(self.settings.browser_hosts.keys()) or "<none>"
raise ValueError(f"Unknown browser host '{resolved_id}'. Known: {known}")
raise errors.ConfigError(f"Unknown browser host '{resolved_id}'. Known: {known}")
return host
@contextlib.asynccontextmanager
async def open_page(self, host_id: str | None = None) -> AsyncIterator[Page]:
host = self._resolve_host(host_id)
cdp_url = f"http://{host.host}:{host.cdp_port}"
playwright = await async_playwright().start()
browser: Browser | None = None
context: BrowserContext | None = None
try:
browser = await playwright.chromium.connect_over_cdp(cdp_url)
page = self._pick_raindrop_page(browser)
if not page:
raise RuntimeError("No Raindrop page found in connected browser.")
if host.kind == HostKind.CDP:
if not host.host or host.port is None:
raise errors.ConfigError("CDP host requires host and port fields.")
cdp_url = f"http://{host.host}:{host.port}"
try:
browser = await playwright.chromium.connect_over_cdp(cdp_url)
except Exception as exc: # pragma: no cover - network dependent
raise errors.BrowserConnectionError(
f"Cannot connect to CDP endpoint {cdp_url}", details={"host_id": host.id}
) from exc
page = self._pick_raindrop_page(browser)
if not page:
raise errors.BrowserConnectionError(
"No Raindrop page found in connected browser.", details={"host_id": host.id}
)
else:
browser_type = _resolve_browser_type(playwright, host.browser)
browser = await browser_type.launch(headless=True)
context = await browser.new_context()
page = await context.new_page()
yield page
finally:
# Avoid killing the remote browser; just drop the connection.
with contextlib.suppress(Exception):
if context:
await context.close()
with contextlib.suppress(Exception):
if browser:
browser.disconnect()
await browser.close()
with contextlib.suppress(Exception):
await playwright.stop()
def _pick_raindrop_page(self, browser: Browser) -> Page | None:
target_substr = self.settings.raindrop_base_url
pages = []
pages: list[Page] = []
for context in browser.contexts:
pages.extend(context.pages)
pages = pages or [page for page in browser.contexts[0].pages] if browser.contexts else []
for page in reversed(pages):
if target_substr in (page.url or ""):
return page
return pages[-1] if pages else None
pages = pages or list(browser.contexts[0].pages) if browser.contexts else []
return next(
(
page
for page in reversed(pages)
if target_substr in (page.url or "")
),
pages[-1] if pages else None,
)
def _resolve_browser_type(playwright: Playwright, browser: str | None):
desired = (browser or "chromium").lower()
if desired == "chromium":
return playwright.chromium
if desired == "firefox":
return playwright.firefox
if desired == "webkit":
return playwright.webkit
raise errors.ConfigError(f"Unsupported headless browser type '{browser}'")

166
src/guide/app/core/config.py
View File

@@ -1,69 +1,185 @@
import json
import os
from enum import Enum
from pathlib import Path
from collections.abc import Mapping
from typing import ClassVar, TypeAlias, TypeGuard, cast
from pydantic import BaseModel, Field
from pydantic_settings import BaseSettings, SettingsConfigDict
CONFIG_DIR = Path(__file__).resolve().parents[4] / "config"
HOSTS_FILE = CONFIG_DIR / "hosts.yaml"
PERSONAS_FILE = CONFIG_DIR / "personas.yaml"
JsonRecord: TypeAlias = dict[str, object]
RecordList: TypeAlias = list[JsonRecord]
class BrowserHost(BaseModel):
def _coerce_mapping(mapping: Mapping[object, object]) -> dict[str, object]:
return {str(key): value for key, value in mapping.items()}
def _is_object_mapping(value: object) -> TypeGuard[Mapping[object, object]]:
return isinstance(value, Mapping)
class HostKind(str, Enum):
CDP = "cdp"
HEADLESS = "headless"
class BrowserHostConfig(BaseModel):
id: str
host: str
cdp_port: int
kind: HostKind
host: str | None = None
port: int | None = None
browser: str | None = None # chromium/firefox/webkit for headless
class PersonaConfig(BaseModel):
id: str
role: str
email: str
login_method: str = "mfa_email"
browser_host_id: str | None = None
class AppSettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix="RAINDROP_DEMO_")
model_config: ClassVar[SettingsConfigDict] = SettingsConfigDict(env_prefix="RAINDROP_DEMO_")
raindrop_base_url: str = "https://app.raindrop.com"
default_browser_host_id: str = "desktop"
browser_hosts: dict[str, BrowserHost] = Field(default_factory=dict)
raindrop_graphql_url: str = "https://app.raindrop.com/graphql"
default_browser_host_id: str = "demo-cdp"
browser_hosts: dict[str, BrowserHostConfig] = Field(default_factory=dict)
personas: dict[str, PersonaConfig] = Field(default_factory=dict)
def _load_hosts_file(path: Path) -> dict[str, BrowserHost]:
def _load_hosts_file(path: Path) -> dict[str, BrowserHostConfig]:
if not path.exists():
return {}
try:
import yaml # type: ignore
import yaml
except ModuleNotFoundError as exc:
raise RuntimeError(
"hosts.yaml found but PyYAML is not installed. "
"Add 'pyyaml' to dependencies or remove hosts.yaml."
"hosts.yaml found but PyYAML is not installed. Add 'pyyaml' to dependencies or remove hosts.yaml."
) from exc
data = yaml.safe_load(path.read_text()) or {}
hosts_raw = data.get("hosts") or data
hosts: dict[str, BrowserHost] = {}
for item in hosts_raw:
host = BrowserHost.model_validate(item)
data_raw: object = yaml.safe_load(path.read_text()) or {}
hosts: dict[str, BrowserHostConfig] = {}
for item in _normalize_records(data_raw, key_name="hosts"):
host = BrowserHostConfig.model_validate(item)
hosts[host.id] = host
return hosts
def _parse_hosts_json(value: str) -> dict[str, BrowserHost]:
decoded = json.loads(value)
hosts: dict[str, BrowserHost] = {}
def _parse_json_hosts(value: str) -> dict[str, BrowserHostConfig]:
decoded_raw: object = cast(object, json.loads(value))
decoded = _ensure_record_list(decoded_raw, "RAINDROP_DEMO_BROWSER_HOSTS_JSON")
hosts: dict[str, BrowserHostConfig] = {}
for item in decoded:
host = BrowserHost.model_validate(item)
host = BrowserHostConfig.model_validate(item)
hosts[host.id] = host
return hosts
def _parse_json_personas(value: str) -> dict[str, PersonaConfig]:
decoded_raw: object = cast(object, json.loads(value))
decoded = _ensure_record_list(decoded_raw, "RAINDROP_DEMO_PERSONAS_JSON")
personas: dict[str, PersonaConfig] = {}
for item in decoded:
persona = PersonaConfig.model_validate(item)
personas[persona.id] = persona
return personas
def _load_personas_file(path: Path) -> dict[str, PersonaConfig]:
if not path.exists():
return {}
try:
import yaml
except ModuleNotFoundError as exc:
raise RuntimeError(
"personas.yaml found but PyYAML is not installed. Add 'pyyaml' to dependencies or remove personas.yaml."
) from exc
data_raw: object = yaml.safe_load(path.read_text()) or {}
personas: dict[str, PersonaConfig] = {}
for item in _normalize_records(data_raw, key_name="personas"):
persona = PersonaConfig.model_validate(item)
personas[persona.id] = persona
return personas
def _normalize_records(data_raw: object, key_name: str) -> RecordList:
if isinstance(data_raw, Mapping):
mapping_raw: dict[str, object] = _coerce_mapping(cast(Mapping[object, object], data_raw))
content: object = mapping_raw.get(key_name, mapping_raw)
else:
content = data_raw
if isinstance(content, Mapping):
mapping_content: dict[str, object] = _coerce_mapping(cast(Mapping[object, object], content))
records: RecordList = []
for key, value in mapping_content.items():
mapping = _ensure_mapping(value)
if "id" not in mapping:
mapping["id"] = key
records.append(mapping)
return records
if isinstance(content, list):
list_content: list[object] = cast(list[object], content)
list_records: RecordList = []
for item in list_content:
if _is_object_mapping(item):
mapping_item = _coerce_mapping(item)
list_records.append(dict(mapping_item))
return list_records
return []
def _ensure_mapping(value: object) -> JsonRecord:
if isinstance(value, Mapping):
return _coerce_mapping(cast(Mapping[object, object], value))
return {}
def _ensure_record_list(value: object, source_label: str) -> RecordList:
if isinstance(value, list):
list_content: list[object] = cast(list[object], value)
list_records: RecordList = []
for item in list_content:
if _is_object_mapping(item):
mapping_item = _coerce_mapping(item)
list_records.append(dict(mapping_item))
return list_records
raise ValueError(f"{source_label} must be a JSON array of objects")
def load_settings() -> AppSettings:
settings = AppSettings()
merged_hosts: dict[str, BrowserHost] = {}
merged_hosts: dict[str, BrowserHostConfig] = {}
merged_personas: dict[str, PersonaConfig] = {}
from_file = _load_hosts_file(HOSTS_FILE)
merged_hosts.update(from_file)
merged_hosts |= _load_hosts_file(HOSTS_FILE)
merged_personas |= _load_personas_file(PERSONAS_FILE)
if env_json := os.environ.get("RAINDROP_DEMO_BROWSER_HOSTS_JSON"):
merged_hosts.update(_parse_hosts_json(env_json))
merged_hosts.update(_parse_json_hosts(env_json))
if merged_hosts:
settings = settings.model_copy(update={"browser_hosts": merged_hosts})
if env_personas := os.environ.get("RAINDROP_DEMO_PERSONAS_JSON"):
merged_personas.update(_parse_json_personas(env_personas))
if merged_hosts or merged_personas:
settings = settings.model_copy(
update={
"browser_hosts": merged_hosts or settings.browser_hosts,
"personas": merged_personas or settings.personas,
}
)
return settings

5
src/guide/app/core/logging.py
View File

@@ -1,11 +1,10 @@
import logging
from typing import Optional
def configure_logging(level: int | str = logging.INFO, correlation_id: Optional[str] = None) -> None:
def configure_logging(level: int | str = logging.INFO, correlation_id: str | None = None) -> None:
logging.basicConfig(
level=level,
format="%(asctime)s [%(levelname)s] %(message)s",
)
if correlation_id:
logging.LoggerAdapter(logging.getLogger(), {"correlation_id": correlation_id})
_ = logging.LoggerAdapter(logging.getLogger(), {"correlation_id": correlation_id})

0
src/guide/app/domain/__init__.py
View File

43
src/guide/app/domain/models.py
View File

@@ -1,43 +0,0 @@
import uuid
from typing import Any, Literal, Optional
from pydantic import BaseModel, Field
from guide.app.core.config import BrowserHost
class ActionRequest(BaseModel):
browser_host_id: Optional[str] = None
params: dict[str, Any] = Field(default_factory=dict)
class ActionContext(BaseModel):
browser_host_id: str
params: dict[str, Any] = Field(default_factory=dict)
correlation_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
class ActionResult(BaseModel):
status: Literal["ok", "error"] = "ok"
details: dict[str, Any] = Field(default_factory=dict)
error: Optional[str] = None
class ActionMetadata(BaseModel):
id: str
description: str
category: str
class ActionResponse(BaseModel):
status: Literal["ok", "error"]
action_id: str
browser_host_id: str
correlation_id: Optional[str] = None
details: dict[str, Any] | None = None
error: Optional[str] = None
class BrowserHostsResponse(BaseModel):
default_browser_host_id: str
browser_hosts: dict[str, BrowserHost]

29
src/guide/app/main.py
View File

@@ -1,26 +1,49 @@
from fastapi import FastAPI
from guide.app.actions import default_registry
from guide.app.actions.registry import default_registry
from guide.app.browser.client import BrowserClient
from guide.app.core.config import load_settings
from guide.app.core.config import AppSettings, load_settings
from guide.app.core.logging import configure_logging
from guide.app.api import router as api_router
from guide.app import errors
from guide.app.models.personas import PersonaStore
from fastapi import Request
from fastapi.exception_handlers import http_exception_handler
from fastapi.exceptions import HTTPException
def create_app() -> FastAPI:
configure_logging()
settings = load_settings()
registry = default_registry()
persona_store = PersonaStore(settings)
registry = default_registry(persona_store, settings.raindrop_base_url)
browser_client = BrowserClient(settings)
app = FastAPI(title="Raindrop Demo Automation", version="0.1.0")
app.state.settings = settings
app.state.action_registry = registry
app.state.browser_client = browser_client
app.state.persona_store = persona_store
# Dependency overrides so FastAPI deps can pull settings without globals
app.dependency_overrides = {AppSettings: lambda: settings}
app.include_router(api_router)
app.add_exception_handler(errors.GuideError, guide_exception_handler)
app.add_exception_handler(Exception, general_exception_handler)
return app
async def guide_exception_handler(request: Request, exc: Exception):
if isinstance(exc, errors.GuideError):
return errors.guide_error_handler(request, exc)
return await general_exception_handler(request, exc)
async def general_exception_handler(request: Request, exc: Exception):
if isinstance(exc, HTTPException):
return await http_exception_handler(request, exc)
return errors.unhandled_error_handler(request, exc)
app = create_app()

12
src/guide/app/strings/__init__.py
View File

@@ -1,12 +0,0 @@
from guide.app.strings.demo_texts import DemoTexts, texts
from guide.app.strings.labels import Labels, labels
from guide.app.strings.selectors import Selectors, selectors
__all__ = [
"DemoTexts",
"Labels",
"Selectors",
"texts",
"labels",
"selectors",
]

8
src/guide/app/strings/demo_texts/__init__.py
View File

@@ -1,12 +1,14 @@
from typing import ClassVar
from guide.app.strings.demo_texts.events import EventTexts
from guide.app.strings.demo_texts.intake import IntakeTexts
from guide.app.strings.demo_texts.suppliers import SupplierTexts
class DemoTexts:
INTAKE = IntakeTexts
SUPPLIERS = SupplierTexts
EVENTS = EventTexts
INTAKE: ClassVar[type[IntakeTexts]] = IntakeTexts
SUPPLIERS: ClassVar[type[SupplierTexts]] = SupplierTexts
EVENTS: ClassVar[type[EventTexts]] = EventTexts
texts = DemoTexts()

5
src/guide/app/strings/demo_texts/events.py
View File

@@ -1,2 +1,5 @@
from typing import ClassVar
class EventTexts:
THREE_BIDS_EVENT_NAME = "Three Bids and a Buy – Demo"
THREE_BIDS_EVENT_NAME: ClassVar[str] = "Three Bids and a Buy – Demo"

7
src/guide/app/strings/demo_texts/intake.py
View File

@@ -1,5 +1,8 @@
from typing import ClassVar
class IntakeTexts:
CONVEYOR_BELT_REQUEST = (
CONVEYOR_BELT_REQUEST: ClassVar[str] = (
"Requesting 500 tons of replacement conveyor belts for Q4 maintenance window."
)
ALT_REQUEST = "Intake for rapid supplier onboarding and risk review."
ALT_REQUEST: ClassVar[str] = "Intake for rapid supplier onboarding and risk review."

7
src/guide/app/strings/demo_texts/suppliers.py
View File

@@ -1,7 +1,10 @@
from typing import ClassVar
class SupplierTexts:
DEFAULT_TRIO = [
DEFAULT_TRIO: ClassVar[list[str]] = [
"Demo Supplier A",
"Demo Supplier B",
"Demo Supplier C",
]
NOTES = "Sourced automatically via demo action."
NOTES: ClassVar[str] = "Sourced automatically via demo action."

12
src/guide/app/strings/labels/__init__.py
View File

@@ -1,13 +1,17 @@
from typing import ClassVar
from guide.app.strings.labels.intake import IntakeLabels
from guide.app.strings.labels.sourcing import SourcingLabels
from guide.app.strings.labels.auth import AuthLabels
class Labels:
INTAKE = IntakeLabels
SOURCING = SourcingLabels
NEXT_BUTTON = "Next"
INTAKE: ClassVar[type[IntakeLabels]] = IntakeLabels
SOURCING: ClassVar[type[SourcingLabels]] = SourcingLabels
AUTH: ClassVar[type[AuthLabels]] = AuthLabels
NEXT_BUTTON: ClassVar[str] = "Next"
labels = Labels()
__all__ = ["Labels", "labels", "IntakeLabels", "SourcingLabels"]
__all__ = ["Labels", "labels", "IntakeLabels", "SourcingLabels", "AuthLabels"]

7
src/guide/app/strings/labels/intake.py
View File

@@ -1,6 +1,9 @@
"""User-visible labels for intake."""
from typing import ClassVar
class IntakeLabels:
DESCRIPTION_PLACEHOLDER = "Describe what you need"
NEXT_BUTTON = "Next"
DESCRIPTION_PLACEHOLDER: ClassVar[str] = "Describe what you need"
NEXT_BUTTON: ClassVar[str] = "Next"

7
src/guide/app/strings/labels/sourcing.py
View File

@@ -1,6 +1,9 @@
"""Labels for sourcing screens."""
from typing import ClassVar
class SourcingLabels:
SUPPLIERS_TAB = "Suppliers"
ADD_BUTTON = "Add"
SUPPLIERS_TAB: ClassVar[str] = "Suppliers"
ADD_BUTTON: ClassVar[str] = "Add"

19
src/guide/app/strings/selectors/__init__.py
View File

@@ -1,14 +1,25 @@
from typing import ClassVar
from guide.app.strings.selectors.intake import IntakeSelectors
from guide.app.strings.selectors.navigation import NavigationSelectors
from guide.app.strings.selectors.sourcing import SourcingSelectors
from guide.app.strings.selectors.auth import AuthSelectors
class Selectors:
INTAKE = IntakeSelectors
SOURCING = SourcingSelectors
NAVIGATION = NavigationSelectors
INTAKE: ClassVar[type[IntakeSelectors]] = IntakeSelectors
SOURCING: ClassVar[type[SourcingSelectors]] = SourcingSelectors
NAVIGATION: ClassVar[type[NavigationSelectors]] = NavigationSelectors
AUTH: ClassVar[type[AuthSelectors]] = AuthSelectors
selectors = Selectors()
__all__ = ["Selectors", "selectors", "IntakeSelectors", "SourcingSelectors", "NavigationSelectors"]
__all__ = [
"Selectors",
"selectors",
"IntakeSelectors",
"SourcingSelectors",
"NavigationSelectors",
"AuthSelectors",
]

7
src/guide/app/strings/selectors/intake.py
View File

@@ -1,6 +1,9 @@
"""Selectors for the intake flow."""
from typing import ClassVar
class IntakeSelectors:
DESCRIPTION_FIELD = '[data-test="intake-description"]'
NEXT_BUTTON = '[data-test="intake-next"]'
DESCRIPTION_FIELD: ClassVar[str] = '[data-test="intake-description"]'
NEXT_BUTTON: ClassVar[str] = '[data-test="intake-next"]'

7
src/guide/app/strings/selectors/navigation.py
View File

@@ -1,6 +1,9 @@
"""Selectors used for simple navigation helpers."""
from typing import ClassVar
class NavigationSelectors:
GLOBAL_SEARCH = '[data-test="global-search"]'
FIRST_RESULT = '[data-test="search-result"]:first-child'
GLOBAL_SEARCH: ClassVar[str] = '[data-test="global-search"]'
FIRST_RESULT: ClassVar[str] = '[data-test="search-result"]:first-child'

9
src/guide/app/strings/selectors/sourcing.py
View File

@@ -1,7 +1,10 @@
"""Selectors for sourcing event setup."""
from typing import ClassVar
class SourcingSelectors:
SUPPLIER_SEARCH_INPUT = '[data-test="supplier-search"]'
ADD_SUPPLIER_BUTTON = '[data-test="add-supplier"]'
SUPPLIER_ROW = '[data-test="supplier-row"]'
SUPPLIER_SEARCH_INPUT: ClassVar[str] = '[data-test="supplier-search"]'
ADD_SUPPLIER_BUTTON: ClassVar[str] = '[data-test="add-supplier"]'
SUPPLIER_ROW: ClassVar[str] = '[data-test="supplier-row"]'

10
src/guide/main.py
View File

@@ -1,20 +1,20 @@
import os
from guide.app.main import app, create_app
from guide.app.main import app
def main() -> None:
"""Run a development server if uvicorn is available."""
try:
import uvicorn # type: ignore
except ModuleNotFoundError: # pragma: no cover - runtime convenience only
import uvicorn
except ModuleNotFoundError as e: # pragma: no cover - runtime convenience only
raise SystemExit(
"uvicorn is required to run the server. Install with `pip install uvicorn`."
)
) from e
host = os.getenv("HOST", "0.0.0.0")
port = int(os.getenv("PORT", "8000"))
uvicorn.run("guide.main:app", host=host, port=port, reload=False)
uvicorn.run(app, host=host, port=port, reload=False)
if __name__ == "__main__":