flowcore_engine.py

#!/usr/bin/env python3
"""
FlowCoreAPI Engine — IAM-enforced single call across the full Achilles stack.
Instead of calling MemGuard → NoLeak → EP → RiskOracle individually,
agents call FlowCore once and get back a unified response with one proof_hash.
All dependency calls use public HTTPS URLs — zero EC2 internal dependency.
ALWAYS fails open on any dependency timeout — never blocks trading loop.
"""
import os
import time
import json
import uuid
import hashlib
import logging
import requests
import psycopg2
from datetime import datetime, timezone

logging.basicConfig(
    filename=os.getenv('LOG_FILE', '/tmp/flowcore.log'),
    level=logging.INFO,
    format='%(asctime)s %(message)s'
)

DB = os.getenv('DATABASE_URL', 'dbname=achilles_db user=achilles password=olympus2026 host=localhost')
TIMEOUT_S = 3

# All dependencies via public HTTPS — Render-to-Render calls
ENDPOINTS = {
    'ep':        os.getenv('EP_GUARD_URL',    'https://achillesalpha.onrender.com/ep/validate'),
    'noleak':    os.getenv('NOLEAK_URL',      'https://achillesalpha.onrender.com/noleak/check'),
    'memguard':  os.getenv('MEMGUARD_URL',    'https://achillesalpha.onrender.com/memguard/check'),
    'riskoracle':os.getenv('RISKORACLE_URL',  'https://achillesalpha.onrender.com/risk/check'),
}

DEFAULT_ROUTE = ['memguard', 'noleak', 'ep', 'riskoracle']

def get_internal_agent_ids():
    return [v for k, v in os.environ.items() if k.endswith('_AGENT_ID') and v]

def is_internal_agent(agent_id):
    return agent_id in get_internal_agent_ids()

def get_db():
    return psycopg2.connect(DB)

def generate_proof_hash(payload):
    try:
        return '0x' + hashlib.sha256(
            json.dumps(payload, sort_keys=True).encode()
        ).hexdigest()
    except Exception:
        return '0x' + hashlib.sha256(str(time.time()).encode()).hexdigest()

def call_dependency(name, payload, fallback):
    """Call one dependency with hard timeout. Always returns (result, latency_ms, error)."""
    start = time.time()
    try:
        r = requests.post(ENDPOINTS[name], json=payload, timeout=TIMEOUT_S)
        result = r.json()
        latency = int((time.time() - start) * 1000)
        logging.info(f"[{name}] ok latency={latency}ms")
        return result, latency, None
    except Exception as e:
        latency = int((time.time() - start) * 1000)
        logging.warning(f"[{name}] failed latency={latency}ms error={e} — using fallback")
        return {**fallback, 'fallback': True}, latency, str(e)

def call_memguard(agent_id, context):
    payload = {
        'state': context.get('memoryState', {}),
        'referenceState': {},
        'agentId': agent_id
    }
    fallback = {'stateValid': True, 'driftScore': 0.03, 'decayDetected': False}
    return call_dependency('memguard', payload, fallback)

def call_noleak(agent_id, action):
    payload = {
        'signal': action,
        'capital': action.get('capital', 0),
        'agent_id': agent_id
    }
    fallback = {'shouldExecute': True, 'score': 0.75}
    return call_dependency('noleak', payload, fallback)

def call_ep(agent_id, action):
    payload = {'agent_id': agent_id, 'plan': action}
    fallback = {'valid': True, 'proof_hash': None}
    return call_dependency('ep', payload, fallback)

def call_riskoracle(agent_id, action, context):
    payload = {
        'agentId': agent_id,
        'action': action,
        'context': context
    }
    fallback = {'riskScore': 0.15, 'recommendedAction': 'proceed', 'confidence': 0.75}
    return call_dependency('riskoracle', payload, fallback)

def check_blocked(name, result):
    """Return True if this dependency result should block execution."""
    if result.get('fallback'):
        return False  # Always fail open on fallback
    checks = {
        'memguard':   not result.get('stateValid', True),
        'noleak':     not result.get('shouldExecute', True),
        'ep':         not result.get('valid', True),
        'riskoracle': result.get('recommendedAction') == 'abort',
    }
    return checks.get(name, False)

def log_call(agent_id, intent_type, approved, blocked_by,
             route, unified_proof_hash, latency_ms, payment_protocol=None):
    try:
        conn = get_db()
        cur = conn.cursor()
        cur.execute("""
            INSERT INTO flowcore_calls
            (caller_agent_id, intent_type, approved, blocked_by,
             route, unified_proof_hash, latency_ms, payment_protocol)
            VALUES (%s,%s,%s,%s,%s,%s,%s,%s)
        """, (agent_id, intent_type, approved, blocked_by,
              json.dumps(route), unified_proof_hash, latency_ms, payment_protocol))
        conn.commit()
        conn.close()
    except Exception as e:
        logging.warning(f"log_call failed: {e}")

