Pebble Engine Example

A minimal reference for a Pebble-compatible audit engine, the HTTP service the Pebble platform submits audits to. server.py implements the platform contract; agent.py is a simple Codex scanner. Replace agent.py with your own scanner, or implement the contract from scratch in any language, and the platform can route audits to your engine with no platform-side changes.

Contract

Method	Path	Purpose
POST	/v1/tasks	Submit an audit → {id, status}
GET	/v1/tasks/{id}	Poll status → {status, finding_count, used_amount_usd, …}
GET	/v1/tasks/{id}/findings	Page findings → {findings, has_more, next_cursor}
POST	/v1/tasks/{id}/cancel	Stop a run
GET	/health	Liveness

Submit body (the platform also sends profile and requester; this reference ignores them):

{ "target": { "kind": "snapshot", "repo_url": "https://github.com/o/r", "ref": "HEAD" },
  "subpath": "contracts",
  "note": "the admin deployer key is trusted",
  "budget_usd": 420.00 }

budget_usd is the remaining budget for this run. Reject at submit with 402 if you expect the run to cost more than that, so an expensive run can't blow the budget.

Alongside snapshot, the contract defines optional diff (base/head) and remediation (ref) target kinds a vendor may implement; this reference implements snapshot.

status ends at completed | cancelled | failed; on completed, page /findings. Each finding: {id, task_id, created_at, title, description (markdown), severity (info|low|medium|high|critical), labels, comments}.

Every /v1 route takes Authorization: Bearer <token> when PEBBLE_ENGINE_API_KEY is set. openapi.yaml is the machine-readable OpenAPI (Swagger) spec of these routes and schemas: generate a server stub from it or validate your responses against it in any stack. tests/test_engine.py encodes the same contract as acceptance tests.

Run

Needs git and the Codex CLI (npm install -g @openai/codex) on PATH:

uv sync && codex login
uv run python -m pebble_engine        # listens on :8900
uv run pytest

curl -s localhost:8900/v1/tasks -d '{"target":{"repo_url":"https://github.com/owner/repo"}}'
curl -s localhost:8900/v1/tasks/<id>
curl -s localhost:8900/v1/tasks/<id>/findings

Or containerized (Codex baked in; pass auth at runtime):

docker build -t pebble-engine . && docker run -p 8900:8900 -e OPENAI_API_KEY=... pebble-engine

Env: PEBBLE_ENGINE_HOST, PEBBLE_ENGINE_PORT, PEBBLE_ENGINE_API_KEY, PEBBLE_ENGINE_MODEL, PEBBLE_ENGINE_EFFORT (defaults to xhigh).

Write your own engine

Replace agent.py. The engine calls run_audit(repo_url, ref, note, model=…, effort=…, subpath=…) and expects a list of {title, description, severity} dicts back; raise to fail the task. server.py owns the wire shapes and needs no changes for a scanner with the same inputs. If you meter spend, fill in Task.cost_usd. Tasks are held in memory; persist and expire them in production. You can also implement the contract from scratch in any language: this repo is a reference rather than a dependency.

Wire into the platform

Your engine stays on your infrastructure. The platform never sees or runs your code; everything goes through the five routes above. When you're ready, send the platform operator your HTTPS base URL and a Bearer API key. That's the whole onboarding: they point the platform at your endpoint and audits start arriving, with no code changes on either side.

The platform relies on these behaviors:

• Fetch the repo yourself. The request carries only repo_url, ref, and an optional subpath, so private-repo access is your responsibility.
• If you reject a submission with a 400 or 402, the platform shows your {"error": …} message directly to the requesting user, so write it for humans. 402 means the account balance is exhausted or the run would exceed budget_usd.
• Make finding_count match what /findings can page out; the platform treats a shortfall as a failed audit. Give findings unique ids and advance next_cursor, since the platform dedupes by id and fails a stalled page.
• Keep tasks queryable through and past terminal state: the platform fetches findings and the final spend after it sees the terminal status. If you return a 404 for a task it's still tracking, it gives up, cancels the run, and reports it failed.
• Report cumulative spend in used_amount_usd as a decimal string and keep it current while the task runs; the platform bills it at any terminal state, including after a cancel.
• Make cancel stop the run and the spend.
• Keep GET /v1/tasks/{id} cheap; audits can run for hours and the platform polls every few seconds.

Validate with the curl sequence above (it works against any implementation) or adapt tests/test_engine.py.