diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8cca724..9deb3bf 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -23,8 +23,7 @@ jobs: python -m pip install --upgrade pip pip install -e ".[dev,rich]" - name: Run tests - run: | - python -m pytest tests/ -v + run: python -m pytest tests/ -v core: name: zero-dependency core @@ -35,12 +34,84 @@ jobs: with: python-version: "3.11" cache: "pip" - - name: Install core + dev only (no optional extras) + - name: Install core + dev only (no optional runtime extras) run: pip install -e ".[dev]" - name: Run tests - # The core runs on the stdlib alone; optional-extra tests skip cleanly. run: python -m pytest -q + e2e: + name: all-extras end-to-end smoke + runs-on: ubuntu-latest + env: + SOLVENT_HOME: ${{ runner.temp }}/solvent + SOLVENT_SKIP_ONBOARD: "1" + SOLVENT_FORCE_STRIPE_SIMULATE: "1" + SOLVENT_DELIVERY_SECRET: "ci-only-delivery-secret-0123456789abcdef" + steps: + - uses: actions/checkout@v7 + - uses: actions/setup-python@v6 + with: + python-version: "3.11" + cache: "pip" + - name: Install every optional feature + run: pip install -e ".[all,dev]" + - name: Run the full suite with every extra installed + run: python -m pytest -q + - name: Exercise offline CLI flows + run: | + solvent --version + solvent help + SOLVENT_HOME="${RUNNER_TEMP}/guardrails" python demo_guardrails.py + python run_demo.py --no-onboard --seed 100 + solvent status + solvent jobs --json + solvent finance --json + solvent webhooks stats + solvent worker --once --keep-balance + solvent workspace setup + solvent workspace list + solvent pairing list + printf '/fund 25\nCI interactive brief\n50\nn\n' | python run_demo.py --interactive --no-onboard --seed 100 + solvent doctor + - name: Exercise the real HTTP server and QR endpoint + run: | + solvent serve --port 8787 --keep-balance >"${RUNNER_TEMP}/solvent-server.log" 2>&1 & + server_pid=$! + trap 'kill "$server_pid" 2>/dev/null || true' EXIT + ready=0 + for attempt in $(seq 1 30); do + if curl -fsS http://127.0.0.1:8787/health > /dev/null; then + ready=1 + break + fi + sleep 1 + done + if [ "$ready" -ne 1 ]; then + cat "${RUNNER_TEMP}/solvent-server.log" + exit 1 + fi + curl -fsS http://127.0.0.1:8787/health + curl -fsS http://127.0.0.1:8787/api/status + curl -fsS http://127.0.0.1:8787/ > /dev/null + curl -fsS -X POST http://127.0.0.1:8787/api/chat \ + -H 'Content-Type: application/json' \ + -d '{"message":"/status"}' > "${RUNNER_TEMP}/chat.json" + grep -q 'Balance' "${RUNNER_TEMP}/chat.json" + timeout 3 curl -fsSN http://127.0.0.1:8787/api/events \ + > "${RUNNER_TEMP}/events.txt" || [ "$?" -eq 124 ] + grep -q 'hello' "${RUNNER_TEMP}/events.txt" + curl -fsS -X POST http://127.0.0.1:8787/jobs \ + -H 'Content-Type: application/json' \ + -d '{"id":"J-http-smoke","topic":"HTTP smoke test","budget_cents":5000,"customer_email":"ci@example.com"}' \ + > "${RUNNER_TEMP}/job.json" + grep -q 'session_id' "${RUNNER_TEMP}/job.json" + curl -fsS http://127.0.0.1:8787/jobs/J-http-smoke + curl -fsS http://127.0.0.1:8787/api/pair/qr -o "${RUNNER_TEMP}/pairing-qr" + test -s "${RUNNER_TEMP}/pairing-qr" + kill "$server_pid" + wait "$server_pid" || true + trap - EXIT + build: name: build + install + CLI smoke runs-on: ubuntu-latest diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md deleted file mode 100644 index 00dd687..0000000 --- a/ARCHITECTURE.md +++ /dev/null @@ -1,115 +0,0 @@ -# SOLVENT — Architecture - -**SOLVENT is an AI agent that operates as a profitable, self-funding business.** -It sells research briefs, collects payment through Stripe, and spends its own -revenue to provision the compute, data, and SaaS it needs to do the work — and -it refuses any job that doesn't clear a margin. The result is a fully automated -company with its own balance sheet that grows on its own. - -## The one-sentence pitch -An agent you can hand a Stripe key and walk away from: it earns, it pays its own -bills, it never works at a loss, and it can't spend a dollar outside policy. - -## Why it matters -Most agents can spend money; few can run as a business. SOLVENT closes the full -loop with **economic self-awareness** — it maintains a treasury, prices against its -own unit costs, and gates every action on projected profit. That's the difference -between a demo bot and a business. - -## The money loop (per job) - -``` - inbound job - │ - ▼ - ┌─────────────┐ margin < floor? ┌───────────┐ - │ MARGIN GATE│ ─────────────────▶ │ DECLINE │ - │ (pricing) │ └───────────┘ - └─────┬───────┘ accept - ▼ - ┌─────────────┐ EARN - │ STRIPE │ ── Checkout Session → webhook (primary) ──▶ + revenue - └─────┬───────┘ idempotent stages in SQLite - ▼ - ┌─────────────┐ FULFIL - │ NEMOTRON │ ── bounded tool-calling research loop ──▶ actual COGS - └─────┬───────┘ - ▼ - ┌─────────────┐ DELIVER - │ HOSTED+SMTP│ ── signed brief URL + optional email - └─────┬───────┘ - ▼ - ┌─────────────┐ SPEND (each payment screened first) - │ GUARDRAILS │ ── NemoClaw policy: allowlist, caps, reserve, ROI - │ → STRIPE │ ── Issuing virtual card (test) or simulated spend ──▶ − expense - └─────┬───────┘ - ▼ - BOOK P&L ──▶ treasury updated, dashboard refreshed -``` - -Revenue is always collected **before** cost is incurred, and no single payment -can violate policy — the business is safe by construction and profitable by rule. - -## How the sponsor stack maps in - -| Layer | Sponsor tech | Where it lives in the code | -|---|---|---| -| **Reasoning / the analyst** | **NVIDIA Nemotron** (Llama-3.1-Nemotron-Ultra) via the OpenAI-compatible endpoint | `solvent/nemotron.py` | -| **Spend safety** | **NVIDIA NemoClaw**-style policy sandbox — every payment screened before it executes | `solvent/guardrails.py` | -| **Earn + Spend** | **Stripe** — Checkout Sessions, webhooks, idempotency keys; Issuing for outbound spend (test mode) | `solvent/stripe_client.py` | -| **Agent orchestration** | Idempotent **stage machine** + bounded Nemotron tool loop | `solvent/stages.py`, `solvent/tools.py`, `solvent/agent.py` | -| **Async processing** | SQLite queue + worker resume | `solvent/worker.py`, `solvent/queue.py` | -| **Delivery** | Hosted briefs + SMTP/outbox | `solvent/delivery.py`, `solvent/server.py` | -| **Conversational surface** | OpenClaw gateway + Hermes tools/memory | `solvent/gateway.py`, `solvent/chat.py`, `solvent/memory.py`, `solvent/hermes_tools.py` | -| **Agent identity** | SOUL/BRAIN/AGENTS workspace (Hermes + OpenClaw) | `solvent/workspace.py`, `.solvent/workspace/` | -| **Telegram** | Long-poll bot + DM pairing | `solvent/channels/telegram.py`, `solvent/pairing.py` | -| **Live dashboard** | SSE chat + treasury | `solvent/server.py`, `solvent/event_hub.py`, `solvent/dashboard_chat.py` | -| **Cross-process notify** | Worker → serve chat outbox | `solvent/notifications.py` | -| **Auto-improvement** | Metrics-driven tuning | `solvent/improver.py` | -| **Observability** | JSON logs + Stripe reconciliation | `solvent/observability.py`, `solvent/reconcile.py` | -| **Economic memory** | the treasury / ledger that makes it a business | `solvent/treasury.py`, `solvent/pricing.py` | - -## Key design choices - -- **Viability:** the margin gate (`pricing.py`) means SOLVENT is *structurally* - incapable of unprofitable work. In the demo it earns $223, spends $13.35, and - books a 94% margin — and declines the one job that doesn't pencil out. -- **Safety:** `guardrails.py` enforces a vendor allowlist, a per-transaction cap, - a rolling 24h budget, a minimum cash reserve, and a no-negative-ROI rule. This - is the answer to "would you actually give an agent a payment credential?" -- **Runs anywhere:** with no API keys it runs on deterministic offline stubs, so - the demo always works. Add `NVIDIA_API_KEY` for live Nemotron and a Stripe - **test** key (`sk_test_...`) for real Payment Links. The client refuses live - Stripe keys outright. In test mode, payment is verified via Checkout Session - polling (or optional webhooks) before fulfilment; refunds use the PaymentIntent - id (`pi_...`). Product/Price catalog is cached locally; Issuing provides - capped virtual cards when enabled on the test account. - -## Files -``` -solvent/ - agent.py thin orchestrator delegating to stages - stages.py idempotent stage machine (quote→paid→fulfill→deliver→spend) - worker.py async job processor + resume incomplete jobs - server.py FastAPI: webhooks, /jobs, /briefs (optional deps) - tools.py allowlisted research tools - delivery.py hosted briefs + SMTP/outbox email - observability.py structured JSON logging - improver.py auto-tune pricing from job metrics - reconcile.py Stripe ↔ ledger reconciliation - gateway.py channel router (Telegram, dashboard) - chat.py Nemotron conversational loop + business tools - workspace.py SOUL/BRAIN/AGENTS prompt assembly - channels/ Telegram adapter - notifications.py worker→serve chat outbox - treasury.py ledger, jobs, stages, metrics (SQLite) - pricing.py margin gate + COGS overrides - guardrails.py NemoClaw-style spend policy - stripe_client.py Checkout Sessions + webhooks - nemotron.py Nemotron client + bounded tool loop - service.py research brief fulfillment - jobs.py sample inbound work - dashboard.py treasury HTML dashboard -run_demo.py CLI demo (--async, --serve) -docs/PRODUCTION.md production deployment guide -``` diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 0ce4c0c..3217533 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -1,40 +1,39 @@ # Contributing to SOLVENT -Thanks for your interest in SOLVENT! This is a hackathon project that I'm actively improving — contributions of all kinds are welcome. - -## Good First Issues - -| Area | What to do | -|---|---| -| **New research topics** | Add more sample jobs in `solvent/jobs.py` | -| **Nemotron prompts** | Improve the brief template in `solvent/service.py` | -| **New guardrail policies** | Add a rule to `solvent/guardrails.py` | -| **Dashboard improvements** | New charts or metrics in `solvent/dashboard.py` | -| **Tests** | Extend the pytest suite in `tests/` | - -## Running Locally +## Setup ```bash git clone https://github.com/ianalloway/solvent-agent.git cd solvent-agent -python3 run_demo.py # no install required for the demo -pip install pytest -python3 -m pytest tests/ -v # run the test suite +pip install -e ".[dev]" +python3 run_demo.py --no-onboard +python3 -m pytest -q ``` -## Pull Request Process +Install only the optional features you are changing: + +```bash +pip install -e ".[stripe]" +pip install -e ".[serve]" +pip install -e ".[telegram]" +pip install -e ".[rich]" +pip install -e ".[qr]" +``` -1. Fork the repo and create a branch: `git checkout -b feat/your-feature` -2. Make your changes. -3. Ensure `python3 -m pytest tests/ -q` passes. -4. Open a pull request with a clear description of what changed and why. +Dependency declarations live in `pyproject.toml`; do not add parallel requirements files. -## Code Style +## Pull requests -- Python 3.10+ with type hints where it helps readability. -- No external dependencies in core `solvent/` (keep it importable with stdlib only). -- Dependencies in `requirements.txt` are for optional live integrations only. +1. Create a focused branch. +2. Keep the standard-library core importable without optional extras. +3. Add or update tests for behavior changes. +4. Run `python3 -m pytest -q`. +5. Open a pull request that explains the user-visible behavior and tradeoffs. -## Questions? +## Project boundaries -Open an [issue](https://github.com/ianalloway/solvent-agent/issues) — happy to discuss ideas! +- `solvent/agent.py` is a thin public orchestrator; durable workflow logic belongs in `solvent/stages.py`. +- Treasury writes and Stripe actions stay behind stages and guardrails. +- API keys belong in environment variables only. +- Live Stripe keys are never accepted. +- Offline mode must keep working without credentials. diff --git a/README.md b/README.md index 911a637..560510d 100644 --- a/README.md +++ b/README.md @@ -2,373 +2,198 @@ # 🪙 SOLVENT -**An AI agent that runs as a profitable, self-funding business.** - -It sells research briefs. It collects payment on Stripe. It spends its own revenue to provision the compute it needs. And it refuses any job that doesn't clear a margin. +**A self-funding analyst agent with a real treasury, margin gate, and spend policy.** [![Python 3.10+](https://img.shields.io/badge/python-3.10%2B-blue?logo=python&logoColor=white)](https://www.python.org/downloads/) [![License: MIT](https://img.shields.io/badge/license-MIT-green)](LICENSE) -[![Hackathon](https://img.shields.io/badge/NVIDIA%20%C3%97%20Stripe%20Hackathon-2024-76b900?logo=nvidia&logoColor=white)](https://www.nvidia.com) [![Stars](https://img.shields.io/github/stars/ianalloway/solvent-agent?style=social)](https://github.com/ianalloway/solvent-agent/stargazers) -[**Quick Start**](#-quick-start) · [**How It Works**](#-how-it-works) · [**Live Demo**](#-the-demo) · [**Make It Real**](#-make-it-real) - ---- - -## The Big Idea - -Most agents can spend money. Almost none can **run as a business.** - -SOLVENT closes the full loop: +SOLVENT sells research briefs, collects payment through Stripe, fulfills the work with NVIDIA Nemotron or an offline stub, pays its own approved vendors, and books the result. It declines work that does not meet its margin floor. +```text +job → quote → collect payment → fulfill → deliver → guardrailed spend → book P&L ``` - Client pays Stripe → Agent earns revenue → Agent fulfils the work - → Agent pays its own vendor bills → P&L booked → balance sheet grows -``` - -Every job is **profit-gated before it starts**. Unprofitable work is declined without touching Stripe. Vendor payments are screened by a NemoClaw-style policy sandbox. The agent literally cannot spend more than it earns. - ---- -## 🚀 Quick Start +## Quick start -**Zero dependencies. No API keys. Works right now.** +The core demo has no third-party dependencies and needs no API keys: ```bash git clone https://github.com/ianalloway/solvent-agent.git cd solvent-agent -python3 run_demo.py +python3 run_demo.py --no-onboard ``` -The agent will run a full batch of 4 analyst jobs — complete with margin gating, Stripe payment simulation, NVIDIA Nemotron fulfillment, guardrail screening, and live P&L — in about 30 seconds. +The batch run processes four sample jobs, including an unprofitable job that is declined. It writes reports under `data/reports/` and renders `treasury_dashboard.html`. -### Install as a package (optional) +![SOLVENT treasury dashboard](docs/dashboard.png) -The core runs on the **standard library alone**, so a bare install pulls in -nothing extra and gives you a `solvent` command: +Install the package for the `solvent` command: ```bash -pip install -e . # editable install from a checkout -# or once published: pipx install solvent-agent - -solvent # run the demo -solvent finance # financial report (income, runway, forecast) -solvent --help # list all commands -solvent --version +pip install -e . +solvent +solvent finance +solvent --help ``` -Third-party features are **opt-in extras** — install only what you need: +Runtime data is written to the repository when running from a checkout and to `~/.solvent` when installed elsewhere. Set `SOLVENT_HOME=/path/to/home` to override it. -```bash -pip install -e ".[stripe]" # real Stripe test-mode payment links -pip install -e ".[serve]" # FastAPI webhooks + hosted briefs -pip install -e ".[telegram]" # Telegram bot channel -pip install -e ".[rich]" # live terminal dashboard (solvent tui) -pip install -e ".[all]" # everything -``` - -When run from a source checkout, runtime data stays under `/data`. When -installed elsewhere, SOLVENT writes to `~/.solvent` instead of into -`site-packages` — override either with `SOLVENT_HOME=/path/to/dir`. - -> **First run**: A short onboarding wizard asks you to choose a model, interaction mode, and whether to enable Stripe test mode. Preferences are saved to `.solvent/config.json` and never committed. - ---- - -## 📊 The Demo +## Optional features -After a run, open the live treasury dashboard: +Third-party integrations are explicit extras; there are no parallel requirements files to keep in sync. ```bash -open treasury_dashboard.html # macOS -``` - -![SOLVENT Treasury Dashboard — live P&L, job cards, resource allocation, transaction log](docs/dashboard.png) - -A typical batch session: - -| Metric | Value | -|---|---| -| Revenue | **$223.00** | -| Operating spend | $13.35 | -| **Net profit** | **$209.65** | -| Margin | **94%** | -| Jobs declined | 1 (below margin floor) | - ---- - -## ⚙️ How It Works - +pip install -e ".[stripe]" # Stripe test-mode checkout and spend +pip install -e ".[serve]" # FastAPI, webhooks, hosted briefs +pip install -e ".[telegram]" # Telegram channel +pip install -e ".[rich]" # terminal dashboard +pip install -e ".[qr]" # pairing QR images +pip install -e ".[all]" # every runtime feature +pip install -e ".[dev]" # pytest + HTTP test client ``` - inbound job - │ - ▼ - ┌─────────────┐ margin < floor? ┌───────────┐ - │ MARGIN GATE│ ─────────────────▶ │ DECLINE │ - │ (pricing) │ └───────────┘ - └─────┬───────┘ accept - ▼ - ┌─────────────┐ EARN - │ STRIPE │ ── Payment Link → poll/webhook until paid ──▶ + revenue - └─────┬───────┘ (records cs_... + pi_... on ledger) - ▼ - ┌─────────────┐ FULFIL - │ NEMOTRON │ ── Llama-3.1-Nemotron-Ultra produces the brief ──▶ resource usage - └─────┬───────┘ - ▼ - ┌─────────────┐ SPEND (every payment screened first) - │ GUARDRAILS │ ── NemoClaw policy: allowlist · caps · reserve · ROI - │ → STRIPE │ ── Issuing virtual card (test) or simulated spend ──▶ − expense - └─────┬───────┘ - ▼ - BOOK P&L ──▶ treasury updated · dashboard refreshed -``` - -Revenue is **always collected before cost is incurred**, and no payment can violate policy. The business is safe by construction and profitable by rule. - ---- - -## 🏗️ Architecture - -| Layer | Technology | File | -|---|---|---| -| **Analyst / reasoning** | NVIDIA Nemotron (Llama-3.1-Nemotron-Ultra) | `solvent/nemotron.py` | -| **Spend safety** | NVIDIA NemoClaw-style policy sandbox | `solvent/guardrails.py` | -| **Earn** | Stripe Payment Links + Checkout Session polling | `solvent/stripe_client.py` | -| **Spend** | Stripe Issuing virtual cards (test mode) | `solvent/stripe_client.py` | -| **Orchestration** | Hermes / Nous tool-calling agent loop | `solvent/agent.py` | -| **Memory** | SQLite treasury + pricing ledger | `solvent/treasury.py` · `solvent/pricing.py` | - -**Key design choices:** - -- **Structural profitability** — `pricing.py` computes unit cost before quoting. If margin < floor, the job never reaches Stripe. -- **Spend policy** — `guardrails.py` enforces vendor allowlist, per-transaction cap, rolling 24h budget, minimum cash reserve, and no-negative-ROI rule. -- **Offline-first** — without API keys the demo runs on deterministic stubs. Add `NVIDIA_API_KEY` + `STRIPE_API_KEY=sk_test_...` to unlock live inference and real Payment Links. -- **Audit trail** — every `cs_...` checkout session ID and `pi_...` PaymentIntent ID is recorded on the ledger before fulfilment begins. ---- +Live Stripe keys are deliberately refused. Use `sk_test_...` or `rk_test_...`. -## 🎮 Running Modes +## Running modes -### Batch demo (default — best for judges) +### Batch demo ```bash -python3 run_demo.py +python3 run_demo.py --no-onboard +# or, after installation: +solvent --no-onboard ``` -4 pre-loaded jobs. ~30 seconds. Shows margin gating, Stripe earn/spend, Nemotron fulfillment, and guardrails in action. - -### Interactive — your own jobs +### Interactive jobs ```bash -python3 run_demo.py --interactive +python3 run_demo.py --interactive --no-onboard ``` -Type a research topic and client budget at the prompt. The agent quotes, pays, fulfils, and books P&L for each one in real time. Keep going until you quit. - -### Add funds mid-session - -```bash -python3 run_demo.py --seed 500 # start with $500 instead of $100 -python3 run_demo.py --keep-balance # resume existing treasury balance -``` +Enter a topic and budget. Use `/fund 200` to add operating capital during the session. -In interactive mode, type `/fund 200` at the prompt to deposit $200 into the live treasury without restarting. - -### Programmatic +### Programmatic API ```python from solvent.agent import Solvent from solvent.jobs import SAMPLE_JOBS -agent = Solvent(seed_cents=10_000) # reset treasury, seed $100 -agent.handle_job(SAMPLE_JOBS[0]) # process one job -snap = agent.run(SAMPLE_JOBS[1:]) # process a list; returns snapshot - -print(snap["balance_cents"], snap["margin_pct"]) +agent = Solvent(seed_cents=10_000) +result = agent.handle_job(SAMPLE_JOBS[0]) +snapshot = agent.t.snapshot() ``` -### Guardrails demo (standalone) +### Guardrails demonstration ```bash python3 demo_guardrails.py ``` -Shows five spend attempts and which ones the NemoClaw-style policy blocks — without starting the agent or touching Stripe. - -### Production mode (webhooks + async worker) - -```bash -pip install -r requirements.txt -r requirements-serve.txt - -python3 -m solvent serve --port 8787 # webhooks + job API + hosted briefs -python3 -m solvent worker # resume incomplete jobs, process queue - -# Interactive voice dashboard (chat + live SSE updates): -open http://127.0.0.1:8787/ - -# Or dev convenience: -python3 run_demo.py --serve --no-onboard -``` - -The hosted dashboard at `/` includes a **chat panel** (type or use the mic with Web Speech API) and **live treasury updates** via Server-Sent Events (`/api/events`). Chat hits `/api/chat`, which routes through the Nemotron agent loop. +This exercises vendor allowlisting, transaction caps, daily budget, cash reserve, and projected-ROI checks without touching Stripe. -See [docs/PRODUCTION.md](docs/PRODUCTION.md) for Stripe webhook setup, SMTP delivery, reconciliation, and auto-tuning. +## Production-shaped stack -### Operations +Install the integrations and set a delivery secret: ```bash -python3 -m solvent reconcile --since 7d # Stripe ↔ ledger drift check -python3 -m solvent tune # propose pricing improvements (dry-run) -python3 -m solvent tune --apply # apply after 5+ completed jobs -python3 -m solvent finance # income statement, unit economics, runway -python3 -m solvent finance --json # machine-readable report -python3 -m solvent finance --reserve 50 # runway to a $50 cash-reserve floor -python3 -m solvent finance --period week # net P&L trend by day | week | month -python3 -m solvent finance --horizon 60 # forecast the balance 60 days out -``` - -`finance` (alias `report`) turns the treasury ledger into the numbers a -business steers by: revenue/cost/net-margin, average profit per job, a cash -**runway** — days of burn remaining, or `cash-flow positive` once the agent -funds itself — a **net-P&L trend** bucketed by day/week/month, and a -**balance forecast** (central projection with a best/worst band whose width -grows with daily volatility). The income statement, runway, trend, and -forecast also render as a **Financial Statement** panel in the HTML dashboard. - ---- - -## 🔑 Make It Real +pip install -e ".[serve,stripe]" +export SOLVENT_DELIVERY_SECRET="$(python3 -c 'import secrets; print(secrets.token_urlsafe(48))')" +export SOLVENT_BASE_URL=http://127.0.0.1:8787 -To use live Nemotron inference and real Stripe test-mode payment links: +# Terminal 1: API, webhooks, hosted briefs, dashboard +solvent serve --port 8787 -```bash -pip install -r requirements.txt - -export NVIDIA_API_KEY=nvapi-... # from build.nvidia.com -export STRIPE_API_KEY=sk_test_... # Stripe test mode only (live keys refused) - -python3 run_demo.py +# Terminal 2: resume and process queued work +solvent worker ``` -With both keys set: - -- Briefs are written by **NVIDIA Nemotron** (Llama-3.1-Nemotron-Ultra). -- Each job creates a real **Stripe Payment Link**. Pay with test card `4242 4242 4242 4242`. -- SOLVENT **polls** the Checkout Session (`cs_...`) until `payment_status == paid` before fulfilling — no instant confirm. -- Optional: set `STRIPE_WEBHOOK_SECRET` and forward `checkout.session.completed` events via `StripeClient.process_webhook()`. -- Optional: enable **Stripe Issuing** on your test account to provision capped single-use virtual debit cards for each vendor payment. - -### Environment variables +The HTTP surface includes: -| Variable | Purpose | -|---|---| -| `SOLVENT_HOME` | Where runtime data (treasury DB, reports, dashboard, logs) is stored. Defaults to the repo when run from a checkout, else `~/.solvent` | -| `NVIDIA_API_KEY` | Live Nemotron inference (`nvapi-...`) | -| `STRIPE_API_KEY` | Stripe test key (`sk_test_...`) | -| `STRIPE_WEBHOOK_SECRET` | Optional webhook verification | -| `STRIPE_PAYMENT_POLL_TIMEOUT` | Seconds to wait for payment (default `120`) | -| `STRIPE_PAYMENT_POLL_INTERVAL` | Poll interval in seconds (default `2`) | -| `SOLVENT_FORCE_STRIPE_SIMULATE` | Force offline simulate mode even with a key | -| `TELEGRAM_BOT_TOKEN` | Telegram bot token from BotFather | -| `SOLVENT_TELEGRAM_DM_POLICY` | `pairing` · `allowlist` · `open` (default `pairing`) | -| `SOLVENT_TELEGRAM_ALLOW_FROM` | Comma-separated Telegram user IDs for allowlist mode | +- `GET /health` +- `GET /` — interactive dashboard +- `GET /api/status` +- `GET /api/events` — Server-Sent Events +- `POST /api/chat` +- `POST /jobs` +- `POST /webhooks/stripe` +- signed `GET /briefs/{job_id}` delivery URLs -Product/Price objects are cached in `.solvent/stripe_catalog.json` so repeated runs reuse a single **SOLVENT Research Brief** product instead of cluttering your Stripe dashboard. +See [docs/PRODUCTION.md](docs/PRODUCTION.md) for Stripe webhook and SMTP setup. ---- - -## 💬 Telegram (conversational channel) - -Full chat on Telegram with OpenClaw-style pairing and Hermes-style tool/memory patterns. See **[docs/TELEGRAM.md](docs/TELEGRAM.md)**. +## Telegram ```bash -pip install -r requirements-telegram.txt +pip install -e ".[telegram,serve,stripe]" export TELEGRAM_BOT_TOKEN=... -python -m solvent serve & # Stripe webhooks + checkout -python -m solvent worker & # fulfill jobs -python -m solvent telegram # long-poll bot - -python -m solvent doctor # diagnostics -python -m solvent pairing list # pending DM codes +solvent serve +solvent worker +solvent telegram ``` -Users pair via `/start`, commission briefs in natural language, receive checkout links, and get push updates when jobs are paid and delivered. - -Personality and operating rules come from the **agent workspace** (`SOUL.md`, `BRAIN.md`, `AGENTS.md`) — see **[docs/WORKSPACE.md](docs/WORKSPACE.md)**. +Unknown users are paired before they can interact with the agent. See [docs/TELEGRAM.md](docs/TELEGRAM.md). ---- - -## 🧪 Tests +## Operations ```bash -pip install pytest -python3 -m pytest tests/ -v -``` - -Unit tests cover: pricing & margin gate · guardrail policy · treasury ledger · Stripe client (simulate + test mode) · config/onboarding. - ---- - -## 📁 Repository Layout +solvent status +solvent jobs +solvent jobs show +solvent jobs events +solvent jobs retry +solvent jobs cancel +solvent finance --json +solvent reconcile --since 7d +solvent tune # dry run +solvent tune --apply +solvent webhooks stats +solvent webhooks list +solvent doctor +solvent tui +``` + +`solvent report` remains an alias for `solvent finance`; `solvent retry ` remains a compatibility alias for `solvent jobs retry `. + +## Safety model + +- **Profitability:** pricing estimates COGS before checkout and rejects work below the configured margin floor. +- **Payment ordering:** revenue is collected before fulfillment cost is incurred. +- **Spend policy:** every vendor payment is checked against an allowlist, per-transaction cap, rolling budget, reserve floor, and job ROI. +- **Idempotency:** job stages and Stripe references are stored in SQLite so retries do not double-charge or double-book. +- **Credential boundary:** API keys stay in environment variables; Stripe live-mode keys are rejected. +- **Offline behavior:** deterministic stubs keep the complete business loop runnable without credentials. + +## Main modules + +```text +solvent/agent.py thin public orchestrator +solvent/stages.py idempotent money-loop state machine +solvent/treasury.py SQLite ledger, jobs, stages, metrics +solvent/pricing.py cost model and margin gate +solvent/guardrails.py outbound spend policy +solvent/stripe_client.py checkout, verification, refund, vendor spend +solvent/service.py research-brief fulfillment +solvent/delivery.py hosted links, HTML, SMTP/outbox +solvent/server.py FastAPI application +solvent/worker.py async resume/queue processor +solvent/gateway.py dashboard and Telegram channel routing +solvent/finance.py statements, runway, trend, forecast +``` + +Repository-specific architecture and safety conventions are also summarized in [AGENTS.md](AGENTS.md). + +## Tests +```bash +pip install -e ".[dev]" +python3 -m pytest -q ``` -solvent/ - agent.py the orchestrator (earn → fulfil → spend → book) - treasury.py SQLite ledger / balance sheet - pricing.py the margin gate - guardrails.py NemoClaw-style spend policy - stripe_client.py two-sided Stripe layer (earn + spend) - nemotron.py NVIDIA Nemotron client (+ offline stub) - service.py the product: an on-demand research brief - jobs.py sample inbound work - dashboard.py renders the treasury to HTML + JSON - finance.py income statement · unit economics · runway · forecast - config.py onboarding wizard and config persistence - gateway.py channel router (Telegram → chat sessions) - chat.py conversational loop + business tools - memory.py Hermes-style session memory - hermes_tools.py progressive tool disclosure bridge - doctor.py stack diagnostics - workspace.py SOUL/BRAIN/AGENTS prompt assembly - channels/ Telegram long-poll adapter -run_demo.py the full business loop (CLI entry point) -demo_guardrails.py the safety story (standalone) -tests/ pytest suite -docs/ screenshots and supporting assets -``` - ---- - -## 🏆 Built For - -**Hermes Agent Accelerated Business Hackathon** — NVIDIA × Stripe × Nous Research -The agent was designed to demonstrate: +CI runs Python 3.10–3.12, verifies the zero-dependency core, builds and installs the wheel, then installs every optional feature and exercises the batch demo, interactive flow, guardrails, worker, command surfaces, real HTTP server, and QR endpoint. -- An agent that is **economically self-aware** — it has a treasury, prices against its own costs, and gates every action on projected profit -- A **complete two-sided Stripe integration** — earns via Payment Links, spends via Issuing virtual cards -- **Provable spend safety** — a NemoClaw-style policy sandbox that makes "give an agent a payment credential" a reasonable thing to do -- **Live inference with NVIDIA Nemotron** — the offline stub means the demo always works, even without API keys +## License ---- - -## 🤝 Contributing - -Issues, PRs, and ideas are very welcome. Some good starting points: - -- Add more sample research topics in `solvent/jobs.py` -- Improve the Nemotron prompt template in `solvent/service.py` -- Add a new guardrail policy to `solvent/guardrails.py` -- Extend the dashboard with charts or new metrics in `solvent/dashboard.py` - ---- - -
- -**If SOLVENT gave you ideas, give it a ⭐** - -
+MIT. See [LICENSE](LICENSE). diff --git a/docs/PRODUCTION.md b/docs/PRODUCTION.md index 1e3b7e5..18e84c9 100644 --- a/docs/PRODUCTION.md +++ b/docs/PRODUCTION.md @@ -1,51 +1,51 @@ -# SOLVENT Production Guide +# SOLVENT production guide -Run SOLVENT as a webhook-first, async agent with hosted delivery. +SOLVENT's production shape is webhook-first: the HTTP process creates checkout sessions and receives Stripe events; the worker resumes paid jobs and fulfills them. -## Quick start (offline demo) +## Install + +Use package extras as the single dependency source: ```bash -python3 run_demo.py --no-onboard +pip install -e ".[serve,stripe]" ``` -## HTTP server + worker (production shape) +## Configure ```bash -pip install -r requirements.txt -pip install -r requirements-serve.txt +export SOLVENT_BASE_URL=https://solvent.example.com +export SOLVENT_DELIVERY_SECRET="$(python3 -c 'import secrets; print(secrets.token_urlsafe(48))')" +export STRIPE_API_KEY=sk_test_... +export STRIPE_WEBHOOK_SECRET=whsec_... +``` -export SOLVENT_BASE_URL=http://127.0.0.1:8787 -export SOLVENT_DELIVERY_SECRET=$(python3 -c 'import secrets; print(secrets.token_urlsafe(48))') +`SOLVENT_DELIVERY_SECRET` must be a non-placeholder value of at least 32 characters. SOLVENT refuses Stripe live-mode keys. -# Terminal 1 — API + webhooks + interactive dashboard -python3 -m solvent serve --port 8787 -open http://127.0.0.1:8787/ +## Run -# Terminal 2 — async job processor -python3 -m solvent worker +Use separate processes so either side can restart independently: + +```bash +# API, Stripe webhooks, hosted briefs, dashboard +solvent serve --host 0.0.0.0 --port 8787 -# Or combined dev mode: -python3 run_demo.py --serve --no-onboard +# Queue and incomplete-job processor +solvent worker ``` -### Interactive dashboard + voice +The dashboard is served at `/`. Health and status endpoints are available at `/health` and `/api/status`. -`python3 -m solvent serve` serves a live dashboard at `/` with: +## Stripe webhook -- **SSE** (`GET /api/events`) — treasury metrics, job cards, and console log update in real time -- **Chat** (`POST /api/chat`) — talk to the agent; supports `/status`, `/jobs`, `/quote topic | 50` -- **Voice** — mic button uses browser Web Speech API (Chrome/Edge); toggle 🔊 for spoken replies -- **Jobs** (`POST /api/job`) — same payload as `POST /jobs` +Create a Stripe test-mode webhook endpoint: -## Stripe setup (test mode) +```text +POST https://solvent.example.com/webhooks/stripe +``` -1. Set `STRIPE_API_KEY=sk_test_...` or restricted `rk_test_...` -2. Create webhook endpoint: `POST {SOLVENT_BASE_URL}/webhooks/stripe` -3. Subscribe to `checkout.session.completed` -4. Set `STRIPE_WEBHOOK_SECRET=whsec_...` -5. Submit jobs via `POST /jobs` — response includes `checkout_url` +Subscribe to `checkout.session.completed`. A verified event records revenue idempotently and lets the worker continue fulfillment. -Polling is disabled by default. For CLI-only test flows: +For a CLI-only polling flow, explicitly enable the legacy poller: ```bash export SOLVENT_ALLOW_POLL=1 @@ -54,57 +54,67 @@ export SOLVENT_ALLOW_POLL=1 ## Submit a job ```bash -curl -X POST http://127.0.0.1:8787/jobs \ +curl -X POST https://solvent.example.com/jobs \ -H 'Content-Type: application/json' \ - -d '{"id":"J99","topic":"EV battery supply chain","budget_cents":7500,"customer_email":"you@example.com"}' + -d '{ + "id": "J99", + "topic": "EV battery supply chain", + "budget_cents": 7500, + "customer_email": "you@example.com" + }' ``` -## Email delivery (optional) +The response includes the job status and checkout URL. Jobs remain `awaiting_payment` until Stripe confirms payment. + +## Delivery -Without SMTP, briefs are written to `data/outbox/{job_id}.eml`. +Signed brief URLs are generated from `SOLVENT_BASE_URL` and `SOLVENT_DELIVERY_SECRET`. + +Without SMTP, messages are written to the local outbox. To send email: ```bash export SMTP_HOST=smtp.example.com export SMTP_PORT=587 export SMTP_USER=... export SMTP_PASS=... -export SMTP_FROM=agent@yourdomain.com +export SMTP_FROM=agent@example.com ``` ## Operations ```bash -# Structured JSON logs -export SOLVENT_LOG_JSON=1 - -# Reconcile Stripe vs ledger -python3 -m solvent reconcile --since 7d - -# Auto-improvement (dry-run by default) -python3 -m solvent tune -python3 -m solvent tune --apply +solvent status +solvent jobs +solvent jobs retry +solvent finance +solvent reconcile --since 7d +solvent tune +solvent tune --apply +solvent webhooks stats +solvent webhooks failed +solvent doctor ``` -## Environment variables +## Runtime data -| Variable | Purpose | -|----------|---------| -| `STRIPE_API_KEY` | Test/restricted Stripe key | -| `STRIPE_WEBHOOK_SECRET` | Webhook signature verification | -| `SOLVENT_BASE_URL` | Checkout success URLs + hosted briefs | -| `SOLVENT_DELIVERY_SECRET` | HMAC token for `/briefs/{id}`; required, at least 32 characters, high entropy, and not a placeholder | -| `SOLVENT_ASYNC` | Non-blocking payment (worker resumes jobs) | -| `SOLVENT_ALLOW_POLL` | Legacy payment polling | -| `SOLVENT_LOG_JSON` | JSON lines to stderr + `data/solvent.log` | -| `NVIDIA_API_KEY` | Live Nemotron (optional) | -| `SMTP_*` | Email delivery | - -## Architecture +All runtime artifacts honor `SOLVENT_HOME`: -``` -POST /jobs → quote → Checkout Session → awaiting_payment - ↓ webhook checkout.session.completed -worker → paid → fulfill (tool agent) → deliver → spend → book +```bash +export SOLVENT_HOME=/var/lib/solvent ``` -Idempotent stages are recorded in SQLite (`job_stages`). Metrics in `job_metrics` feed `solvent tune`. +That directory contains the treasury database, webhook log, reports, dashboard data, logs, and outboxes. Back it up and mount it on persistent storage. + +## Relevant environment variables + +| Variable | Purpose | +|---|---| +| `SOLVENT_HOME` | Runtime data root | +| `SOLVENT_BASE_URL` | Public URL used in checkout and delivery links | +| `SOLVENT_DELIVERY_SECRET` | HMAC secret for signed brief URLs | +| `STRIPE_API_KEY` | Stripe test or restricted test key | +| `STRIPE_WEBHOOK_SECRET` | Stripe webhook signature secret | +| `SOLVENT_ALLOW_POLL` | Enable legacy CLI payment polling | +| `SOLVENT_LOG_JSON` | Emit structured JSON logs | +| `NVIDIA_API_KEY` | Enable live Nemotron inference | +| `SMTP_*` | Optional email delivery | diff --git a/docs/TELEGRAM.md b/docs/TELEGRAM.md index ac263b1..84f46ea 100644 --- a/docs/TELEGRAM.md +++ b/docs/TELEGRAM.md @@ -1,124 +1,79 @@ -# Telegram + SOLVENT +# Telegram channel -Talk to your self-funding research agent on Telegram. Pairing, conversational commissioning, and job lifecycle notifications are implemented with OpenClaw-style channel security and Hermes-style tool/memory patterns — all in Python, no external gateway runtime. +The Telegram adapter uses the same gateway, memory, pricing, checkout, and guardrails as the dashboard and CLI. -## Prerequisites - -1. Create a bot via [@BotFather](https://t.me/BotFather) and copy the token. -2. Install Telegram dependencies: - -```bash -pip install -r requirements-telegram.txt -``` - -3. Export the token (never commit it): +## Install and configure ```bash +pip install -e ".[telegram,serve,stripe]" export TELEGRAM_BOT_TOKEN="123456:ABC..." ``` -4. Optional: set DM policy (default `pairing`): +Create the token with [@BotFather](https://t.me/BotFather). Keep it in the environment and never commit it. -```bash -export SOLVENT_TELEGRAM_DM_POLICY=pairing # pairing | allowlist | open -``` - -## Run the stack - -In separate terminals: +The default direct-message policy is pairing: ```bash -python -m solvent serve # Stripe webhooks + checkout + browser dashboard -python -m solvent worker # fulfill paid jobs -python -m solvent telegram # long-poll bot +export SOLVENT_TELEGRAM_DM_POLICY=pairing ``` -`serve` also hosts the **interactive dashboard** at `http://127.0.0.1:8787/` — typed chat, Web Speech mic input, and live treasury updates via SSE. Job lifecycle messages from `worker` reach the dashboard via `data/chat_outbox.jsonl` when `serve` is running. +Supported values are `pairing`, `allowlist`, and `open`. For allowlist mode, set `SOLVENT_TELEGRAM_ALLOW_FROM` to comma-separated Telegram user IDs. + +## Run -Verify configuration: +Start each long-running process separately: ```bash -python -m solvent doctor +solvent serve # checkout, webhooks, browser dashboard +solvent worker # paid-job fulfillment +solvent telegram # Telegram long polling ``` -## Pairing (OpenClaw pattern) - -Unknown users messaging the bot receive a code: - -1. User sends `/start` → bot replies with `TG-XXXXXX` -2. Operator approves: +Verify local configuration with: ```bash -python -m solvent pairing list -python -m solvent pairing approve TG-XXXXXX +solvent doctor ``` -Approved user IDs are stored in `.solvent/telegram_allowlist.json` (gitignored). +## Pair users -## Example conversation +With the default policy, an unknown user sends `/start` and receives a code. Approve it from the operator terminal: +```bash +solvent pairing list +solvent pairing approve TG-XXXXXX ``` -You: What's my treasury balance? -Bot: [uses treasury_status tool] Balance $100.00, margin 94%... -You: I want a research brief on EV battery supply chains, budget $50, email me@co.com -Bot: Margin looks good. Confirm to proceed? -You: Yes, submit it -Bot: Checkout: https://checkout.stripe.com/... +Approved IDs are stored in the local SOLVENT configuration directory and are not committed. -[payment completes via worker] -Bot: Payment received for job Tabc12345. Fulfillment starting. -Bot: Your brief for Tabc12345 is ready: http://127.0.0.1:8787/brief/... -``` - -## Slash commands +## Commands | Command | Action | -|---------|--------| +|---|---| | `/help` | Usage summary | | `/status` | Treasury snapshot | | `/jobs` | Recent jobs | | `/quote topic \| 50.00` | Margin preview | +| `/pair qr` | Create a short-lived OpenClaw pairing token | -Free-form chat also works — Nemotron uses business tools (`quote_brief`, `submit_brief`) behind guardrails. - -## Agent workspace (brain + soul) - -Chat loads `.solvent/workspace/` each turn: +Free-form conversation can quote and submit research briefs. `submit_brief` still passes through the same margin gate and checkout stages as every other channel. -- **SOUL.md** — identity and tone (Hermes slot #1) -- **BRAIN.md** — active pipeline / what needs attention now -- **AGENTS.md** — operating rules - -```bash -python -m solvent workspace setup -python -m solvent workspace list -``` +## Security properties -See **[WORKSPACE.md](WORKSPACE.md)** for the full file map. +- Pairing is required by default. +- Messages are rate-limited per user. +- The bot token is read only from the environment. +- Stripe live-mode keys are refused. +- Telegram cannot bypass treasury, pricing, payment, or spend-policy code. +- Outbound text is capped to Telegram's message limit. -## Security +## Agent workspace -- Bot token only in environment variables -- Default DM policy requires pairing before chat -- Rate limit: 30 messages/user/hour -- `submit_brief` runs the same margin gate + checkout stages as the CLI agent -- Live Stripe keys (`sk_live_*`) are refused - -## Config file - -`.solvent/config.json` supports: - -```json -{ - "telegram_enabled": true, - "telegram_dm_policy": "pairing", - "telegram_allow_from": ["123456789"] -} -``` - -Re-run onboarding to set Telegram options: +Chat identity and operating context come from the local workspace: ```bash -python3 run_demo.py --onboard +solvent workspace setup +solvent workspace list ``` + +See [WORKSPACE.md](WORKSPACE.md) for the workspace file map. diff --git a/pyproject.toml b/pyproject.toml index 26ad3d2..ea5877c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,12 +26,13 @@ rich = ["rich>=13.0.0"] serve = ["fastapi>=0.110.0", "uvicorn[standard]>=0.27.0"] qr = ["qrcode[pil]>=7.0"] telegram = ["python-telegram-bot>=21.0"] -dev = ["pytest>=7.0.0"] +dev = ["pytest>=7.0.0", "httpx>=0.27.0"] all = [ "stripe>=9.0.0", "rich>=13.0.0", "fastapi>=0.110.0", "uvicorn[standard]>=0.27.0", + "qrcode[pil]>=7.0", "python-telegram-bot>=21.0", ] diff --git a/run_demo.py b/run_demo.py index 46b61fd..f78ef6f 100644 --- a/run_demo.py +++ b/run_demo.py @@ -1,7 +1,6 @@ #!/usr/bin/env python3 -"""Compatibility wrapper for running the SOLVENT demo from a source checkout.""" +"""Run the SOLVENT demo from a source checkout.""" -from solvent.cli import * # noqa: F403 from solvent.cli import main diff --git a/solvent/__main__.py b/solvent/__main__.py index 351b713..fcd7a85 100644 --- a/solvent/__main__.py +++ b/solvent/__main__.py @@ -1,6 +1,11 @@ -"""SOLVENT CLI entry point: python3 -m solvent [serve|worker|telegram|doctor|pairing|...]""" +"""SOLVENT command-line entry point.""" +from __future__ import annotations + +from importlib import import_module import sys +from typing import Final + HELP = """\ SOLVENT — a self-funding analyst agent. @@ -12,8 +17,9 @@ (none) run the batch demo / interactive session init first-run setup: create dirs, DB, and workspace files status live summary: balance, jobs, API key presence; --watch to auto-refresh - upgrade check for newer version on PyPI; --check exits 1 if outdated - jobs list/show/retry/cancel jobs (jobs --help for sub-commands) + upgrade check for a newer version on PyPI; --check exits 1 if outdated + jobs list/show/retry/cancel jobs (jobs --help for subcommands) + retry compatibility alias for jobs retry logs tail the structured event log; -f to follow, --job/--stage to filter config show/get/set/reset local configuration values serve webhooks + job API + hosted briefs @@ -25,123 +31,79 @@ doctor stack diagnostics pairing manage Telegram DM pairing codes workspace seed the agent workspace (SOUL/BRAIN/AGENTS) - retry re-run a stuck job webhooks inspect the webhook event log (stats|list|failed) + tui live terminal dashboard (requires the rich extra) help show this message version print the installed version Run `solvent --help` where supported for command-specific options. """ +# Keep optional features lazily imported: the zero-dependency core must remain +# importable without FastAPI, Stripe, Rich, or Telegram installed. +_COMMANDS: Final[dict[str, tuple[str, str]]] = { + "init": ("init", "main"), + "status": ("status", "main"), + "jobs": ("job_cmd", "main"), + "upgrade": ("upgrade", "main"), + "logs": ("logs", "main"), + "config": ("config_cmd", "main"), + "serve": ("server", "main"), + "worker": ("worker", "main"), + "tune": ("improver", "main"), + "reconcile": ("reconcile", "main"), + "doctor": ("doctor", "main"), + "telegram": ("channels.telegram", "main"), + "pairing": ("pairing", "main"), + "workspace": ("workspace", "main"), + "tui": ("tui", "run"), + "finance": ("finance", "main"), + "report": ("finance", "main"), + "webhooks": ("webhook_log", "main"), +} -def _print_version(): + +def _print_version() -> None: from . import __version__ + print(f"solvent {__version__}") +def _dispatch(command: str): + module_name, function_name = _COMMANDS[command] + sys.argv.pop(1) + handler = getattr(import_module(f"solvent.{module_name}"), function_name) + return handler() + + +def _run_demo(): + from .cli import main as demo_main + + return demo_main() + + def main(): - if len(sys.argv) > 1 and sys.argv[1] in ("version", "--version", "-V"): - _print_version() - return - if len(sys.argv) > 1 and sys.argv[1] in ("help", "--help", "-h"): + if len(sys.argv) == 1: + return _run_demo() + + command = sys.argv[1] + if command in ("version", "--version", "-V"): + return _print_version() + if command in ("help", "--help", "-h"): print(HELP) - return - if len(sys.argv) > 1 and sys.argv[1] == "init": - sys.argv.pop(1) - from .init import main as init_main - init_main() - elif len(sys.argv) > 1 and sys.argv[1] == "status": - sys.argv.pop(1) - from .status import main as status_main - status_main() - elif len(sys.argv) > 1 and sys.argv[1] == "jobs": - sys.argv.pop(1) - from .job_cmd import main as jobs_main - jobs_main() - elif len(sys.argv) > 1 and sys.argv[1] == "upgrade": - sys.argv.pop(1) - from .upgrade import main as upgrade_main - upgrade_main() - elif len(sys.argv) > 1 and sys.argv[1] == "logs": - sys.argv.pop(1) - from .logs import main as logs_main - logs_main() - elif len(sys.argv) > 1 and sys.argv[1] == "config": - sys.argv.pop(1) - from .config_cmd import main as config_main - config_main() - elif len(sys.argv) > 1 and sys.argv[1] == "serve": - sys.argv.pop(1) - from .server import main as serve_main - serve_main() - elif len(sys.argv) > 1 and sys.argv[1] == "worker": - sys.argv.pop(1) - from .worker import main as worker_main - worker_main() - elif len(sys.argv) > 1 and sys.argv[1] == "tune": - sys.argv.pop(1) - from .improver import main as tune_main - tune_main() - elif len(sys.argv) > 1 and sys.argv[1] == "reconcile": - sys.argv.pop(1) - from .reconcile import main as reconcile_main - reconcile_main() - elif len(sys.argv) > 1 and sys.argv[1] == "doctor": - sys.argv.pop(1) - from .doctor import main as doctor_main - doctor_main() - elif len(sys.argv) > 1 and sys.argv[1] == "telegram": - sys.argv.pop(1) - from .channels.telegram import main as telegram_main - telegram_main() - elif len(sys.argv) > 1 and sys.argv[1] == "pairing": - sys.argv.pop(1) - from .pairing import main as pairing_main - pairing_main() - elif len(sys.argv) > 1 and sys.argv[1] == "workspace": - sys.argv.pop(1) - from .workspace import main as workspace_main - workspace_main() - elif len(sys.argv) > 1 and sys.argv[1] == "tui": - from .tui import run - run() - elif len(sys.argv) > 1 and sys.argv[1] == "retry": - job_id = sys.argv[2] if len(sys.argv) > 2 else None - if not job_id: - print("Usage: python -m solvent retry ") - sys.exit(1) - from .stages import SolventStages - from .treasury import Treasury - from .guardrails import Guardrails - from .stripe_client import StripeClient - t = Treasury() - s = SolventStages(treasury=t, guard=Guardrails(t), stripe=StripeClient()) - result = s.retry_job(job_id) - import json - print(json.dumps(result, indent=2, default=str)) - elif len(sys.argv) > 1 and sys.argv[1] in ("finance", "report"): - sys.argv.pop(1) - from .finance import main as finance_main - finance_main() - elif len(sys.argv) > 1 and sys.argv[1] == "webhooks": - from .webhook_log import WebhookLog - import json - wl = WebhookLog() - sub = sys.argv[2] if len(sys.argv) > 2 else "stats" - if sub == "stats": - print(json.dumps(wl.stats(), indent=2)) - elif sub == "list": - for row in wl.list_recent(20): - print(f"{row['received_at_fmt']} [{row['status']}] {row['event_type']} {row['event_id'][:16]}") - elif sub == "failed": - for row in wl.list_failed(): - print(f" {row['event_id'][:16]} {row['event_type']} err={row['error'][:60]}") - else: - # No subcommand: fall through to the demo / interactive CLI. - from .upgrade import background_update_hint - background_update_hint() - from .cli import main as demo_main - demo_main() + return None + + # Preserve the old top-level spelling without maintaining a second retry + # implementation. job_cmd owns all job operations. + if command == "retry": + sys.argv[1:2] = ["jobs", "retry"] + command = "jobs" + + if command in _COMMANDS: + return _dispatch(command) + + # Demo options such as --interactive and --seed belong to solvent.cli. + return _run_demo() if __name__ == "__main__": diff --git a/solvent/agent.py b/solvent/agent.py index 56b7227..8a59e55 100644 --- a/solvent/agent.py +++ b/solvent/agent.py @@ -1,25 +1,16 @@ -""" -agent.py — the SOLVENT orchestrator. - -For each inbound job the agent runs an idempotent stage machine: - 1. QUOTES through the margin gate - 2. EARNS via Stripe Checkout (webhook-first; sync confirm in demo mode) - 3. FULFILS via bounded Nemotron tool-calling - 4. DELIVERS hosted brief + optional email - 5. SPENDS on vendors (guardrail-screened) - 6. BOOKS P&L into the treasury -""" +"""Public SOLVENT orchestrator over the idempotent stage machine.""" from __future__ import annotations import os import time +from collections.abc import Callable -from .treasury import Treasury from .guardrails import Guardrails from .pricing import PricingPolicy -from .stripe_client import StripeClient from .stages import StageRunner, validate_and_coerce_job +from .stripe_client import StripeClient +from .treasury import Treasury class Solvent: @@ -27,7 +18,7 @@ def __init__( self, seed_cents: int = 10_000, fresh: bool = True, - on_event: callable | None = None, + on_event: Callable[[dict], None] | None = None, *, sync_payment: bool | None = None, ): @@ -35,13 +26,20 @@ def __init__( if fresh: self.t.reset() self.t.seed(seed_cents) + self.guard = Guardrails(self.t) self.stripe = StripeClient() self.pricing = PricingPolicy() self.log: list[dict] = [] self.on_event = on_event + if sync_payment is None: - sync_payment = os.environ.get("SOLVENT_ASYNC", "").strip() not in ("1", "true", "yes") + sync_payment = os.environ.get("SOLVENT_ASYNC", "").strip() not in ( + "1", + "true", + "yes", + ) + self._runner = StageRunner( self.t, self.guard, @@ -51,19 +49,16 @@ def __init__( sync_payment=sync_payment, ) - def _capture_event(self, event: dict): - self.log.append(event) - if self.on_event: - self.on_event(event) - - def _emit(self, **event): - if "ts" not in event: - event["ts"] = time.time() + def _capture_event(self, event: dict) -> dict: self.log.append(event) if self.on_event: self.on_event(event) return event + def _emit(self, **event) -> dict: + event.setdefault("ts", time.time()) + return self._capture_event(event) + def handle_job(self, job: dict) -> dict: return self._runner.run_job(job) @@ -71,15 +66,16 @@ def advance_job(self, job_id: str) -> dict: return self._runner.advance_job(job_id) def enqueue_job(self, job: dict) -> dict: - """Validate and persist a job for async worker processing.""" - job, err = validate_and_coerce_job(job, self.t) - if err: - return self._emit(stage="declined", job_id=job.get("id", "unknown") if job else "unknown", reason=err) - q = self._runner._stage_quote(job) - if q.get("stage") == "declined" or not q.get("accept"): - return q - checkout = self._runner._stage_checkout(job, q) - return checkout + """Validate and persist a job for asynchronous worker processing.""" + job, error = validate_and_coerce_job(job, self.t) + if error: + job_id = job.get("id", "unknown") if job else "unknown" + return self._emit(stage="declined", job_id=job_id, reason=error) + + quote_result = self._runner._stage_quote(job) + if quote_result.get("stage") == "declined" or not quote_result.get("accept"): + return quote_result + return self._runner._stage_checkout(job, quote_result) def run(self, jobs: list[dict]) -> dict: for job in jobs: diff --git a/solvent/channels/telegram.py b/solvent/channels/telegram.py index 602e059..d7ff22c 100644 --- a/solvent/channels/telegram.py +++ b/solvent/channels/telegram.py @@ -10,15 +10,19 @@ def send_telegram_message(external_id: str, text: str) -> None: - """Sync outbound via Telegram HTTP API (safe from worker threads).""" + """Send an outbound message through Telegram's HTTP API.""" token = os.environ.get("TELEGRAM_BOT_TOKEN", "").strip() if not token: return url = f"https://api.telegram.org/bot{token}/sendMessage" payload = json.dumps({"chat_id": int(external_id), "text": text[:4000]}).encode() - req = urllib.request.Request(url, data=payload, headers={"Content-Type": "application/json"}) + request = urllib.request.Request( + url, + data=payload, + headers={"Content-Type": "application/json"}, + ) try: - urllib.request.urlopen(req, timeout=15) + urllib.request.urlopen(request, timeout=15) except Exception: pass @@ -26,11 +30,19 @@ def send_telegram_message(external_id: str, text: str) -> None: def _require_ptb(): try: from telegram import Update - from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes + from telegram.ext import ( + Application, + CommandHandler, + ContextTypes, + MessageHandler, + filters, + ) + return Update, Application, CommandHandler, MessageHandler, filters, ContextTypes except ImportError as exc: raise RuntimeError( - "python-telegram-bot required. Install: pip install -r requirements-telegram.txt" + "python-telegram-bot is required. Install with: " + "pip install 'solvent-agent[telegram]'" ) from exc @@ -45,7 +57,7 @@ def build_application(gateway: Gateway | None = None) -> object: if not token: raise RuntimeError("TELEGRAM_BOT_TOKEN not set") - gw = gateway or Gateway() + gateway = gateway or Gateway() register_outbound("telegram", send_telegram_message) async def on_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: @@ -53,16 +65,17 @@ async def on_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None return user = update.effective_user text = update.message.text or "" - username = user.username - reply = gw.handle_inbound("telegram", str(user.id), text, user_label=username) + reply = gateway.handle_inbound( + "telegram", + str(user.id), + text, + user_label=user.username, + ) await _reply(update, reply) app = Application.builder().token(token).build() - app.add_handler(CommandHandler("start", on_message)) - app.add_handler(CommandHandler("help", on_message)) - app.add_handler(CommandHandler("status", on_message)) - app.add_handler(CommandHandler("jobs", on_message)) - app.add_handler(CommandHandler("quote", on_message)) + for command in ("start", "help", "status", "jobs", "quote"): + app.add_handler(CommandHandler(command, on_message)) app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, on_message)) return app diff --git a/solvent/cli.py b/solvent/cli.py index 6e30261..be87494 100644 --- a/solvent/cli.py +++ b/solvent/cli.py @@ -1,25 +1,14 @@ #!/usr/bin/env python3 -""" -solvent.cli — watch SOLVENT run as a business. +"""Run SOLVENT's offline batch demo or interactive terminal flow.""" -Runs four inbound jobs through the agent or starts an interactive session. -You'll see the agent quote jobs, decline unprofitable ones, collect Stripe payments, -run Nemotron reasoning, screen payments through guardrails, and book the P&L. -Outputs a premium, interactive treasury_dashboard.html at the end. +from __future__ import annotations -First run: interactive onboarding wizard (see ONBOARDING.md). -Skip wizard: --no-onboard or SOLVENT_SKIP_ONBOARD=1 -Reconfigure: --onboard -""" - -import sys -import time import argparse +import sys from pathlib import Path -from solvent.agent import Solvent -from solvent.jobs import SAMPLE_JOBS -from solvent.treasury import fmt + from solvent import dashboard +from solvent.agent import Solvent from solvent.config import ( SolventConfig, apply_config, @@ -27,111 +16,123 @@ default_config, load_config, ) +from solvent.jobs import SAMPLE_JOBS from solvent.onboarding import run_wizard, should_skip_onboarding, wants_reconfigure +from solvent.treasury import fmt + -# ANSI terminal colors (zero dependencies) C_RESET = "\033[0m" C_BOLD = "\033[1m" C_RED = "\033[31m" C_GREEN = "\033[32m" C_YELLOW = "\033[33m" C_BLUE = "\033[34m" -C_MAGENTA = "\033[35m" C_CYAN = "\033[36m" C_GREY = "\033[90m" BAR = f"{C_GREY}{'─' * 66}{C_RESET}" -def show_spinner(duration: float, label: str): - """Render a clean animated terminal spinner.""" - frames = ["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"] - end_time = time.time() + duration - i = 0 - sys.stdout.write(f" {C_GREY}{label}{C_RESET} ") - sys.stdout.flush() - while time.time() < end_time: - sys.stdout.write(f"\r {C_GREY}{label}{C_RESET} {C_CYAN}{frames[i % len(frames)]}{C_RESET}") - sys.stdout.flush() - time.sleep(0.08) - i += 1 - sys.stdout.write(f"\r {C_GREY}{label}{C_RESET} {C_GREEN}Done!{C_RESET}\n") - sys.stdout.flush() - - -def print_event(e: dict): - """Callback function mapping Agent logs to beautifully colorized terminal stdout.""" - st = e.get("stage") - jid = e.get("job_id", "") - prefix = f"{C_CYAN}[{jid}]{C_RESET}" - - if st == "quote": - verdict = f"{C_GREEN}ACCEPT{C_RESET}" if e["accept"] else f"{C_RED}DECLINE{C_RESET}" - print(f" ⚖️ {prefix} Margin Gate: price {C_BOLD}{fmt(e['price'])}{C_RESET} | est cost {fmt(e['est_cost'])} | projected margin {e['margin_pct']}% → {verdict}") - time.sleep(0.4) - - elif st == "declined": - print(f" ✋ {prefix} {C_RED}Declined:{C_RESET} {e['reason']}") - time.sleep(0.4) - - elif st == "invoice": - show_spinner(0.6, f"Generating Stripe Payment Link for {jid}...") - tag = "simulated" if e["simulated"] else "LIVE" +def print_step(label: str) -> None: + """Print a completed demo step without artificial delays or animation.""" + print(f" {C_GREY}{label}{C_RESET} {C_GREEN}Done!{C_RESET}") + + +def print_event(event: dict) -> None: + """Render a StageRunner event for the terminal demo.""" + stage = event.get("stage") + job_id = event.get("job_id", "") + prefix = f"{C_CYAN}[{job_id}]{C_RESET}" + + if stage == "quote": + verdict = f"{C_GREEN}ACCEPT{C_RESET}" if event["accept"] else f"{C_RED}DECLINE{C_RESET}" + print( + f" ⚖️ {prefix} Margin Gate: price {C_BOLD}{fmt(event['price'])}{C_RESET} " + f"| est cost {fmt(event['est_cost'])} | projected margin " + f"{event['margin_pct']}% → {verdict}" + ) + elif stage == "declined": + print(f" ✋ {prefix} {C_RED}Declined:{C_RESET} {event['reason']}") + elif stage == "invoice": + print_step(f"Generating Stripe Payment Link for {job_id}...") + tag = "simulated" if event["simulated"] else "LIVE" print(f" 📄 {prefix} Issued checkout link [{tag}]") - print(f" {C_GREY}URL:{C_RESET} {C_BLUE}{e['url']}{C_RESET}") - time.sleep(0.4) - - elif st == "paid": - show_spinner(0.8, f"Polling Stripe invoice confirmation for {jid}...") - print(f" 💵 {prefix} {C_GREEN}Stripe Confirmed:{C_RESET} Payment of {C_GREEN}+{fmt(e['amount'])}{C_RESET} received") - time.sleep(0.4) - - elif st == "fulfilled": - show_spinner(1.2, f"Nemotron compiling sell-side research brief for {jid}...") - filename = Path(e['deliverable']).name - print(f" 📝 {prefix} {C_GREEN}Fulfillment Finished:{C_RESET} Deliverable saved to {C_BLUE}data/reports/{filename}{C_RESET} ({e['tokens']} tokens)") - time.sleep(0.4) - - elif st == "spend": - time.sleep(0.3) - print(f" 🛡️ {C_GREY}↳ Spend Approved:{C_RESET} Scoped payment {C_RED}−{fmt(e['amount'])}{C_RESET} to {C_BOLD}{e['vendor']}{C_RESET} ({e['memo']})") - - elif st == "spend_blocked": - time.sleep(0.3) - print(f" 🛑 {C_RED}↳ Spend BLOCKED:{C_RESET} Guardrail rejected transaction of {fmt(e['amount'])} to {e['vendor']} ({e['memo']})") - - elif st == "refunded": - time.sleep(0.3) - print(f" ↩️ {prefix} {C_RED}Escrow Refunded:{C_RESET} Returned {C_RED}{fmt(e['amount'])}{C_RESET} to customer ({e['reason']})") - - elif st == "booked": - - tag = f"{C_GREEN}profit{C_RESET}" if e["job_pnl"] >= 0 else f"{C_RED}loss{C_RESET}" - time.sleep(0.4) - print(f" ✓ {prefix} booked P&L: {tag} {C_BOLD}{fmt(e['job_pnl'])}{C_RESET} | treasury cash balance: {C_YELLOW}{fmt(e['balance'])}{C_RESET}") - print() - - -def print_results(snap: dict): - """Print overall financial results summary.""" + print(f" {C_GREY}URL:{C_RESET} {C_BLUE}{event['url']}{C_RESET}") + elif stage == "paid": + print_step(f"Confirming Stripe payment for {job_id}...") + print( + f" 💵 {prefix} {C_GREEN}Stripe Confirmed:{C_RESET} " + f"Payment of {C_GREEN}+{fmt(event['amount'])}{C_RESET} received" + ) + elif stage == "fulfilled": + print_step(f"Compiling research brief for {job_id}...") + filename = Path(event["deliverable"]).name + print( + f" 📝 {prefix} {C_GREEN}Fulfillment Finished:{C_RESET} " + f"Deliverable saved to {C_BLUE}data/reports/{filename}{C_RESET} " + f"({event['tokens']} tokens)" + ) + elif stage == "spend": + print( + f" 🛡️ {C_GREY}↳ Spend Approved:{C_RESET} Scoped payment " + f"{C_RED}−{fmt(event['amount'])}{C_RESET} to " + f"{C_BOLD}{event['vendor']}{C_RESET} ({event['memo']})" + ) + elif stage == "spend_blocked": + print( + f" 🛑 {C_RED}↳ Spend BLOCKED:{C_RESET} Guardrail rejected " + f"transaction of {fmt(event['amount'])} to {event['vendor']} " + f"({event['memo']})" + ) + elif stage == "refunded": + print( + f" ↩️ {prefix} {C_RED}Escrow Refunded:{C_RESET} Returned " + f"{C_RED}{fmt(event['amount'])}{C_RESET} to customer " + f"({event['reason']})" + ) + elif stage == "booked": + tag = f"{C_GREEN}profit{C_RESET}" if event["job_pnl"] >= 0 else f"{C_RED}loss{C_RESET}" + print( + f" ✓ {prefix} booked P&L: {tag} " + f"{C_BOLD}{fmt(event['job_pnl'])}{C_RESET} | treasury cash balance: " + f"{C_YELLOW}{fmt(event['balance'])}{C_RESET}\n" + ) + + +def print_results(snapshot: dict) -> None: + """Print the financial result of a demo session.""" print(BAR) print(f"📊 {C_BOLD}SESSION SUMMARY & BALANCE SHEET{C_RESET}") - print(f" Revenue {C_GREEN}{fmt(snap['revenue_cents'])}{C_RESET}") - print(f" Operating spend {C_RED}{fmt(snap['expense_cents'])}{C_RESET}") - print(f" Net profit {C_GREEN if snap['net_profit_cents']>=0 else C_RED}{fmt(snap['net_profit_cents'])}{C_RESET} ({snap['margin_pct']}% margin)") - print(f" Cash balance {C_YELLOW}{fmt(snap['balance_cents'])}{C_RESET} (seed was {fmt(snap['capital_cents'])})") - - grew = snap['balance_cents'] - snap['capital_cents'] - if grew > 0: - print(f" {C_GREEN}→ The agent grew its own treasury by {fmt(grew)} this session!{C_RESET}") - elif grew < 0: - print(f" {C_RED}→ The agent ran at a loss of {fmt(abs(grew))} this session.{C_RESET}") + print(f" Revenue {C_GREEN}{fmt(snapshot['revenue_cents'])}{C_RESET}") + print(f" Operating spend {C_RED}{fmt(snapshot['expense_cents'])}{C_RESET}") + profit_color = C_GREEN if snapshot["net_profit_cents"] >= 0 else C_RED + print( + f" Net profit {profit_color}{fmt(snapshot['net_profit_cents'])}{C_RESET} " + f"({snapshot['margin_pct']}% margin)" + ) + print( + f" Cash balance {C_YELLOW}{fmt(snapshot['balance_cents'])}{C_RESET} " + f"(seed was {fmt(snapshot['capital_cents'])})" + ) + + growth = snapshot["balance_cents"] - snapshot["capital_cents"] + if growth > 0: + print(f" {C_GREEN}→ The agent grew its treasury by {fmt(growth)}.{C_RESET}") + elif growth < 0: + print(f" {C_RED}→ The agent ran at a loss of {fmt(abs(growth))}.{C_RESET}") else: - print(" → The agent broke perfectly even.") + print(" → The agent broke even.") + + +def _finish_session(agent: Solvent) -> None: + snapshot = agent.t.snapshot() + print_results(snapshot) + path = dashboard.render(snapshot, agent.log) + print(f"\n Dashboard: {C_BLUE}{path}{C_RESET}\n{BAR}\n") -def run_batch_demo(seed_cents: int = 10_000, fresh: bool = True): - """Run the standard batch demo of 4 predefined jobs.""" +def run_batch_demo(seed_cents: int = 10_000, fresh: bool = True) -> None: + """Run the four predefined jobs through the complete money loop.""" print(f"\n🪙 {C_BOLD}SOLVENT — Standard Batch Run (4 Inbound Jobs){C_RESET}") print(BAR) @@ -140,23 +141,65 @@ def run_batch_demo(seed_cents: int = 10_000, fresh: bool = True): for job in SAMPLE_JOBS: print(f"{C_BOLD}■ {job['id']}: {job['topic']}{C_RESET}") - show_spinner(0.4, "Analyzing inbound job specifications...") + print_step("Analyzing inbound job specifications...") agent.handle_job(job) - snap = agent.t.snapshot() - print_results(snap) - path = dashboard.render(snap, agent.log) - print(f"\n Dashboard: {C_BLUE}{path}{C_RESET}\n{BAR}\n") + _finish_session(agent) + + +def _read_positive_cents(prompt: str) -> int | None: + raw = input(prompt).strip() + try: + cents = int(float(raw) * 100) + except (ValueError, OverflowError): + print(f"{C_RED}Invalid numeric entry. Use a value such as 49.00.{C_RESET}\n") + return None + if cents <= 0: + print(f"{C_RED}Amount must be greater than zero.{C_RESET}\n") + return None + return cents + + +def _fund(agent: Solvent, command: str) -> bool: + """Handle /fund and return True when the input was a funding command.""" + if not command.startswith("/fund"): + return False + parts = command.split() + if len(parts) != 2: + print(f"{C_RED}Usage: /fund (for example, /fund 150.00){C_RESET}\n") + return True + try: + cents = int(float(parts[1]) * 100) + except (ValueError, OverflowError): + cents = 0 + if cents <= 0: + print(f"{C_RED}Fund amount must be positive.{C_RESET}\n") + return True + + agent.t.seed(cents, memo="User injected operating capital") + print( + f" 💵 {C_GREEN}Deposit Confirmed:{C_RESET} Added " + f"{C_GREEN}+{fmt(cents)}{C_RESET} of operating capital." + ) + agent._emit( + stage="booked", + job_id="SYSTEM", + job_pnl=0, + balance=agent.t.balance_cents(), + status="completed", + title="User Capital Injection", + ) + return True -def run_interactive_mode(seed_cents: int = 10_000, fresh: bool = True): - """Run interactive CLI mode letting the user test custom briefs and pricing.""" +def run_interactive_mode(seed_cents: int = 10_000, fresh: bool = True) -> None: + """Accept custom research jobs from stdin until the user stops.""" print(f"\n🪙 {C_BOLD}SOLVENT — Interactive Agent Terminal{C_RESET}") print(BAR) agent = Solvent(seed_cents=seed_cents, fresh=fresh, on_event=print_event) print(f" Current balance: {C_YELLOW}{fmt(agent.t.balance_cents())}{C_RESET}") - print(f" {C_GREY}Tip: Type /fund (e.g. /fund 100) to add funds to the treasury.{C_RESET}\n") + print(f" {C_GREY}Tip: type /fund 100 to add operating capital.{C_RESET}\n") job_index = 1 while True: @@ -165,108 +208,56 @@ def run_interactive_mode(seed_cents: int = 10_000, fresh: bool = True): if not topic: print(f"{C_RED}Request topic cannot be blank.{C_RESET}\n") continue - - if topic.startswith("/fund"): - parts = topic.split() - if len(parts) < 2: - print(f"{C_RED}Usage: /fund (e.g., /fund 150.00){C_RESET}\n") - continue - try: - fund_amt = float(parts[1]) - fund_cents = int(fund_amt * 100) - if fund_cents <= 0: - print(f"{C_RED}Fund amount must be positive.{C_RESET}\n") - continue - # Add capital to treasury - agent.t.seed(fund_cents, memo="User injected operating capital") - print(f" 💵 {C_GREEN}Deposit Confirmed:{C_RESET} Added {C_GREEN}+{fmt(fund_cents)}{C_RESET} of operating capital to treasury.") - # Emit booked event to trigger dashboard update - agent._emit( - stage="booked", - job_id="SYSTEM", - job_pnl=0, - balance=agent.t.balance_cents(), - status="completed", - title="User Capital Injection" - ) - print(f" Current Capital Balance: {C_YELLOW}{fmt(agent.t.balance_cents())}{C_RESET}\n") - continue - except ValueError: - print(f"{C_RED}Invalid amount. Usage: /fund {C_RESET}\n") - continue - - budget_str = input(f"{C_CYAN}Client Budget in USD (e.g. 50.00):{C_RESET} $").strip() - try: - budget_cents = int(float(budget_str) * 100) - if budget_cents <= 0: - print(f"{C_RED}Budget must be greater than 0.{C_RESET}\n") - continue - except ValueError: - print(f"{C_RED}Invalid numeric entry. Use standard formats like 49.00.{C_RESET}\n") + if _fund(agent, topic): continue - job_id = f"I{job_index}" - job_index += 1 - - # Scale resource specifications based on budget - is_large = budget_cents >= 5000 - est_tokens = 12000 if is_large else 7500 - market_calls = 3 if is_large else 2 - search_calls = 9 if is_large else 6 + budget_cents = _read_positive_cents( + f"{C_CYAN}Client Budget in USD (for example, 50.00):{C_RESET} $" + ) + if budget_cents is None: + continue - custom_job = { - "id": job_id, + is_large = budget_cents >= 5_000 + job = { + "id": f"I{job_index}", "topic": topic, "context": "Interactive customer-submitted request.", "customer_email": "interactive@user.example", "budget_cents": budget_cents, - "est_tokens": est_tokens, - "market_data_calls": market_calls, - "web_search_calls": search_calls + "est_tokens": 12_000 if is_large else 7_500, + "market_data_calls": 3 if is_large else 2, + "web_search_calls": 9 if is_large else 6, } + job_index += 1 - print(f"\n{C_BOLD}■ {job_id}: {topic}{C_RESET}") - show_spinner(0.4, "Analyzing inbound job specifications...") - agent.handle_job(custom_job) + print(f"\n{C_BOLD}■ {job['id']}: {topic}{C_RESET}") + print_step("Analyzing inbound job specifications...") + agent.handle_job(job) - cont = input(f"Submit another research request? ({C_BOLD}y/N{C_RESET}): ").strip().lower() + again = input(f"Submit another research request? ({C_BOLD}y/N{C_RESET}): ").strip().lower() print() - if cont != "y": + if again != "y": break - snap = agent.t.snapshot() - print_results(snap) - path = dashboard.render(snap, agent.log) - print(f"\n Dashboard: {C_BLUE}{path}{C_RESET}\n{BAR}\n") - + _finish_session(agent) -def resolve_config(args) -> SolventConfig: - """Load, onboard, or default user preferences.""" +def resolve_config() -> SolventConfig: + """Load saved preferences, onboarding when necessary.""" if wants_reconfigure(): return run_wizard() - if not config_exists() and not should_skip_onboarding(): return run_wizard() + return load_config() or default_config() - cfg = load_config() - if cfg is None: - cfg = default_config() - return cfg - - -def main(): - # Fast-path: tui subcommand bypasses argparse entirely - if len(sys.argv) > 1 and sys.argv[1] == "tui": - from .tui import run - run() - return +def main() -> None: parser = argparse.ArgumentParser( description="Run SOLVENT — a self-funding analyst agent.", ) parser.add_argument( - "-i", "--interactive", + "-i", + "--interactive", action="store_true", help="submit custom research jobs from the terminal (overrides saved mode)", ) @@ -289,23 +280,23 @@ def main(): parser.add_argument( "--keep-balance", action="store_true", - help="keep existing database balance (do not reset treasury)", + help="keep the existing database balance", ) args = parser.parse_args() - cfg = resolve_config(args) - apply_config(cfg) + config = resolve_config() + apply_config(config) seed_cents = int(args.seed * 100) fresh = not args.keep_balance - if args.interactive: run_interactive_mode(seed_cents=seed_cents, fresh=fresh) - elif cfg.interaction_mode == "programmatic": + elif config.interaction_mode == "programmatic": from solvent.onboarding import _print_programmatic_guidance + _print_programmatic_guidance() - sys.exit(0) - elif cfg.interaction_mode == "interactive": + raise SystemExit(0) + elif config.interaction_mode == "interactive": run_interactive_mode(seed_cents=seed_cents, fresh=fresh) else: run_batch_demo(seed_cents=seed_cents, fresh=fresh) diff --git a/solvent/doctor.py b/solvent/doctor.py index 5ec62a2..7381ac4 100644 --- a/solvent/doctor.py +++ b/solvent/doctor.py @@ -1,4 +1,4 @@ -"""Hermes-style diagnostics for SOLVENT.""" +"""Local diagnostics for SOLVENT's runtime, storage, and optional features.""" from __future__ import annotations @@ -10,13 +10,14 @@ from .treasury import Treasury from .workspace import ensure_workspace, list_workspace_files -# Optional extras: (human label, import name, what it unlocks). + _EXTRAS = [ ("rich", "rich", "terminal dashboard"), - ("fastapi", "fastapi", "webhooks + hosted briefs"), - ("uvicorn", "uvicorn", "server runtime"), - ("stripe", "stripe", "live Stripe test-mode"), + ("fastapi", "fastapi", "webhooks and hosted briefs"), + ("uvicorn", "uvicorn", "HTTP server runtime"), + ("stripe", "stripe", "Stripe test-mode integration"), ("telegram", "telegram", "Telegram channel"), + ("qrcode", "qrcode", "pairing QR images"), ] @@ -26,68 +27,91 @@ def _module_available(name: str) -> bool: def run_checks() -> list[dict]: checks: list[dict] = [] - t = Treasury() + treasury = Treasury() - def add(name: str, ok: bool, detail: str = ""): + def add(name: str, ok: bool, detail: str = "") -> None: checks.append({"name": name, "ok": ok, "detail": detail}) - # --- install / runtime environment --------------------------------- - pv = sys.version_info - add("python_version", pv >= (3, 10), f"{pv.major}.{pv.minor}.{pv.micro} (need >= 3.10)") + version = sys.version_info + add( + "python_version", + version >= (3, 10), + f"{version.major}.{version.minor}.{version.micro} (need >= 3.10)", + ) try: - from importlib.metadata import PackageNotFoundError, version + from importlib.metadata import PackageNotFoundError, version as package_version + try: - add("install_mode", True, f"pip-installed (v{version('solvent-agent')})") + detail = f"pip-installed (v{package_version('solvent-agent')})" except PackageNotFoundError: - add("install_mode", True, "source checkout (not pip-installed)") + detail = "source checkout" except Exception as exc: # pragma: no cover - add("install_mode", True, f"unknown ({exc})") + detail = f"unknown ({exc})" + add("install_mode", True, detail) home = base_dir() - ddir = data_dir() - add("data_home", os.access(ddir, os.W_OK), f"{home} (set SOLVENT_HOME to relocate)") + runtime_dir = data_dir() + add("data_home", os.access(runtime_dir, os.W_OK), str(home)) - available = [f"{label} {'✓' if _module_available(mod) else '✗'}" for label, mod, _ in _EXTRAS] + available = [ + f"{label} {'✓' if _module_available(module) else '✗'}" + for label, module, _ in _EXTRAS + ] add("optional_extras", True, ", ".join(available)) - add("sqlite_writable", t.path.parent.exists() or True, str(t.path)) + add( + "sqlite_writable", + os.access(treasury.path.parent, os.W_OK), + str(treasury.path), + ) try: - with t._conn() as conn: - conn.execute("SELECT 1") + with treasury._conn() as connection: + connection.execute("SELECT 1") add("sqlite_connect", True) except Exception as exc: add("sqlite_connect", False, str(exc)) - nvidia = bool(os.environ.get("NVIDIA_API_KEY")) - add("nvidia_api_key", nvidia, "set" if nvidia else "offline stub mode") - - stripe = os.environ.get("STRIPE_API_KEY", "") - if stripe.startswith("sk_live_"): - add("stripe_key", False, "live keys refused") - elif stripe.startswith("sk_test_") or stripe.startswith("rk_test_"): - add("stripe_key", True, "test key present") + add( + "nvidia_mode", + True, + "live key set" if os.environ.get("NVIDIA_API_KEY") else "offline stub", + ) + + stripe_key = os.environ.get("STRIPE_API_KEY", "") + if stripe_key.startswith("sk_live_"): + add("stripe_mode", False, "live keys are refused") + elif stripe_key.startswith(("sk_test_", "rk_test_")): + add("stripe_mode", True, "test key present") else: - add("stripe_key", True, "simulate mode (no key)") + add("stripe_mode", True, "simulate mode") - tg = os.environ.get("TELEGRAM_BOT_TOKEN", "") - add("telegram_token", bool(tg), "set" if tg else "telegram disabled") + add( + "telegram_mode", + True, + "token set" if os.environ.get("TELEGRAM_BOT_TOKEN") else "disabled", + ) stuck = [ - j for j in t.list_jobs() - if j.get("status") in ("awaiting_payment", "in_progress", "paid_pending_fulfill") + job + for job in treasury.list_jobs() + if job.get("status") + in ("awaiting_payment", "in_progress", "paid_pending_fulfill") ] - add("stuck_jobs", len(stuck) == 0, f"{len(stuck)} stuck" if stuck else "none") + add("stuck_jobs", not stuck, f"{len(stuck)} stuck" if stuck else "none") - bal = t.balance_cents() - add("treasury_balance", bal >= 0, f"${bal/100:.2f}") + balance = treasury.balance_cents() + add("treasury_balance", balance >= 0, f"${balance / 100:.2f}") ensure_workspace() - files = list_workspace_files() - core = {"SOUL.md", "AGENTS.md", "BRAIN.md"} - present = {f["name"] for f in files if f["exists"]} - missing = core - present - add("workspace_files", not missing, "ok" if not missing else f"missing {', '.join(sorted(missing))}") + core_files = {"SOUL.md", "AGENTS.md", "BRAIN.md"} + present = {row["name"] for row in list_workspace_files() if row["exists"]} + missing = core_files - present + add( + "workspace_files", + not missing, + "ok" if not missing else f"missing {', '.join(sorted(missing))}", + ) return checks @@ -98,25 +122,23 @@ def add(name: str, ok: bool, detail: str = ""): solvent finance financial report (income · runway · forecast) solvent --help list all commands -Enable live features: - pip install -e ".[all]" rich TUI · Stripe · server · Telegram +Enable optional features: + pip install -e ".[all]" server · Stripe · Telegram · TUI · QR export NVIDIA_API_KEY=nvapi-... live Nemotron inference export STRIPE_API_KEY=sk_test_... real test-mode payment links """ -def main(): - checks = run_checks() +def main() -> None: failed = 0 - for c in checks: - mark = "OK" if c["ok"] else "FAIL" - detail = f" — {c['detail']}" if c.get("detail") else "" - print(f"[{mark}] {c['name']}{detail}") - if not c["ok"]: - failed += 1 + for check in run_checks(): + mark = "OK" if check["ok"] else "FAIL" + detail = f" — {check['detail']}" if check.get("detail") else "" + print(f"[{mark}] {check['name']}{detail}") + failed += not check["ok"] print(QUICKSTART) if failed: - sys.exit(1) + raise SystemExit(1) if __name__ == "__main__": diff --git a/solvent/job_cmd.py b/solvent/job_cmd.py index 1763f6a..6eb6d9d 100644 --- a/solvent/job_cmd.py +++ b/solvent/job_cmd.py @@ -1,5 +1,5 @@ """ -job_cmd.py — CLI for inspecting and managing SOLVENT jobs. +CLI for inspecting and managing SOLVENT jobs. Usage: solvent jobs list recent jobs (last 20) @@ -53,16 +53,16 @@ def _ago(ts) -> str: if delta < 60: return f"{int(delta)}s" if delta < 3600: - return f"{int(delta/60)}m" + return f"{int(delta / 60)}m" if delta < 86400: - return f"{int(delta/3600)}h" - return f"{int(delta/86400)}d" + return f"{int(delta / 3600)}h" + return f"{int(delta / 86400)}d" def _fmt_cents(cents) -> str: if cents is None: return "—" - return f"${int(cents)/100:,.2f}" + return f"${int(cents) / 100:,.2f}" def _status_label(status: str) -> str: @@ -77,11 +77,13 @@ def _col(s: str, width: int) -> str: return s.ljust(width) -# --------------------------------------------------------------------------- -# Sub-commands -# --------------------------------------------------------------------------- - -def cmd_list(treasury, *, status_filter: Optional[str] = None, limit: int = 20, as_json: bool = False): +def cmd_list( + treasury, + *, + status_filter: Optional[str] = None, + limit: int = 20, + as_json: bool = False, +): jobs = treasury.list_jobs() if status_filter: jobs = [j for j in jobs if j.get("status") == status_filter] @@ -102,15 +104,16 @@ def cmd_list(treasury, *, status_filter: Optional[str] = None, limit: int = 20, print(header) print("─" * len(header)) - for j in reversed(jobs): - jid = j.get("id", "")[:17] - status = j.get("status", "") + for job in reversed(jobs): + job_id = job.get("id", "")[:17] + status = job.get("status", "") emoji = _STATUS_EMOJI.get(status, "? ") - topic = (j.get("topic") or "")[:41] - budget = _fmt_cents(j.get("budget_cents")) - ago = _ago(j.get("updated_at") or j.get("created_at")) + topic = (job.get("topic") or "")[:41] + budget = _fmt_cents(job.get("budget_cents")) + ago = _ago(job.get("updated_at") or job.get("created_at")) print( - f"{_col(jid, 18)} {emoji}{_col(status, 20)} {_col(topic, 42)} {_col(budget, 8)} {ago}" + f"{_col(job_id, 18)} {emoji}{_col(status, 20)} " + f"{_col(topic, 42)} {_col(budget, 8)} {ago}" ) @@ -118,7 +121,7 @@ def cmd_show(treasury, job_id: str, *, as_json: bool = False): job = treasury.get_job(job_id) if not job: print(f"Job not found: {job_id}", file=sys.stderr) - sys.exit(1) + raise SystemExit(1) try: pnl = treasury.job_pnl_cents(job_id) @@ -150,17 +153,17 @@ def cmd_show(treasury, job_id: str, *, as_json: bool = False): if metrics: print("\nMetrics:") - for k, v in metrics.items(): - if k == "job_id": + for key, value in metrics.items(): + if key == "job_id": continue - print(f" {k}: {v}") + print(f" {key}: {value}") def cmd_events(treasury, job_id: str, *, limit: int = 20, as_json: bool = False): job = treasury.get_job(job_id) if not job: print(f"Job not found: {job_id}", file=sys.stderr) - sys.exit(1) + raise SystemExit(1) events = treasury.list_events(job_id=job_id, limit=limit) @@ -174,18 +177,37 @@ def cmd_events(treasury, job_id: str, *, limit: int = 20, as_json: bool = False) print(f"Events for {job_id}:") print(f" {'TIME':17} {'STAGE':20}") - print(f" {'─'*17} {'─'*20}") - for ev in events: - ts = _fmt_ts(ev.get("ts")) - stage = ev.get("stage", "") + print(f" {'─' * 17} {'─' * 20}") + for event in events: + ts = _fmt_ts(event.get("ts")) + stage = event.get("stage", "") print(f" {ts:17} {stage}") +def cmd_retry(treasury, job_id: str, *, runner=None): + """Retry through the same StageRunner used by the worker and agent.""" + if runner is None: + from .guardrails import Guardrails + from .stages import StageRunner + from .stripe_client import StripeClient + + runner = StageRunner(treasury, Guardrails(treasury), StripeClient()) + + try: + result = runner.retry_job(job_id) + except ValueError as exc: + print(str(exc), file=sys.stderr) + raise SystemExit(1) from exc + + print(json.dumps(result, indent=2, default=str)) + return result + + def cmd_cancel(treasury, job_id: str): job = treasury.get_job(job_id) if not job: print(f"Job not found: {job_id}", file=sys.stderr) - sys.exit(1) + raise SystemExit(1) current = job.get("status", "") if current in ("completed", "cancelled"): @@ -197,53 +219,55 @@ def cmd_cancel(treasury, job_id: str): print(f"Job {job_id} cancelled (was: {current})") -# --------------------------------------------------------------------------- -# Entry point -# --------------------------------------------------------------------------- - def main(): import argparse - p = argparse.ArgumentParser( + parser = argparse.ArgumentParser( prog="solvent jobs", description="Inspect and manage SOLVENT jobs.", ) - p.add_argument("--json", action="store_true", dest="as_json", help="output as JSON") + parser.add_argument("--json", action="store_true", dest="as_json", help="output as JSON") - sub = p.add_subparsers(dest="cmd") + sub = parser.add_subparsers(dest="cmd") - list_p = sub.add_parser("list", help="list recent jobs") - list_p.add_argument("--status", choices=_ALL_STATUSES, help="filter by status") - list_p.add_argument("--limit", type=int, default=20) + list_parser = sub.add_parser("list", help="list recent jobs") + list_parser.add_argument("--status", choices=_ALL_STATUSES, help="filter by status") + list_parser.add_argument("--limit", type=int, default=20) - show_p = sub.add_parser("show", help="full detail for one job") - show_p.add_argument("job_id") + show_parser = sub.add_parser("show", help="full detail for one job") + show_parser.add_argument("job_id") - ev_p = sub.add_parser("events", help="event log for one job") - ev_p.add_argument("job_id") - ev_p.add_argument("--limit", type=int, default=20) + events_parser = sub.add_parser("events", help="event log for one job") + events_parser.add_argument("job_id") + events_parser.add_argument("--limit", type=int, default=20) - cancel_p = sub.add_parser("cancel", help="cancel a job") - cancel_p.add_argument("job_id") + retry_parser = sub.add_parser("retry", help="retry a failed or awaiting-payment job") + retry_parser.add_argument("job_id") - args = p.parse_args() - cmd = args.cmd or "list" + cancel_parser = sub.add_parser("cancel", help="cancel a job") + cancel_parser.add_argument("job_id") + + args = parser.parse_args() + command = args.cmd or "list" from .treasury import Treasury + treasury = Treasury() - if cmd == "list": + if command == "list": cmd_list( treasury, status_filter=getattr(args, "status", None), limit=getattr(args, "limit", 20), as_json=args.as_json, ) - elif cmd == "show": + elif command == "show": cmd_show(treasury, args.job_id, as_json=args.as_json) - elif cmd == "events": + elif command == "events": cmd_events(treasury, args.job_id, limit=args.limit, as_json=args.as_json) - elif cmd == "cancel": + elif command == "retry": + cmd_retry(treasury, args.job_id) + elif command == "cancel": cmd_cancel(treasury, args.job_id) diff --git a/solvent/server.py b/solvent/server.py index db823e4..3ab176e 100644 --- a/solvent/server.py +++ b/solvent/server.py @@ -1,4 +1,4 @@ -"""HTTP server: Stripe webhooks, job API, interactive dashboard + chat.""" +"""FastAPI surface for jobs, Stripe webhooks, briefs, dashboard, and chat.""" from __future__ import annotations @@ -7,12 +7,13 @@ import os import threading import time +from pathlib import Path from .agent import Solvent -from .gateway import Gateway, register_outbound -from .delivery import verify_delivery_token, markdown_to_html, is_safe_job_id +from .delivery import is_safe_job_id, markdown_to_html, verify_delivery_token from .event_hub import EventHub -from .paths import data_dir, reports_dir as reports_dir_fn +from .gateway import Gateway, register_outbound +from .paths import data_dir, reports_dir from .stripe_client import StripeClient from .webhook_log import WebhookLog @@ -49,16 +50,53 @@ def _require_fastapi(): try: from fastapi import FastAPI, HTTPException from fastapi.responses import HTMLResponse, JSONResponse, StreamingResponse + return FastAPI, HTTPException, HTMLResponse, JSONResponse, StreamingResponse except ImportError as exc: raise RuntimeError( - "FastAPI is required for `solvent serve`. " - "Install with: pip install -r requirements-serve.txt" + "FastAPI is required for `solvent serve`. Install with: " + "pip install 'solvent-agent[serve]'" ) from exc +def _pairing_host() -> tuple[str, int]: + base_url = os.environ.get("SOLVENT_BASE_URL", "") + host = base_url.replace("https://", "").replace("http://", "").split("/")[0] + try: + port = int(host.split(":")[-1]) if ":" in host else 443 + except ValueError: + port = 443 + return host.split(":")[0], port + + +def _job_payload(body: JobBody) -> dict: + payload = body.model_dump(exclude_none=True) + if not payload.get("id"): + import uuid + + payload["id"] = "J" + uuid.uuid4().hex[:8] + return payload + + +def _brief_path(job_id: str) -> Path | None: + if not is_safe_job_id(job_id): + return None + + root = reports_dir().resolve() + for suffix in (".html", ".md"): + path = (root / f"{job_id}{suffix}").resolve() + try: + path.relative_to(root) + except ValueError: + return None + if path.is_file(): + return path + return None + + def create_app(seed_cents: int = 10_000, fresh: bool = False) -> object: FastAPI, HTTPException, HTMLResponse, JSONResponse, StreamingResponse = _require_fastapi() + hub = EventHub() agent = Solvent(seed_cents=seed_cents, fresh=fresh, sync_payment=False) gateway = Gateway(agent=agent) @@ -67,13 +105,14 @@ def create_app(seed_cents: int = 10_000, fresh: bool = False) -> object: status_lock = threading.Lock() last_status_json = "" - def _refresh_status() -> dict: + def refresh_status() -> dict: from . import dashboard + return dashboard.build_status_data(agent.t.snapshot(), agent.log) - def _publish_status() -> None: + def publish_status() -> None: nonlocal last_status_json - data = _refresh_status() + data = refresh_status() payload = json.dumps(data, sort_keys=True) with status_lock: if payload == last_status_json: @@ -81,29 +120,33 @@ def _publish_status() -> None: last_status_json = payload hub.publish("status", {"data": data}) - def _on_agent_event(event: dict) -> None: - agent._capture_event(event) - data = _refresh_status() - hub.publish("agent_event", {"event": event, "data": data}) + def publish_agent_event(event: dict) -> None: + # StageRunner already routes through agent._capture_event. This callback + # only publishes; calling capture here would recurse and duplicate events. + hub.publish("agent_event", {"event": event, "data": refresh_status()}) - agent._runner.on_event = _on_agent_event - agent.on_event = _on_agent_event + agent.on_event = publish_agent_event - def _dashboard_outbound(external_id: str, text: str) -> None: - hub.publish("chat", {"role": "assistant", "text": text, "session_id": external_id}) + def dashboard_outbound(external_id: str, text: str) -> None: + hub.publish( + "chat", + {"role": "assistant", "text": text, "session_id": external_id}, + ) - register_outbound("dashboard", _dashboard_outbound) + register_outbound("dashboard", dashboard_outbound) app = FastAPI(title="SOLVENT", version="2.1") + app.state.agent = agent app.state.webhook_log = webhook_log @app.on_event("startup") - async def _startup(): + async def startup() -> None: hub.bind_loop(asyncio.get_running_loop()) from . import dashboard + dashboard.render(agent.t.snapshot(), agent.log, live=True) - async def _poll_external_status(): + async def poll_external_status() -> None: status_path = data_dir() / "dashboard_status.json" last_mtime = 0.0 while True: @@ -114,7 +157,9 @@ async def _poll_external_status(): last_mtime = mtime data = json.loads(status_path.read_text(encoding="utf-8")) hub.publish("status", {"data": data}) + from .notifications import drain_chat_outbox + for row in drain_chat_outbox(): hub.publish( "chat", @@ -129,7 +174,7 @@ async def _poll_external_status(): pass await asyncio.sleep(2.0) - asyncio.create_task(_poll_external_status()) + asyncio.create_task(poll_external_status()) @app.get("/health") def health(): @@ -137,60 +182,55 @@ def health(): @app.get("/api/pair/qr") def api_pair_qr(): - """Generate an OpenClaw pairing token and return a QR code PNG (or JSON fallback).""" - import os token = agent.t.create_openclaw_token(ttl=600) - base_url = os.environ.get("SOLVENT_BASE_URL", "") - host = base_url.replace("https://", "").replace("http://", "").split("/")[0] - try: - port = int(host.split(":")[-1]) if ":" in host else 443 - host_name = host.split(":")[0] - except ValueError: - port = 443 - host_name = host - from . import qr as _qr - png = _qr.png_bytes(token, host=host_name, port=port) + host, port = _pairing_host() + from . import qr + + png = qr.png_bytes(token, host=host, port=port) if png: from starlette.responses import Response + return Response(content=png, media_type="image/png") - return JSONResponse({"token": token, "note": "install qrcode[pil] for PNG output"}) + return JSONResponse( + {"token": token, "note": "install solvent-agent[qr] for PNG output"} + ) @app.post("/api/pair/verify") - async def api_pair_verify(req: Request): - body = await req.json() + async def api_pair_verify(request: Request): + body = await request.json() token = (body.get("token") or "").strip() if not token: raise HTTPException(400, "token required") - ok = agent.t.verify_openclaw_token(token) - if not ok: + if not agent.t.verify_openclaw_token(token): raise HTTPException(403, "invalid or expired token") return {"verified": True} @app.get("/") def dashboard_page(): from . import dashboard + path = dashboard.render(agent.t.snapshot(), agent.log, live=True) return HTMLResponse(path.read_text(encoding="utf-8")) @app.get("/api/status") def api_status(): - return JSONResponse(_refresh_status()) + return JSONResponse(refresh_status()) @app.get("/api/events") async def api_events(): - q = hub.subscribe() + queue = hub.subscribe() async def stream(): try: yield EventHub.sse_format({"type": "hello", "ts": time.time()}) while True: try: - msg = await asyncio.wait_for(q.get(), timeout=15.0) - yield EventHub.sse_format(msg) + message = await asyncio.wait_for(queue.get(), timeout=15.0) + yield EventHub.sse_format(message) except asyncio.TimeoutError: yield EventHub.sse_format({"type": "ping", "ts": time.time()}) finally: - hub.unsubscribe(q) + hub.unsubscribe(queue) return StreamingResponse(stream(), media_type="text/event-stream") @@ -200,28 +240,20 @@ async def api_chat(body: ChatBody): if not message: raise HTTPException(400, "message required") session_id = body.session_id or "dashboard-default" - reply = gateway.handle_inbound("dashboard", session_id, message, user_label="dashboard") - _publish_status() + reply = gateway.handle_inbound( + "dashboard", + session_id, + message, + user_label="dashboard", + ) + publish_status() return {"reply": reply, "session_id": session_id} @app.post("/api/job") - async def api_job(body: JobBody): - payload = body.model_dump(exclude_none=True) - if not payload.get("id"): - import uuid - payload["id"] = "J" + uuid.uuid4().hex[:8] - result = agent.enqueue_job(payload) - _publish_status() - return JSONResponse(result) - @app.post("/jobs") - async def create_job(body: JobBody): - payload = body.model_dump(exclude_none=True) - if not payload.get("id"): - import uuid - payload["id"] = "J" + uuid.uuid4().hex[:8] - result = agent.enqueue_job(payload) - _publish_status() + def enqueue(body: JobBody): + result = agent.enqueue_job(_job_payload(body)) + publish_status() return JSONResponse(result) @app.get("/jobs/{job_id}") @@ -229,53 +261,48 @@ def get_job(job_id: str): row = agent.t.get_job(job_id) if not row: raise HTTPException(404, "job not found") - metrics = agent.t.get_metrics(job_id) - return {"job": dict(row), "metrics": metrics} + return {"job": dict(row), "metrics": agent.t.get_metrics(job_id)} @app.get("/api/receipt/{job_id}") def get_receipt(job_id: str, token: str = "", request: Request = None): - """Return a plaintext job receipt. - - Access requires either: - - A valid delivery token (``?token=...``), OR - - The request originating from localhost (127.0.0.1 / ::1). - """ - row = agent.t.get_job(job_id) if not row: raise HTTPException(404, "job not found") - # Auth: localhost OR valid delivery token - is_local = False - if request is not None: - client_host = getattr(request.client, "host", "") if request.client else "" - is_local = client_host in ("127.0.0.1", "::1", "localhost", "testclient") - - if not is_local: - if not verify_delivery_token(job_id, token): - raise HTTPException(403, "invalid or expired delivery token") + client_host = "" + if request is not None and request.client: + client_host = getattr(request.client, "host", "") + is_local = client_host in ("127.0.0.1", "::1", "localhost", "testclient") + if not is_local and not verify_delivery_token(job_id, token): + raise HTTPException(403, "invalid or expired delivery token") from .receipt import build_receipt - job_dict = dict(row) - pnl = agent.t.job_pnl_cents(job_id) - balance = agent.t.balance_cents() - text = build_receipt(job_dict, pnl, balance) - from starlette.responses import PlainTextResponse + + text = build_receipt( + dict(row), + agent.t.job_pnl_cents(job_id), + agent.t.balance_cents(), + ) return PlainTextResponse(text) @app.post("/webhooks/stripe") - async def stripe_webhook(req: Request): - payload = await req.body() - sig = req.headers.get("Stripe-Signature", "") - event_id = json.loads(payload).get("id", "unknown") if payload else "unknown" - event_type = json.loads(payload).get("type", "") if payload else "" + async def stripe_webhook(request: Request): + payload = await request.body() + signature = request.headers.get("Stripe-Signature", "") + try: + envelope = json.loads(payload) if payload else {} + except json.JSONDecodeError: + envelope = {} + event_id = envelope.get("id", "unknown") + event_type = envelope.get("type", "") webhook_log.record(event_id, event_type, payload, "received") + try: - payment = stripe.process_webhook(payload, sig, treasury=agent.t) + payment = stripe.process_webhook(payload, signature, treasury=agent.t) if payment and payment.get("job_id"): agent._runner.handle_webhook_payment(payment) - _publish_status() + publish_status() webhook_log.mark_processed(event_id) except Exception as exc: webhook_log.mark_error(event_id, str(exc)) @@ -288,62 +315,31 @@ def get_brief(job_id: str, token: str = ""): raise HTTPException(404, "brief not found") if not verify_delivery_token(job_id, token): raise HTTPException(403, "invalid or expired delivery token") - reports_dir = reports_dir_fn().resolve() - - path_html = (reports_dir / f"{job_id}.html").resolve() - try: - path_html.relative_to(reports_dir) - if path_html.is_file(): - return HTMLResponse(path_html.read_text(encoding="utf-8")) - except ValueError: - raise HTTPException(404, "brief not found") from None - - path_md = (reports_dir / f"{job_id}.md").resolve() - try: - path_md.relative_to(reports_dir) - if path_md.is_file(): - return HTMLResponse(markdown_to_html(path_md.read_text(encoding="utf-8"))) - except ValueError: - raise HTTPException(404, "brief not found") from None - - raise HTTPException(404, "brief not found") + path = _brief_path(job_id) + if path is None: + raise HTTPException(404, "brief not found") + if path.suffix == ".html": + return HTMLResponse(path.read_text(encoding="utf-8")) + return HTMLResponse(markdown_to_html(path.read_text(encoding="utf-8"))) @app.get("/api/briefs") def list_briefs(): - reports_dir = reports_dir_fn() - if not reports_dir.is_dir(): - return [] - stems = set() - for p in reports_dir.glob("*"): - if p.suffix in (".md", ".html") and p.stem != ".gitkeep": - stems.add(p.stem) - return sorted(list(stems)) + root = reports_dir() + stems = { + path.stem + for path in root.glob("*") + if path.suffix in (".md", ".html") and path.stem != ".gitkeep" + } + return sorted(stems) @app.get("/api/briefs/{job_id}") def get_brief_api(job_id: str): - reports_dir = reports_dir_fn().resolve() - - path_html = (reports_dir / f"{job_id}.html").resolve() - try: - path_html.relative_to(reports_dir) - if path_html.is_file(): - return HTMLResponse(path_html.read_text(encoding="utf-8")) - except ValueError: - raise HTTPException(404, "brief not found") from None - - path_md = (reports_dir / f"{job_id}.md").resolve() - try: - path_md.relative_to(reports_dir) - if path_md.is_file(): - return HTMLResponse(markdown_to_html(path_md.read_text(encoding="utf-8"))) - except ValueError: - raise HTTPException(404, "brief not found") from None - - raise HTTPException(404, "brief not found") - - # ------------------------------------------------------------------ - # Webhook monitoring + replay endpoints - # ------------------------------------------------------------------ + path = _brief_path(job_id) + if path is None: + raise HTTPException(404, "brief not found") + if path.suffix == ".html": + return HTMLResponse(path.read_text(encoding="utf-8")) + return HTMLResponse(markdown_to_html(path.read_text(encoding="utf-8"))) @app.get("/api/webhooks") def api_webhooks_list(): @@ -362,7 +358,7 @@ async def api_webhooks_replay(event_id: str): payment = stripe.process_webhook(stored, sig_header="", treasury=agent.t) if payment and payment.get("job_id"): agent._runner.handle_webhook_payment(payment) - _publish_status() + publish_status() webhook_log.mark_processed(event_id) except Exception as exc: webhook_log.mark_error(event_id, str(exc)) @@ -372,18 +368,24 @@ async def api_webhooks_replay(event_id: str): return app -def main(): +def main() -> None: import argparse + parser = argparse.ArgumentParser(description="SOLVENT HTTP server") parser.add_argument("--port", type=int, default=int(os.environ.get("SOLVENT_PORT", "8787"))) parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--seed", type=float, default=100.0) parser.add_argument("--keep-balance", action="store_true") args = parser.parse_args() + try: import uvicorn except ImportError as exc: - raise RuntimeError("uvicorn required: pip install -r requirements-serve.txt") from exc + raise RuntimeError( + "uvicorn is required for `solvent serve`. Install with: " + "pip install 'solvent-agent[serve]'" + ) from exc + app = create_app(seed_cents=int(args.seed * 100), fresh=not args.keep_balance) uvicorn.run(app, host=args.host, port=args.port) diff --git a/solvent/upgrade.py b/solvent/upgrade.py index b62838a..6345e8a 100644 --- a/solvent/upgrade.py +++ b/solvent/upgrade.py @@ -1,39 +1,24 @@ -""" -upgrade.py — version check and upgrade helper for SOLVENT. - -Usage: - solvent upgrade print upgrade instructions if a newer version exists - solvent upgrade --check exit 1 if current version is behind PyPI (CI-friendly) - solvent upgrade --install run pip install to upgrade in-place - -Background hint: call background_update_hint() at startup to fire a -non-blocking daemon thread that prints a one-line upgrade notice to stderr -if outdated. Rate-limited to once per day; skipped when -SOLVENT_NO_UPDATE_CHECK=1 is set. -""" +"""Explicit version check and upgrade helper for SOLVENT.""" from __future__ import annotations import json -import os import sys -import time import urllib.error import urllib.request from typing import Optional -_CHECK_INTERVAL = 86400 # 24 hours in seconds _PYPI_URL = "https://pypi.org/pypi/solvent-agent/json" -_TIMEOUT = 5 # seconds +_TIMEOUT = 5 -def _parse_version(v: str) -> tuple[int, ...]: - """Parse 'X.Y.Z' into a comparable tuple; non-numeric parts become 0.""" +def _parse_version(version: str) -> tuple[int, ...]: + """Parse X.Y.Z into a comparable tuple; non-numeric parts become zero.""" parts = [] - for seg in v.strip().lstrip("v").split("."): + for segment in version.strip().lstrip("v").split("."): try: - parts.append(int(seg)) + parts.append(int(segment)) except ValueError: parts.append(0) return tuple(parts) @@ -41,29 +26,26 @@ def _parse_version(v: str) -> tuple[int, ...]: def current_version() -> str: from . import __version__ + return __version__ def latest_pypi_version() -> Optional[str]: - """Fetch the latest release version from PyPI. Returns None on any error.""" + """Fetch the latest release version from PyPI, or None on any error.""" try: - req = urllib.request.Request( + request = urllib.request.Request( _PYPI_URL, headers={"User-Agent": "solvent-agent/upgrade-check"}, ) - with urllib.request.urlopen(req, timeout=_TIMEOUT) as resp: - data = json.loads(resp.read().decode()) + with urllib.request.urlopen(request, timeout=_TIMEOUT) as response: + data = json.loads(response.read().decode()) return data["info"]["version"] except (urllib.error.URLError, KeyError, json.JSONDecodeError, OSError): return None def check_upgrade(*, quiet: bool = False) -> dict: - """ - Compare current version to latest on PyPI. - - Returns a dict with keys: current, latest, up_to_date, error. - """ + """Compare the installed version with the latest version on PyPI.""" current = current_version() latest = latest_pypi_version() @@ -73,7 +55,6 @@ def check_upgrade(*, quiet: bool = False) -> dict: return {"current": current, "latest": None, "up_to_date": True, "error": True} up_to_date = _parse_version(current) >= _parse_version(latest) - if not quiet: if up_to_date: print(f"solvent {current} is up to date.") @@ -82,83 +63,55 @@ def check_upgrade(*, quiet: bool = False) -> dict: print(" Upgrade with: pip install --upgrade solvent-agent") print(" Or full extras: pip install --upgrade 'solvent-agent[all]'") - return {"current": current, "latest": latest, "up_to_date": up_to_date, "error": False} + return { + "current": current, + "latest": latest, + "up_to_date": up_to_date, + "error": False, + } def main(): import argparse import subprocess - p = argparse.ArgumentParser( + parser = argparse.ArgumentParser( description="Check for solvent-agent upgrades on PyPI.", formatter_class=argparse.RawDescriptionHelpFormatter, ) - p.add_argument( + parser.add_argument( "--check", action="store_true", help="exit 1 if a newer version is available (CI-friendly)", ) - p.add_argument( + parser.add_argument( "--install", action="store_true", help="run pip install --upgrade solvent-agent if a newer version is available", ) - p.add_argument( + parser.add_argument( "--json", action="store_true", dest="as_json", help="output result as JSON", ) - args = p.parse_args() + args = parser.parse_args() result = check_upgrade(quiet=args.as_json) - if args.as_json: print(json.dumps(result, indent=2)) if not result["up_to_date"] and args.install: print("\nRunning: pip install --upgrade solvent-agent") - rc = subprocess.call([sys.executable, "-m", "pip", "install", "--upgrade", "solvent-agent"]) - sys.exit(rc) + return_code = subprocess.call( + [sys.executable, "-m", "pip", "install", "--upgrade", "solvent-agent"] + ) + raise SystemExit(return_code) if args.check and not result["up_to_date"]: - sys.exit(1) + raise SystemExit(1) - sys.exit(0) - - -def background_update_hint() -> None: - """Fire a daemon thread that prints a one-line upgrade hint to stderr if outdated. - - Rate-limited to once per _CHECK_INTERVAL seconds via a timestamp file. - Skipped when SOLVENT_NO_UPDATE_CHECK=1 is set. - """ - if os.environ.get("SOLVENT_NO_UPDATE_CHECK"): - return - - import threading - - def _check() -> None: - try: - from .paths import data_dir - stamp_path = data_dir() / ".upgrade_check" - if stamp_path.exists(): - last = float(stamp_path.read_text().strip()) - if time.time() - last < _CHECK_INTERVAL: - return - latest = latest_pypi_version() - stamp_path.write_text(str(time.time())) - if latest and not (_parse_version(current_version()) >= _parse_version(latest)): - print( - f"\n[solvent] Update available: {current_version()} → {latest}" - f" (pip install --upgrade solvent-agent)\n", - file=sys.stderr, - ) - except Exception: - pass # never let the background check crash the agent - - t = threading.Thread(target=_check, daemon=True) - t.start() + raise SystemExit(0) if __name__ == "__main__": diff --git a/solvent/webhook_log.py b/solvent/webhook_log.py index 8b7592a..7be6131 100644 --- a/solvent/webhook_log.py +++ b/solvent/webhook_log.py @@ -1,19 +1,4 @@ -""" -webhook_log.py — durable log of received Stripe webhook events. - -Stores every event received with: event_id, type, received_at, status -(processed/error/skipped), error_message. Allows replaying failed events. - -Schema: - webhook_events( - event_id TEXT PRIMARY KEY, - event_type TEXT, - payload BLOB, - received_at REAL, - status TEXT, - error TEXT - ) -""" +"""Durable SQLite log and CLI for received Stripe webhook events.""" from __future__ import annotations @@ -22,6 +7,8 @@ from pathlib import Path from typing import Any +from .paths import data_dir + class WebhookLog: """SQLite-backed durable log of Stripe webhook events.""" @@ -41,22 +28,20 @@ class WebhookLog: "CREATE INDEX IF NOT EXISTS idx_wh_received_at ON webhook_events (received_at)", ] - def __init__(self, db_path: str = ".solvent/webhooks.db") -> None: - if db_path == ":memory:": + def __init__(self, db_path: str | Path | None = None) -> None: + db_path = db_path or (data_dir() / "webhooks.db") + if str(db_path) == ":memory:": self._db_path = ":memory:" else: - Path(db_path).parent.mkdir(parents=True, exist_ok=True) - self._db_path = db_path + path = Path(db_path) + path.parent.mkdir(parents=True, exist_ok=True) + self._db_path = path self._conn = sqlite3.connect(self._db_path, check_same_thread=False) self._conn.row_factory = sqlite3.Row - for stmt in self._DDL: - self._conn.execute(stmt) + for statement in self._DDL: + self._conn.execute(statement) self._conn.commit() - # ------------------------------------------------------------------ - # Write helpers - # ------------------------------------------------------------------ - def record( self, event_id: str, @@ -77,7 +62,7 @@ def record( self._conn.commit() def mark_processed(self, event_id: str) -> None: - """Update status → 'processed'.""" + """Update status to processed and clear any previous error.""" self._conn.execute( "UPDATE webhook_events SET status = 'processed', error = '' WHERE event_id = ?", (event_id,), @@ -85,57 +70,53 @@ def mark_processed(self, event_id: str) -> None: self._conn.commit() def mark_error(self, event_id: str, error: str) -> None: - """Update status → 'error' with an error message.""" + """Update status to error and store the error message.""" self._conn.execute( "UPDATE webhook_events SET status = 'error', error = ? WHERE event_id = ?", (error, event_id), ) self._conn.commit() - # ------------------------------------------------------------------ - # Read helpers - # ------------------------------------------------------------------ - @staticmethod def _row_to_dict(row: sqlite3.Row) -> dict[str, Any]: - d = dict(row) - # Add a human-readable timestamp field import datetime + + data = dict(row) try: - d["received_at_fmt"] = datetime.datetime.fromtimestamp( - d["received_at"] + data["received_at_fmt"] = datetime.datetime.fromtimestamp( + data["received_at"] ).strftime("%Y-%m-%d %H:%M:%S") except Exception: - d["received_at_fmt"] = "" - return d + data["received_at_fmt"] = "" + return data def list_recent(self, limit: int = 50) -> list[dict]: - """Return up to *limit* events sorted by received_at DESC.""" - cur = self._conn.execute( + """Return up to *limit* events sorted by received_at descending.""" + cursor = self._conn.execute( "SELECT * FROM webhook_events ORDER BY received_at DESC LIMIT ?", (limit,), ) - return [self._row_to_dict(r) for r in cur.fetchall()] + return [self._row_to_dict(row) for row in cursor.fetchall()] def list_failed(self) -> list[dict]: - """Return all events where status = 'error'.""" - cur = self._conn.execute( + """Return all events whose status is error.""" + cursor = self._conn.execute( "SELECT * FROM webhook_events WHERE status = 'error' ORDER BY received_at DESC" ) - return [self._row_to_dict(r) for r in cur.fetchall()] + return [self._row_to_dict(row) for row in cursor.fetchall()] def get_payload(self, event_id: str) -> bytes | None: - """Retrieve the raw stored payload for replay, or None if not found.""" - cur = self._conn.execute( + """Retrieve the stored raw payload, or None if it is missing.""" + cursor = self._conn.execute( "SELECT payload FROM webhook_events WHERE event_id = ?", (event_id,), ) - row = cur.fetchone() + row = cursor.fetchone() return bytes(row["payload"]) if row else None def stats(self) -> dict[str, Any]: - """Return aggregate counts: total, processed, error, skipped, last_24h.""" - cur = self._conn.execute( + """Return aggregate event counts.""" + cursor = self._conn.execute( """ SELECT COUNT(*) AS total, @@ -147,7 +128,7 @@ def stats(self) -> dict[str, Any]: """, (time.time() - 86400,), ) - row = cur.fetchone() + row = cursor.fetchone() return { "total": row["total"] or 0, "processed": row["processed"] or 0, @@ -155,3 +136,48 @@ def stats(self) -> dict[str, Any]: "skipped": row["skipped"] or 0, "last_24h": row["last_24h"] or 0, } + + +def main(log: WebhookLog | None = None) -> None: + """Inspect webhook records from the command line.""" + import argparse + import json + + parser = argparse.ArgumentParser( + prog="solvent webhooks", + description="Inspect the durable Stripe webhook log.", + ) + parser.add_argument( + "command", + nargs="?", + default="stats", + choices=("stats", "list", "failed"), + ) + parser.add_argument("--limit", type=int, default=20) + args = parser.parse_args() + + webhook_log = log or WebhookLog() + if args.command == "stats": + print(json.dumps(webhook_log.stats(), indent=2)) + return + + rows = ( + webhook_log.list_recent(args.limit) + if args.command == "list" + else webhook_log.list_failed()[: args.limit] + ) + for row in rows: + if args.command == "list": + print( + f"{row['received_at_fmt']} [{row['status']}] " + f"{row['event_type']} {row['event_id'][:16]}" + ) + else: + print( + f"{row['event_id'][:16]} {row['event_type']} " + f"err={row['error'][:60]}" + ) + + +if __name__ == "__main__": + main() diff --git a/tests/test_cli_routing.py b/tests/test_cli_routing.py index 99ae3b3..b453482 100644 --- a/tests/test_cli_routing.py +++ b/tests/test_cli_routing.py @@ -1,10 +1,8 @@ -"""The __main__ subcommand router must dispatch, not fall through to the demo. - -Regression: a stray `from .cli import main` used to shadow the router so every -documented subcommand (serve/worker/doctor/...) silently ran the demo CLI. -""" +"""The package entry point must dispatch documented commands exactly once.""" +import io import unittest +from contextlib import redirect_stdout from unittest.mock import patch import solvent.__main__ as entry @@ -12,59 +10,102 @@ class TestCliRouting(unittest.TestCase): def _routes_to(self, argv, target_module, target_attr="main"): - with patch.object(entry.sys, "argv", argv), \ - patch(f"solvent.{target_module}.{target_attr}") as mock_target, \ - patch("solvent.cli.main") as mock_demo: + with ( + patch.object(entry.sys, "argv", argv), + patch(f"solvent.{target_module}.{target_attr}") as target, + patch("solvent.cli.main") as demo, + ): entry.main() - return mock_target, mock_demo + return target, demo def test_finance_routes_to_finance(self): target, demo = self._routes_to(["solvent", "finance"], "finance") - target.assert_called_once() + target.assert_called_once_with() demo.assert_not_called() def test_report_alias_routes_to_finance(self): target, demo = self._routes_to(["solvent", "report"], "finance") - target.assert_called_once() + target.assert_called_once_with() demo.assert_not_called() def test_doctor_routes_to_doctor(self): target, demo = self._routes_to(["solvent", "doctor"], "doctor") - target.assert_called_once() + target.assert_called_once_with() + demo.assert_not_called() + + def test_webhooks_routes_to_webhook_cli(self): + target, demo = self._routes_to(["solvent", "webhooks", "stats"], "webhook_log") + target.assert_called_once_with() + demo.assert_not_called() + + def test_tui_routes_to_run(self): + target, demo = self._routes_to(["solvent", "tui"], "tui", "run") + target.assert_called_once_with() + demo.assert_not_called() + + def test_retry_alias_delegates_to_jobs_cli(self): + seen_argv = [] + + def capture_argv(): + seen_argv.append(list(entry.sys.argv)) + + with ( + patch.object(entry.sys, "argv", ["solvent", "retry", "J-123"]), + patch("solvent.job_cmd.main", side_effect=capture_argv) as jobs, + patch("solvent.cli.main") as demo, + ): + entry.main() + + jobs.assert_called_once_with() + self.assertEqual(seen_argv, [["solvent", "retry", "J-123"]]) demo.assert_not_called() def test_no_subcommand_runs_demo(self): - with patch.object(entry.sys, "argv", ["solvent"]), \ - patch("solvent.cli.main") as mock_demo: + with patch.object(entry.sys, "argv", ["solvent"]), patch("solvent.cli.main") as demo: + entry.main() + demo.assert_called_once_with() + + def test_demo_options_run_demo(self): + with ( + patch.object(entry.sys, "argv", ["solvent", "--interactive"]), + patch("solvent.cli.main") as demo, + ): entry.main() - mock_demo.assert_called_once() + demo.assert_called_once_with() def test_version_prints_and_does_not_run_demo(self): - import io - from contextlib import redirect_stdout from solvent import __version__ - for argv in (["solvent", "version"], ["solvent", "--version"], ["solvent", "-V"]): - buf = io.StringIO() - with patch.object(entry.sys, "argv", argv), \ - patch("solvent.cli.main") as mock_demo, redirect_stdout(buf): + for argv in ( + ["solvent", "version"], + ["solvent", "--version"], + ["solvent", "-V"], + ): + buffer = io.StringIO() + with ( + patch.object(entry.sys, "argv", argv), + patch("solvent.cli.main") as demo, + redirect_stdout(buffer), + ): entry.main() - self.assertIn(__version__, buf.getvalue()) - mock_demo.assert_not_called() + self.assertIn(__version__, buffer.getvalue()) + demo.assert_not_called() def test_help_lists_subcommands(self): - import io - from contextlib import redirect_stdout - - buf = io.StringIO() - with patch.object(entry.sys, "argv", ["solvent", "help"]), \ - patch("solvent.cli.main") as mock_demo, redirect_stdout(buf): + buffer = io.StringIO() + with ( + patch.object(entry.sys, "argv", ["solvent", "help"]), + patch("solvent.cli.main") as demo, + redirect_stdout(buffer), + ): entry.main() - out = buf.getvalue() - self.assertIn("Commands:", out) - self.assertIn("finance", out) - self.assertIn("serve", out) - mock_demo.assert_not_called() + + output = buffer.getvalue() + self.assertIn("Commands:", output) + self.assertIn("finance", output) + self.assertIn("serve", output) + self.assertIn("jobs", output) + demo.assert_not_called() if __name__ == "__main__": diff --git a/tests/test_job_cmd.py b/tests/test_job_cmd.py index 3880eaa..09add9c 100644 --- a/tests/test_job_cmd.py +++ b/tests/test_job_cmd.py @@ -1,241 +1,229 @@ -"""tests/test_job_cmd.py — unit tests for the solvent jobs CLI.""" +"""Unit tests for the solvent jobs CLI.""" import io import json import sys import unittest -from contextlib import redirect_stdout +import uuid +from contextlib import redirect_stderr, redirect_stdout from unittest import mock -from solvent.treasury import Treasury from solvent.job_cmd import ( - cmd_list, cmd_show, cmd_events, cmd_cancel, - _fmt_cents, _ago, _fmt_ts, _col, + _ago, + _col, + _fmt_cents, + _fmt_ts, + cmd_cancel, + cmd_events, + cmd_list, + cmd_retry, + cmd_show, ) +from solvent.treasury import Treasury def _fresh_treasury(): - return Treasury(path=f"/tmp/solvent_jobcmd_{id(object())}.db") + return Treasury(path=f"/tmp/solvent_jobcmd_{uuid.uuid4().hex}.db") -def _seed(t: Treasury, n: int = 3) -> list[str]: - """Insert n jobs and return their IDs.""" +def _seed(treasury: Treasury, count: int = 3) -> list[str]: ids = [] statuses = ["awaiting_payment", "in_progress", "completed", "failed"] - for i in range(n): - jid = f"job_{i:03d}" - t.upsert_job(jid, statuses[i % len(statuses)], topic=f"Topic {i}", budget_cents=(i + 1) * 1000) - ids.append(jid) + for index in range(count): + job_id = f"job_{index:03d}" + treasury.upsert_job( + job_id, + statuses[index % len(statuses)], + topic=f"Topic {index}", + budget_cents=(index + 1) * 1000, + ) + ids.append(job_id) return ids class TestHelpers(unittest.TestCase): - - def test_fmt_cents_basic(self): + def test_fmt_cents(self): self.assertEqual(_fmt_cents(500), "$5.00") - - def test_fmt_cents_none(self): self.assertEqual(_fmt_cents(None), "—") - def test_fmt_ts_none(self): + def test_fmt_ts(self): + import time + self.assertEqual(_fmt_ts(None), "—") + self.assertRegex(_fmt_ts(time.time()), r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}") - def test_fmt_ts_valid(self): + def test_ago(self): import time - result = _fmt_ts(time.time()) - self.assertRegex(result, r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}") - def test_ago_seconds(self): - import time self.assertIn("s", _ago(time.time() - 10)) - - def test_ago_none(self): self.assertEqual(_ago(None), "") - def test_col_truncates(self): + def test_col(self): + self.assertEqual(len(_col("hi", 10)), 10) self.assertEqual(len(_col("A" * 30, 10)), 10) self.assertTrue(_col("A" * 30, 10).endswith("…")) - def test_col_pads(self): - self.assertEqual(len(_col("hi", 10)), 10) - class TestCmdList(unittest.TestCase): - def test_empty_treasury_prints_no_jobs(self): - t = _fresh_treasury() - buf = io.StringIO() - with redirect_stdout(buf): - cmd_list(t) - self.assertIn("No jobs", buf.getvalue()) + buffer = io.StringIO() + with redirect_stdout(buffer): + cmd_list(_fresh_treasury()) + self.assertIn("No jobs", buffer.getvalue()) def test_lists_jobs(self): - t = _fresh_treasury() - _seed(t, 3) - buf = io.StringIO() - with redirect_stdout(buf): - cmd_list(t) - out = buf.getvalue() - self.assertIn("job_000", out) - self.assertIn("Topic 0", out) + treasury = _fresh_treasury() + _seed(treasury, 3) + buffer = io.StringIO() + with redirect_stdout(buffer): + cmd_list(treasury) + self.assertIn("job_000", buffer.getvalue()) + self.assertIn("Topic 0", buffer.getvalue()) def test_status_filter(self): - t = _fresh_treasury() - _seed(t, 4) - buf = io.StringIO() - with redirect_stdout(buf): - cmd_list(t, status_filter="completed") - out = buf.getvalue() - # Only completed jobs should appear - self.assertNotIn("awaiting_payment", out) + treasury = _fresh_treasury() + _seed(treasury, 4) + buffer = io.StringIO() + with redirect_stdout(buffer): + cmd_list(treasury, status_filter="completed") + self.assertNotIn("awaiting_payment", buffer.getvalue()) def test_json_output_is_list(self): - t = _fresh_treasury() - _seed(t, 2) - buf = io.StringIO() - with redirect_stdout(buf): - cmd_list(t, as_json=True) - data = json.loads(buf.getvalue()) - self.assertIsInstance(data, list) + treasury = _fresh_treasury() + _seed(treasury, 2) + buffer = io.StringIO() + with redirect_stdout(buffer): + cmd_list(treasury, as_json=True) + self.assertIsInstance(json.loads(buffer.getvalue()), list) def test_limit_respected(self): - t = _fresh_treasury() - _seed(t, 10) - buf = io.StringIO() - with redirect_stdout(buf): - cmd_list(t, limit=3) - # Should not contain all 10 job IDs in output - out = buf.getvalue() - # Count distinct job_ occurrences - count = sum(1 for i in range(10) if f"job_{i:03d}" in out) + treasury = _fresh_treasury() + _seed(treasury, 10) + buffer = io.StringIO() + with redirect_stdout(buffer): + cmd_list(treasury, limit=3) + output = buffer.getvalue() + count = sum(1 for index in range(10) if f"job_{index:03d}" in output) self.assertLessEqual(count, 3) - def test_no_jobs_with_status_filter_message(self): - t = _fresh_treasury() - buf = io.StringIO() - with redirect_stdout(buf): - cmd_list(t, status_filter="failed") - self.assertIn("failed", buf.getvalue()) - class TestCmdShow(unittest.TestCase): - def test_show_missing_job_exits(self): - t = _fresh_treasury() with self.assertRaises(SystemExit): - cmd_show(t, "nonexistent_id") + cmd_show(_fresh_treasury(), "missing") def test_show_existing_job(self): - t = _fresh_treasury() - t.upsert_job("job_show", "completed", topic="My topic", budget_cents=5000) - buf = io.StringIO() - with redirect_stdout(buf): - cmd_show(t, "job_show") - out = buf.getvalue() - self.assertIn("job_show", out) - self.assertIn("My topic", out) - self.assertIn("completed", out) + treasury = _fresh_treasury() + treasury.upsert_job("job_show", "completed", topic="My topic", budget_cents=5000) + buffer = io.StringIO() + with redirect_stdout(buffer): + cmd_show(treasury, "job_show") + output = buffer.getvalue() + self.assertIn("job_show", output) + self.assertIn("My topic", output) + self.assertIn("completed", output) def test_show_json(self): - t = _fresh_treasury() - t.upsert_job("job_json", "in_progress", topic="JSON topic", budget_cents=1000) - buf = io.StringIO() - with redirect_stdout(buf): - cmd_show(t, "job_json", as_json=True) - data = json.loads(buf.getvalue()) + treasury = _fresh_treasury() + treasury.upsert_job("job_json", "in_progress", topic="JSON topic", budget_cents=1000) + buffer = io.StringIO() + with redirect_stdout(buffer): + cmd_show(treasury, "job_json", as_json=True) + data = json.loads(buffer.getvalue()) self.assertEqual(data["id"], "job_json") self.assertIn("pnl_cents", data) class TestCmdEvents(unittest.TestCase): - def test_events_missing_job_exits(self): - t = _fresh_treasury() with self.assertRaises(SystemExit): - cmd_events(t, "nope") - - def test_events_no_events_message(self): - t = _fresh_treasury() - t.upsert_job("ev_job", "awaiting_payment") - buf = io.StringIO() - with redirect_stdout(buf): - cmd_events(t, "ev_job") - self.assertIn("No events", buf.getvalue()) - - def test_events_lists_recorded(self): - t = _fresh_treasury() - t.upsert_job("ev_job2", "awaiting_payment") - t.record_event("ev_job2", "quote", {"amount": 500}) - t.record_event("ev_job2", "paid", {}) - buf = io.StringIO() - with redirect_stdout(buf): - cmd_events(t, "ev_job2") - out = buf.getvalue() - self.assertIn("quote", out) - self.assertIn("paid", out) + cmd_events(_fresh_treasury(), "missing") + + def test_events_lists_recorded_events(self): + treasury = _fresh_treasury() + treasury.upsert_job("events", "awaiting_payment") + treasury.record_event("events", "quote", {"amount": 500}) + treasury.record_event("events", "paid", {}) + buffer = io.StringIO() + with redirect_stdout(buffer): + cmd_events(treasury, "events") + self.assertIn("quote", buffer.getvalue()) + self.assertIn("paid", buffer.getvalue()) def test_events_json(self): - t = _fresh_treasury() - t.upsert_job("ev_j3", "in_progress") - t.record_event("ev_j3", "started", {}) - buf = io.StringIO() - with redirect_stdout(buf): - cmd_events(t, "ev_j3", as_json=True) - data = json.loads(buf.getvalue()) - self.assertIsInstance(data, list) - self.assertGreater(len(data), 0) + treasury = _fresh_treasury() + treasury.upsert_job("events", "in_progress") + treasury.record_event("events", "started", {}) + buffer = io.StringIO() + with redirect_stdout(buffer): + cmd_events(treasury, "events", as_json=True) + self.assertIsInstance(json.loads(buffer.getvalue()), list) -class TestCmdCancel(unittest.TestCase): +class TestCmdRetry(unittest.TestCase): + def test_retry_uses_supplied_runner_and_prints_json(self): + runner = mock.Mock() + runner.retry_job.return_value = {"stage": "booked", "status": "completed"} + buffer = io.StringIO() - def test_cancel_missing_exits(self): - t = _fresh_treasury() - with self.assertRaises(SystemExit): - cmd_cancel(t, "nope") + with redirect_stdout(buffer): + result = cmd_retry(_fresh_treasury(), "job_retry", runner=runner) + + runner.retry_job.assert_called_once_with("job_retry") + self.assertEqual(result["status"], "completed") + self.assertEqual(json.loads(buffer.getvalue())["stage"], "booked") + + def test_retry_user_error_exits_cleanly(self): + runner = mock.Mock() + runner.retry_job.side_effect = ValueError("not retryable") + stderr = io.StringIO() + + with redirect_stderr(stderr), self.assertRaises(SystemExit) as context: + cmd_retry(_fresh_treasury(), "job_retry", runner=runner) + + self.assertEqual(context.exception.code, 1) + self.assertIn("not retryable", stderr.getvalue()) + +class TestCmdCancel(unittest.TestCase): def test_cancel_sets_status(self): - t = _fresh_treasury() - t.upsert_job("cancel_me", "in_progress") - buf = io.StringIO() - with redirect_stdout(buf): - cmd_cancel(t, "cancel_me") - job = t.get_job("cancel_me") - self.assertEqual(job["status"], "cancelled") - self.assertIn("cancelled", buf.getvalue()) - - def test_cancel_already_done_is_noop(self): - t = _fresh_treasury() - t.upsert_job("done_job", "completed") - buf = io.StringIO() - with redirect_stdout(buf): - cmd_cancel(t, "done_job") - job = t.get_job("done_job") - self.assertEqual(job["status"], "completed") - self.assertIn("already", buf.getvalue()) + treasury = _fresh_treasury() + treasury.upsert_job("cancel_me", "in_progress") + with redirect_stdout(io.StringIO()): + cmd_cancel(treasury, "cancel_me") + self.assertEqual(treasury.get_job("cancel_me")["status"], "cancelled") + + def test_cancel_completed_job_is_noop(self): + treasury = _fresh_treasury() + treasury.upsert_job("done", "completed") + buffer = io.StringIO() + with redirect_stdout(buffer): + cmd_cancel(treasury, "done") + self.assertEqual(treasury.get_job("done")["status"], "completed") + self.assertIn("already", buffer.getvalue()) class TestJobsCLI(unittest.TestCase): - - def test_main_list_no_args(self): + def test_main_defaults_to_list(self): from solvent.job_cmd import main - buf = io.StringIO() - with mock.patch("solvent.job_cmd.cmd_list") as mock_list, \ - mock.patch.object(sys, "argv", ["solvent-jobs"]): - # cmd_list is called; it should receive a Treasury and no filter - mock_list.side_effect = lambda t, **kw: print("No jobs", file=sys.stdout) - with redirect_stdout(buf): - main() - mock_list.assert_called_once() - - def test_main_list_subcommand(self): + + with ( + mock.patch("solvent.job_cmd.cmd_list") as command, + mock.patch.object(sys, "argv", ["solvent-jobs"]), + ): + main() + command.assert_called_once() + + def test_main_routes_retry(self): from solvent.job_cmd import main - buf = io.StringIO() - with mock.patch("solvent.job_cmd.cmd_list") as mock_list, \ - mock.patch.object(sys, "argv", ["solvent-jobs", "list"]): - mock_list.side_effect = lambda t, **kw: print("No jobs", file=sys.stdout) - with redirect_stdout(buf): - main() - mock_list.assert_called_once() + + with ( + mock.patch("solvent.job_cmd.cmd_retry") as command, + mock.patch.object(sys, "argv", ["solvent-jobs", "retry", "J-1"]), + ): + main() + command.assert_called_once() + self.assertEqual(command.call_args.args[1], "J-1") if __name__ == "__main__": diff --git a/tests/test_server.py b/tests/test_server.py index 836a244..4fc599b 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -1,20 +1,26 @@ -"""Tests for hosted brief serving.""" +"""Tests for the hosted HTTP surface.""" import os import unittest -from pathlib import Path from unittest import mock from solvent.delivery import make_delivery_token from solvent.server import create_app +from solvent.paths import reports_dir class TestHostedBriefs(unittest.TestCase): def setUp(self): - self._env = mock.patch.dict(os.environ, {"SOLVENT_DELIVERY_SECRET": "x" * 32}, clear=False) + self._env = mock.patch.dict( + os.environ, + { + "SOLVENT_DELIVERY_SECRET": "x" * 32, + "SOLVENT_FORCE_STRIPE_SIMULATE": "1", + }, + clear=False, + ) self._env.start() - self.reports_dir = Path(__file__).resolve().parent.parent / "data" / "reports" - self.reports_dir.mkdir(parents=True, exist_ok=True) + self.reports_dir = reports_dir() self.report_path = self.reports_dir / "victim-job.md" self.report_path.write_text("# Private brief", encoding="utf-8") @@ -22,47 +28,75 @@ def tearDown(self): self.report_path.unlink(missing_ok=True) self._env.stop() - def test_brief_serving_requires_exact_job_id_match(self): + def _client(self, *, fresh: bool = True): try: from fastapi.testclient import TestClient except ImportError: self.skipTest("FastAPI test client is not installed") + return TestClient(create_app(fresh=fresh)) - client = TestClient(create_app()) - token = make_delivery_token("victim") - response = client.get(f"/briefs/victim?token={token}") + def test_brief_serving_requires_exact_job_id_match(self): + client = self._client() + response = client.get(f"/briefs/victim?token={make_delivery_token('victim')}") self.assertEqual(response.status_code, 404) - token = make_delivery_token("victim-job") - response = client.get(f"/briefs/victim-job?token={token}") + response = client.get( + f"/briefs/victim-job?token={make_delivery_token('victim-job')}" + ) self.assertEqual(response.status_code, 200) self.assertIn("Private brief", response.text) def test_interactive_dashboard_routes(self): - try: - from fastapi.testclient import TestClient - except ImportError: - self.skipTest("FastAPI test client is not installed") - - client = TestClient(create_app(fresh=True)) + client = self._client() self.assertEqual(client.get("/health").status_code, 200) - dash = client.get("/") - self.assertEqual(dash.status_code, 200) - self.assertIn("Agent Chat", dash.text) + + dashboard = client.get("/") + self.assertEqual(dashboard.status_code, 200) + self.assertIn("Agent Chat", dashboard.text) + status = client.get("/api/status") self.assertEqual(status.status_code, 200) self.assertIn("balance_cents", status.json()) def test_api_chat_status_command(self): + client = self._client() + response = client.post("/api/chat", json={"message": "/status"}) + self.assertEqual(response.status_code, 200) + self.assertIn("Balance", response.json()["reply"]) + + def test_job_submission_publishes_once_without_event_recursion(self): + client = self._client() + response = client.post( + "/jobs", + json={ + "id": "J-server-event", + "topic": "Server event wiring", + "budget_cents": 5000, + "customer_email": "server@test.example", + }, + ) + self.assertEqual(response.status_code, 200) + self.assertIn("session_id", response.json()) + + stored = client.get("/jobs/J-server-event") + self.assertEqual(stored.status_code, 200) + self.assertEqual(stored.json()["job"]["status"], "awaiting_payment") + + agent = client.app.state.agent + stages = [event.get("stage") for event in agent.log] + self.assertEqual(stages.count("quote"), 1) + self.assertEqual(stages.count("invoice"), 1) + + def test_pairing_endpoint_returns_png_when_qr_extra_is_installed(self): try: - from fastapi.testclient import TestClient + import qrcode # noqa: F401 except ImportError: - self.skipTest("FastAPI test client is not installed") + self.skipTest("QR extra is not installed") - client = TestClient(create_app(fresh=True)) - response = client.post("/api/chat", json={"message": "/status"}) + response = self._client().get("/api/pair/qr") self.assertEqual(response.status_code, 200) - self.assertIn("Balance", response.json()["reply"]) + self.assertEqual(response.headers["content-type"], "image/png") + self.assertTrue(response.content.startswith(b"\x89PNG")) if __name__ == "__main__": diff --git a/tests/test_upgrade_hint.py b/tests/test_upgrade_hint.py deleted file mode 100644 index 1ff8a9a..0000000 --- a/tests/test_upgrade_hint.py +++ /dev/null @@ -1,158 +0,0 @@ -"""Tests for background_update_hint() in solvent/upgrade.py.""" - -from __future__ import annotations - -import sys -import threading -import time -from io import StringIO -from pathlib import Path -from unittest import mock - -import pytest - -from solvent.upgrade import background_update_hint - - -# --------------------------------------------------------------------------- -# Helpers -# --------------------------------------------------------------------------- - -def _join_threads(timeout: float = 2.0) -> None: - """Wait for all non-main daemon threads to finish.""" - for t in threading.enumerate(): - if t is not threading.main_thread(): - t.join(timeout=timeout) - - -# --------------------------------------------------------------------------- -# Env-var guard -# --------------------------------------------------------------------------- - -def test_env_var_suppresses_hint(tmp_path: Path) -> None: - """SOLVENT_NO_UPDATE_CHECK=1 must prevent any thread from starting.""" - with mock.patch.dict("os.environ", {"SOLVENT_NO_UPDATE_CHECK": "1"}): - with mock.patch("solvent.upgrade.latest_pypi_version") as mock_pypi: - background_update_hint() - _join_threads() - mock_pypi.assert_not_called() - - -# --------------------------------------------------------------------------- -# Happy-path: outdated version prints to stderr -# --------------------------------------------------------------------------- - -def test_prints_hint_when_outdated(tmp_path: Path) -> None: - with mock.patch.dict( - "os.environ", - {"SOLVENT_NO_UPDATE_CHECK": "", "SOLVENT_HOME": str(tmp_path)}, - clear=False, - ): - with mock.patch("solvent.upgrade.latest_pypi_version", return_value="999.0.0"): - with mock.patch("solvent.upgrade.current_version", return_value="0.1.0"): - with mock.patch("solvent.paths.data_dir", return_value=tmp_path): - stderr = StringIO() - with mock.patch("sys.stderr", stderr): - background_update_hint() - _join_threads() - output = stderr.getvalue() - assert "999.0.0" in output - assert "0.1.0" in output - assert "pip install" in output - - -# --------------------------------------------------------------------------- -# Up-to-date: no output -# --------------------------------------------------------------------------- - -def test_no_hint_when_up_to_date(tmp_path: Path) -> None: - with mock.patch.dict("os.environ", {"SOLVENT_NO_UPDATE_CHECK": ""}, clear=False): - with mock.patch("solvent.upgrade.latest_pypi_version", return_value="0.1.0"): - with mock.patch("solvent.upgrade.current_version", return_value="0.1.0"): - with mock.patch("solvent.paths.data_dir", return_value=tmp_path): - stderr = StringIO() - with mock.patch("sys.stderr", stderr): - background_update_hint() - _join_threads() - output = stderr.getvalue() - assert output == "" - - -# --------------------------------------------------------------------------- -# PyPI unreachable: no crash, no output -# --------------------------------------------------------------------------- - -def test_no_hint_when_pypi_unreachable(tmp_path: Path) -> None: - with mock.patch.dict("os.environ", {"SOLVENT_NO_UPDATE_CHECK": ""}, clear=False): - with mock.patch("solvent.upgrade.latest_pypi_version", return_value=None): - with mock.patch("solvent.paths.data_dir", return_value=tmp_path): - stderr = StringIO() - with mock.patch("sys.stderr", stderr): - background_update_hint() - _join_threads() - output = stderr.getvalue() - assert output == "" - - -# --------------------------------------------------------------------------- -# Rate-limiting: second call within interval is suppressed -# --------------------------------------------------------------------------- - -def test_rate_limit_suppresses_second_call(tmp_path: Path) -> None: - stamp = tmp_path / ".upgrade_check" - stamp.write_text(str(time.time())) # pretend we checked just now - - with mock.patch.dict("os.environ", {"SOLVENT_NO_UPDATE_CHECK": ""}, clear=False): - with mock.patch("solvent.upgrade.latest_pypi_version") as mock_pypi: - with mock.patch("solvent.paths.data_dir", return_value=tmp_path): - background_update_hint() - _join_threads() - mock_pypi.assert_not_called() - - -def test_rate_limit_allows_call_after_interval(tmp_path: Path) -> None: - stamp = tmp_path / ".upgrade_check" - old_ts = time.time() - 90000 # 25 hours ago - stamp.write_text(str(old_ts)) - - with mock.patch.dict("os.environ", {"SOLVENT_NO_UPDATE_CHECK": ""}, clear=False): - with mock.patch("solvent.upgrade.latest_pypi_version", return_value="999.0.0"): - with mock.patch("solvent.upgrade.current_version", return_value="0.1.0"): - with mock.patch("solvent.paths.data_dir", return_value=tmp_path): - stderr = StringIO() - with mock.patch("sys.stderr", stderr): - background_update_hint() - _join_threads() - output = stderr.getvalue() - assert "999.0.0" in output - - -# --------------------------------------------------------------------------- -# Stamp file is written after a successful check -# --------------------------------------------------------------------------- - -def test_stamp_written_after_check(tmp_path: Path) -> None: - stamp = tmp_path / ".upgrade_check" - assert not stamp.exists() - - with mock.patch.dict("os.environ", {"SOLVENT_NO_UPDATE_CHECK": ""}, clear=False): - with mock.patch("solvent.upgrade.latest_pypi_version", return_value="0.1.0"): - with mock.patch("solvent.upgrade.current_version", return_value="0.1.0"): - with mock.patch("solvent.paths.data_dir", return_value=tmp_path): - background_update_hint() - _join_threads() - - assert stamp.exists() - ts = float(stamp.read_text()) - assert abs(ts - time.time()) < 5 - - -# --------------------------------------------------------------------------- -# Exceptions inside thread must not propagate -# --------------------------------------------------------------------------- - -def test_exception_in_thread_does_not_raise(tmp_path: Path) -> None: - with mock.patch.dict("os.environ", {"SOLVENT_NO_UPDATE_CHECK": ""}, clear=False): - with mock.patch("solvent.paths.data_dir", side_effect=RuntimeError("boom")): - background_update_hint() - _join_threads() # no exception should surface here diff --git a/tests/test_webhook_log.py b/tests/test_webhook_log.py index 1a91f18..dd9ae06 100644 --- a/tests/test_webhook_log.py +++ b/tests/test_webhook_log.py @@ -1,126 +1,139 @@ -"""Tests for solvent.webhook_log — in-memory SQLite.""" +"""Tests for the durable webhook log and its CLI.""" from __future__ import annotations +import io +import json +import sys +import tempfile import time import unittest +from pathlib import Path +from contextlib import redirect_stdout +from unittest import mock -from solvent.webhook_log import WebhookLog +from solvent.webhook_log import WebhookLog, main class TestWebhookLog(unittest.TestCase): def setUp(self): - self.wl = WebhookLog(db_path=":memory:") + self.log = WebhookLog(db_path=":memory:") def test_record_appears_in_list_recent(self): - self.wl.record("evt_001", "payment_intent.created", b'{"id":"evt_001"}', "received") - rows = self.wl.list_recent() + self.log.record("evt_001", "payment_intent.created", b'{"id":"evt_001"}', "received") + rows = self.log.list_recent() self.assertEqual(len(rows), 1) self.assertEqual(rows[0]["event_id"], "evt_001") self.assertEqual(rows[0]["event_type"], "payment_intent.created") self.assertEqual(rows[0]["status"], "received") def test_list_recent_sorted_by_received_at_desc(self): - self.wl.record("evt_A", "charge.succeeded", b"a", "received") + self.log.record("evt_A", "charge.succeeded", b"a", "received") time.sleep(0.01) - self.wl.record("evt_B", "charge.failed", b"b", "received") - rows = self.wl.list_recent() - self.assertEqual(rows[0]["event_id"], "evt_B") - self.assertEqual(rows[1]["event_id"], "evt_A") + self.log.record("evt_B", "charge.failed", b"b", "received") + rows = self.log.list_recent() + self.assertEqual([row["event_id"] for row in rows], ["evt_B", "evt_A"]) def test_list_recent_respects_limit(self): - for i in range(10): - self.wl.record(f"evt_{i:03}", "ping", b"x", "received") - rows = self.wl.list_recent(limit=3) - self.assertEqual(len(rows), 3) + for index in range(10): + self.log.record(f"evt_{index:03}", "ping", b"x", "received") + self.assertEqual(len(self.log.list_recent(limit=3)), 3) def test_received_at_fmt_populated(self): - self.wl.record("evt_fmt", "foo", b"y", "received") - rows = self.wl.list_recent() - self.assertNotEqual(rows[0]["received_at_fmt"], "") - - def test_record_idempotent_via_replace(self): - self.wl.record("evt_dup", "ping", b"first", "received") - self.wl.record("evt_dup", "ping", b"second", "processed") - rows = self.wl.list_recent() - self.assertEqual(len(rows), 1) - self.assertEqual(rows[0]["status"], "processed") + self.log.record("evt_fmt", "foo", b"y", "received") + self.assertTrue(self.log.list_recent()[0]["received_at_fmt"]) - def test_mark_processed_changes_status(self): - self.wl.record("evt_p", "charge.captured", b"p", "received") - self.wl.mark_processed("evt_p") - rows = self.wl.list_recent() + def test_record_is_idempotent(self): + self.log.record("evt_dup", "ping", b"first", "received") + self.log.record("evt_dup", "ping", b"second", "processed") + rows = self.log.list_recent() + self.assertEqual(len(rows), 1) self.assertEqual(rows[0]["status"], "processed") - def test_mark_processed_clears_error(self): - self.wl.record("evt_p2", "charge.captured", b"p2", "error", error="boom") - self.wl.mark_processed("evt_p2") - rows = self.wl.list_recent() - self.assertEqual(rows[0]["error"], "") + def test_mark_processed_changes_status_and_clears_error(self): + self.log.record("evt_p", "charge.captured", b"p", "error", error="boom") + self.log.mark_processed("evt_p") + row = self.log.list_recent()[0] + self.assertEqual(row["status"], "processed") + self.assertEqual(row["error"], "") def test_mark_error_stores_message(self): - self.wl.record("evt_e", "payment_intent.failed", b"e", "received") - self.wl.mark_error("evt_e", "Signature mismatch") - rows = self.wl.list_recent() - self.assertEqual(rows[0]["status"], "error") - self.assertEqual(rows[0]["error"], "Signature mismatch") - - def test_mark_error_overrides_previous_error(self): - self.wl.record("evt_e2", "payment_intent.failed", b"e2", "error", error="first") - self.wl.mark_error("evt_e2", "second error") - rows = self.wl.list_recent() - self.assertEqual(rows[0]["error"], "second error") - - def test_list_failed_only_returns_error_rows(self): - self.wl.record("evt_ok", "ping", b"ok", "processed") - self.wl.record("evt_bad", "ping", b"bad", "error", error="oops") - self.wl.record("evt_skip", "ping", b"skip", "skipped") - rows = self.wl.list_failed() - self.assertEqual(len(rows), 1) - self.assertEqual(rows[0]["event_id"], "evt_bad") - - def test_list_failed_empty_when_no_errors(self): - self.wl.record("evt_x", "ping", b"x", "processed") - self.assertEqual(self.wl.list_failed(), []) + self.log.record("evt_e", "payment_intent.failed", b"e", "received") + self.log.mark_error("evt_e", "Signature mismatch") + row = self.log.list_recent()[0] + self.assertEqual(row["status"], "error") + self.assertEqual(row["error"], "Signature mismatch") + + def test_list_failed_only_returns_errors(self): + self.log.record("evt_ok", "ping", b"ok", "processed") + self.log.record("evt_bad", "ping", b"bad", "error", error="oops") + self.log.record("evt_skip", "ping", b"skip", "skipped") + rows = self.log.list_failed() + self.assertEqual([row["event_id"] for row in rows], ["evt_bad"]) def test_stats_returns_correct_counts(self): - self.wl.record("s1", "t", b"", "processed") - self.wl.record("s2", "t", b"", "processed") - self.wl.record("s3", "t", b"", "error") - self.wl.record("s4", "t", b"", "skipped") - self.wl.record("s5", "t", b"", "received") - s = self.wl.stats() - self.assertEqual(s["total"], 5) - self.assertEqual(s["processed"], 2) - self.assertEqual(s["error"], 1) - self.assertEqual(s["skipped"], 1) + self.log.record("s1", "t", b"", "processed") + self.log.record("s2", "t", b"", "processed") + self.log.record("s3", "t", b"", "error") + self.log.record("s4", "t", b"", "skipped") + self.log.record("s5", "t", b"", "received") + stats = self.log.stats() + self.assertEqual(stats["total"], 5) + self.assertEqual(stats["processed"], 2) + self.assertEqual(stats["error"], 1) + self.assertEqual(stats["skipped"], 1) + self.assertEqual(stats["last_24h"], 5) def test_stats_on_empty_db(self): - s = self.wl.stats() - self.assertEqual(s["total"], 0) - self.assertEqual(s["processed"], 0) - self.assertEqual(s["error"], 0) - self.assertEqual(s["skipped"], 0) - self.assertEqual(s["last_24h"], 0) - - def test_stats_last_24h_counts_recent(self): - self.wl.record("r1", "t", b"", "received") - self.wl.record("r2", "t", b"", "processed") - s = self.wl.stats() - self.assertEqual(s["last_24h"], 2) - - def test_get_payload_returns_stored_bytes(self): - raw = b'{"id": "evt_raw", "type": "charge.succeeded"}' - self.wl.record("evt_raw", "charge.succeeded", raw, "received") - self.assertEqual(self.wl.get_payload("evt_raw"), raw) - - def test_get_payload_returns_none_when_missing(self): - self.assertIsNone(self.wl.get_payload("evt_nonexistent")) - - def test_get_payload_returns_bytes_type(self): - self.wl.record("evt_bytes", "t", b"hello", "received") - result = self.wl.get_payload("evt_bytes") - self.assertIsInstance(result, bytes) + self.assertEqual( + self.log.stats(), + {"total": 0, "processed": 0, "error": 0, "skipped": 0, "last_24h": 0}, + ) + + def test_get_payload(self): + raw = b'{"id": "evt_raw"}' + self.log.record("evt_raw", "charge.succeeded", raw, "received") + self.assertEqual(self.log.get_payload("evt_raw"), raw) + self.assertIsNone(self.log.get_payload("missing")) + + def test_default_database_honors_solvent_home(self): + with tempfile.TemporaryDirectory() as tmp: + with mock.patch.dict("os.environ", {"SOLVENT_HOME": tmp}, clear=False): + log = WebhookLog() + self.assertEqual(Path(log._db_path), Path(tmp).resolve() / "data" / "webhooks.db") + + +class TestWebhookLogCLI(unittest.TestCase): + def setUp(self): + self.log = WebhookLog(db_path=":memory:") + + def _run(self, *args): + buffer = io.StringIO() + with mock.patch.object(sys, "argv", ["solvent-webhooks", *args]), redirect_stdout(buffer): + main(self.log) + return buffer.getvalue() + + def test_default_command_prints_stats_json(self): + data = json.loads(self._run()) + self.assertEqual(data["total"], 0) + + def test_list_prints_recent_events(self): + self.log.record("evt_list", "checkout.session.completed", b"x", "processed") + output = self._run("list") + self.assertIn("evt_list", output) + self.assertIn("processed", output) + + def test_failed_prints_error_events(self): + self.log.record("evt_failed", "checkout.session.completed", b"x", "error", "boom") + output = self._run("failed") + self.assertIn("evt_failed", output) + self.assertIn("boom", output) + + def test_limit_applies(self): + for index in range(3): + self.log.record(f"evt_{index}", "ping", b"x", "processed") + output = self._run("list", "--limit", "1") + self.assertEqual(output.count("evt_"), 1) if __name__ == "__main__":