Document intelligence pipelines on a filesystem.

from trove_sdk import TroveAdminClient, TroveClient

admin = TroveAdminClient(api_key="trove-admin-...", workspace_id="ws-abc123")

# One namespace-locked key per client. Even if it leaks, it can only see
# one tenant's documents.
key = admin.create_key(name="finscope-acme-corp", namespace="acme-corp")
fs  = TroveClient(api_key=key.api_key, namespace="acme-corp")

fs.exec("mkdir -p workspace/inbox/{invoices,receipts,contracts,expenses}")
fs.exec("mkdir -p workspace/extracted workspace/reports workspace/flags")

import base64, json, anthropic

claude = anthropic.Anthropic()

PROMPT = """Extract structured invoice data. Output ONLY JSON:
{"invoice_no": ..., "vendor": ..., "issued": "YYYY-MM-DD",
 "line_items": [{"description":..., "qty":..., "amount_usd":...}],
 "total_usd": ...}"""

def extract_invoice(image_bytes: bytes) -> dict:
    b64 = base64.standard_b64encode(image_bytes).decode()
    resp = claude.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1500,
        system=PROMPT,
        messages=[{"role": "user", "content": [
            {"type": "image", "source": {
                "type": "base64", "media_type": "image/png", "data": b64}},
            {"type": "text", "text": "Extract this invoice."},
        ]}],
    )
    return json.loads("".join(b.text for b in resp.content if b.type == "text"))

# Pull the upload from TroveFiles, run vision, write structured JSON back —
# all inside the customer's namespace.
raw  = fs.read_bytes("workspace/inbox/invoices/INV-2026-0001.png")
data = extract_invoice(raw)
fs.write("workspace/extracted/invoices/INV-2026-0001.json", json.dumps(data, indent=2))

# Aggregate every extracted invoice/receipt/expense into monthly rollups
# and compliance flags — running INSIDE the namespace via TroveFiles's sandbox.
# Documents never leave the tenant; the analytics live next to the data.

ROLLUP = """
import json, glob, os
from collections import defaultdict

monthly, dups, over = defaultdict(lambda: {'total_usd': 0.0}), defaultdict(list), []
for path in sorted(glob.glob('extracted/invoices/*.json')):
    d = json.load(open(path))
    monthly[d['issued'][:7]]['total_usd'] += d['total_usd']
    dups[(d['vendor'], round(d['total_usd'], 2))].append(path)
    if d['total_usd'] > 5000:
        over.append({'invoice_no': d['invoice_no'], 'total_usd': d['total_usd']})

os.makedirs('reports/monthly', exist_ok=True)
os.makedirs('flags', exist_ok=True)
for m, agg in monthly.items():
    json.dump(agg, open(f'reports/monthly/{m}.json', 'w'))
json.dump([{'vendor': v, 'total': t, 'count': len(p)}
           for (v, t), p in dups.items() if len(p) > 1],
          open('flags/duplicate_invoices.json', 'w'))
json.dump(over, open('flags/over_threshold.json', 'w'))
"""

fs.write("workspace/scripts/rollup.py", ROLLUP)
result = fs.exec_detailed("python3 scripts/rollup.py")
assert result.exit_code == 0, result.stderr

from fastapi import FastAPI, Request, HTTPException
from trove_sdk import verify_webhook, WebhookSignatureError
import os

app = FastAPI()

@app.post("/trove/events")
async def receive(request: Request):
    body = await request.body()
    try:
        event = verify_webhook(
            secret=os.environ["TROVE_WEBHOOK_SECRET"],
            body=body,
            signature_header=request.headers["x-trove-signature"],
        )
    except WebhookSignatureError:
        raise HTTPException(401)

    if event.type != "file.written":
        return {"ok": True}

    path = event.data["path"]
    # Route by document type. Each branch reads the upload from the
    # customer's namespace, extracts, writes structured JSON back, and
    # re-runs the aggregator. The whole pipeline is event-driven —
    # no cron, no polling, no shared queue.
    if path.startswith("workspace/inbox/invoices/"):
        await process_invoice(event.namespace, path)
    elif path.startswith("workspace/inbox/receipts/"):
        await process_receipt(event.namespace, path)
    elif path.startswith("workspace/inbox/expenses/"):
        await process_expense(event.namespace, path)
    elif path.startswith("workspace/inbox/contracts/"):
        await process_contract(event.namespace, path)
    return {"ok": True}

# After every Nth document, snapshot the namespace for compliance
# retention. Snapshots are tar-on-S3 with 30-day default retention
# and can be restored to a clean namespace at any time.
if docs_seen % 6 == 0:
    snap = fs.create_snapshot(label=f"finscope-{run_id}-{docs_seen}docs")
    audit_log.append({"event": "snapshot",
                      "snapshot_id": snap.snapshot_id,
                      "size_bytes":  snap.size_bytes,
                      "label":       snap.label})