def run_flow(agent_id, intent, context=None, route=None, payment_protocol=None):
    """
    Core FlowCore check — runs full pre-execution stack in one call.
    Route order: memguard -> noleak -> ep -> riskoracle (default)
    ALWAYS fails open on dependency timeout.
    ALWAYS returns valid response — never raises.
    """
    start = time.time()
    ctx = context or {}
    action = intent.get('action', intent)
    intent_type = intent.get('type', 'unknown')
    flow_id = str(uuid.uuid4())
    active_route = route or DEFAULT_ROUTE
    results = {}
    proof_hashes = []
    blocked_by = None

    try:
        for step in active_route:
            if step == 'memguard':
                r, _, _ = call_memguard(agent_id, ctx)
                results['memguard'] = {
                    'stateValid': r.get('stateValid', True),
                    'driftScore': r.get('driftScore', 0),
                    'decayDetected': r.get('decayDetected', False),
                    'fallback': r.get('fallback', False)
                }
                if r.get('proofHash'): proof_hashes.append(r['proofHash'])

            elif step == 'noleak':
                r, _, _ = call_noleak(agent_id, action)
                results['noleak'] = {
                    'shouldExecute': r.get('shouldExecute', True),
                    'score': r.get('score', 0.75),
                    'fallback': r.get('fallback', False)
                }
                if r.get('ger', {}).get('hash'): proof_hashes.append(r['ger']['hash'])

            elif step == 'ep':
                r, _, _ = call_ep(agent_id, action)
                results['ep'] = {
                    'valid': r.get('valid', True),
                    'proof_hash': r.get('proof_hash'),
                    'fallback': r.get('fallback', False)
                }
                if r.get('proof_hash'): proof_hashes.append(r['proof_hash'])

            elif step == 'riskoracle':
                r, _, _ = call_riskoracle(agent_id, action, ctx)
                results['riskoracle'] = {
                    'riskScore': r.get('riskScore', 0.15),
                    'recommendedAction': r.get('recommendedAction', 'proceed'),
                    'confidence': r.get('confidence', 0.75),
                    'breakdown': r.get('breakdown', {}),
                    'fallback': r.get('fallback', False)
                }
                if r.get('proofHash'): proof_hashes.append(r['proofHash'])

            # Check if this step blocks — only on non-fallback results
            if check_blocked(step, results.get(step, {})):
                blocked_by = step
                break

        approved = blocked_by is None
        latency_ms = int((time.time() - start) * 1000)

        # Generate unified proof hash covering all results
        unified_proof = generate_proof_hash({
            'flowId': flow_id,
            'agentId': agent_id,
            'route': active_route,
            'results': results,
            'approved': approved,
            'ts': time.time()
        })

        response = {
            'flowId': flow_id,
            'approved': approved,
            'blockedBy': blocked_by,
            'results': results,
            'route': active_route,
            'unifiedProofHash': unified_proof,
            'componentProofHashes': proof_hashes,
            'paymentProtocol': payment_protocol,
            'latencyMs': latency_ms,
            'timestamp': datetime.now(timezone.utc).isoformat(),
            'schemaVersion': 'v1'
        }

        # Fire-and-forget log
        log_call(agent_id, intent_type, approved, blocked_by,
                 active_route, unified_proof, latency_ms, payment_protocol)

        logging.info(
            f"[{agent_id}] intent={intent_type} approved={approved} "
            f"blocked_by={blocked_by} latency={latency_ms}ms route={active_route}"
        )
        return response

    except Exception as e:
        latency_ms = int((time.time() - start) * 1000)
        logging.error(f"run_flow failed: {e}")
        unified_proof = generate_proof_hash({'error': str(e), 'ts': time.time()})
        log_call(agent_id, intent_type, True, None,
                 active_route, unified_proof, latency_ms)
        return {
            'flowId': flow_id,
            'approved': True,
            'blockedBy': None,
            'results': results,
            'route': active_route,
            'unifiedProofHash': unified_proof,
            'componentProofHashes': proof_hashes,
            'latencyMs': latency_ms,
            'timestamp': datetime.now(timezone.utc).isoformat(),
            'schemaVersion': 'v1',
            'fallback': True
        }

if __name__ == '__main__':
    print("FlowCoreAPI smoke test...")
    result = run_flow(
        agent_id='achilles',
        intent={'type': 'trade', 'action': {'type': 'swap', 'capital': 50}},
        context={'walletBalance': 500, 'memoryState': {'price': 2000}}
    )
    print(json.dumps(result, indent=2))