Add 10-point flow test harness and make manual run/curate endpoints non-blocking

app.py

@@ -3,6 +3,7 @@ from fastapi.responses import JSONResponse
 from fastapi.templating import Jinja2Templates
 from sqlalchemy import func
 from datetime import datetime, timedelta
+import threading
 from db import init_db, SessionLocal, BotDecision, TradeExecution, CuratedInsight
 from bot import start_scheduler, run_cycle, curate_cycle
 from services import account_snapshot
@@ -21,13 +22,13 @@ def health():
 
 @app.post("/run-now")
 def run_now():
-    run_cycle()
+    threading.Thread(target=run_cycle, daemon=True).start()
-    return {"ok": True, "ran": True}
+    return {"ok": True, "queued": True, "task": "run_cycle"}
 
 @app.post("/curate-now")
 def curate_now():
-    curate_cycle()
+    threading.Thread(target=curate_cycle, daemon=True).start()
-    return {"ok": True, "curated": True}
+    return {"ok": True, "queued": True, "task": "curate_cycle"}
 
 @app.get("/api/account")
 def api_account():

scripts/flow_tests.py

@@ -0,0 +1,140 @@
+#!/usr/bin/env python3
+import json
+import sqlite3
+import time
+from pathlib import Path
+
+import requests
+
+ROOT = Path('/home/drjones/.openclaw/workspace/alpaca-llm-bot-v1')
+DB = ROOT / 'bot.db'
+BASE = 'http://127.0.0.1:8089'
+
+results = []
+
+def record(name, ok, detail=''):
+    results.append({'test': name, 'ok': bool(ok), 'detail': str(detail)[:400]})
+
+def test_health():
+    r = requests.get(f'{BASE}/health', timeout=5)
+    record('health_endpoint', r.ok and r.json().get('ok') is True, r.text[:120])
+
+def test_metrics_schema():
+    r = requests.get(f'{BASE}/api/metrics?hours=24', timeout=8)
+    ok = r.ok
+    detail = ''
+    if ok:
+        j = r.json()
+        ok = j.get('ok') is True and 'decisionSeries' in j and 'activitySeries' in j and 'leaderboard' in j
+        detail = f"keys={list(j.keys())[:6]}"
+    record('metrics_endpoint_schema', ok, detail)
+
+def _counts(cur):
+    out = {}
+    for t in ['insights','decisions','trades']:
+        cur.execute(f'select count(*) from {t}')
+        out[t] = cur.fetchone()[0]
+    return out
+
+def test_curate_creates_insights():
+    con = sqlite3.connect(DB)
+    cur = con.cursor()
+    before = _counts(cur)['insights']
+    try:
+        requests.post(f'{BASE}/curate-now', timeout=25)
+    except Exception:
+        pass
+    time.sleep(1)
+    after = _counts(cur)['insights']
+    record('curate_increases_insights_or_stays', after >= before, f'before={before} after={after}')
+    con.close()
+
+def test_run_now_creates_decision_activity():
+    con = sqlite3.connect(DB)
+    cur = con.cursor()
+    before = _counts(cur)['decisions']
+    try:
+        requests.post(f'{BASE}/run-now', timeout=35)
+    except Exception:
+        pass
+    time.sleep(1)
+    after = _counts(cur)['decisions']
+    record('run_now_increases_decisions_or_stays', after >= before, f'before={before} after={after}')
+    con.close()
+
+def test_decision_integrity():
+    con = sqlite3.connect(DB)
+    cur = con.cursor()
+    cur.execute("select count(*) from decisions where symbol is null or symbol='' or action is null or status is null")
+    bad = cur.fetchone()[0]
+    record('decision_integrity_nonnull_fields', bad == 0, f'bad_rows={bad}')
+    con.close()
+
+def test_notional_cap():
+    from config import settings
+    con = sqlite3.connect(DB)
+    cur = con.cursor()
+    cur.execute('select max(notional) from trades')
+    mx = cur.fetchone()[0] or 0
+    record('trade_notional_within_cap', mx <= settings.max_order_usd + 1e-9, f'max_notional={mx} cap={settings.max_order_usd}')
+    con.close()
+
+def test_status_values():
+    allowed = {'planned','executed','failed','risk_blocked','skipped'}
+    con = sqlite3.connect(DB)
+    cur = con.cursor()
+    cur.execute('select distinct status from decisions order by status')
+    vals = {r[0] for r in cur.fetchall()}
+    ok = vals.issubset(allowed)
+    record('decision_status_enum', ok, f'statuses={sorted(vals)}')
+    con.close()
+
+def test_ollama_tags():
+    from config import settings
+    try:
+        r = requests.get(settings.ollama_url + '/api/tags', timeout=8)
+        ok = r.ok and isinstance(r.json().get('models', []), list)
+        record('ollama_tags_reachable', ok, f"models={len(r.json().get('models',[])) if r.ok else 'n/a'}")
+    except Exception as e:
+        record('ollama_tags_reachable', False, str(e))
+
+def test_qdrant_reachable():
+    from config import settings
+    try:
+        r = requests.get(settings.qdrant_url + '/collections', timeout=8)
+        ok = r.ok
+        record('qdrant_reachable', ok, r.text[:120])
+    except Exception as e:
+        record('qdrant_reachable', False, str(e))
+
+def test_model_config_split():
+    from config import settings
+    ok = bool(settings.ollama_curator_model) and bool(settings.ollama_decision_model)
+    record('model_split_configured', ok, f"curator={settings.ollama_curator_model} decision={settings.ollama_decision_model}")
+
+
+def main():
+    # import local config path
+    import sys
+    sys.path.insert(0, str(ROOT))
+
+    test_health()
+    test_metrics_schema()
+    test_curate_creates_insights()
+    test_run_now_creates_decision_activity()
+    test_decision_integrity()
+    test_notional_cap()
+    test_status_values()
+    test_ollama_tags()
+    test_qdrant_reachable()
+    test_model_config_split()
+
+    passed = sum(1 for r in results if r['ok'])
+    total = len(results)
+    summary = {'passed': passed, 'total': total, 'results': results, 'ts': int(time.time())}
+    out = ROOT / 'logs_flow_tests.json'
+    out.write_text(json.dumps(summary, indent=2))
+    print(json.dumps(summary, indent=2))
+
+if __name__ == '__main__':
+    main()