main.py

"""CodeRifts enforcement with AutoGen — abort on BLOCK."""
import requests

CODERIFTS_URL = "https://app.coderifts.com/api/v1/public/preflight"


def coderifts_preflight(spec_url: str) -> dict:
    """Call CodeRifts public preflight."""
    resp = requests.get(f"{CODERIFTS_URL}?url={spec_url}")
    return resp.json()


def coderifts_termination_check(message: dict) -> bool:
    """AutoGen termination function: return True to stop the conversation."""
    spec_url = message.get("spec_url", "")
    if not spec_url:
        return False

    decision = coderifts_preflight(spec_url)
    print(f"[CODERIFTS] decision={decision['decision']} risk={decision.get('risk_score')}")

    if decision["decision"] == "BLOCK":
        print("[AUTOGEN] Terminating conversation — CodeRifts BLOCK")
        return True

    return False


def api_tool(spec_url: str) -> str:
    """Simulated API tool that an AutoGen agent would call."""
    # In production, this would be registered as an AutoGen tool:
    #   @user_proxy.register_for_execution()
    #   @assistant.register_for_llm(description="Call external API")
    #   def api_tool(spec_url: str) -> str:

    decision = coderifts_preflight(spec_url)

    if decision["decision"] == "BLOCK":
        raise RuntimeError(
            f"CodeRifts BLOCK: risk_score={decision.get('risk_score')}, "
            f"patterns={decision.get('patterns')}"
        )

    if decision["decision"] == "PENDING":
        import time
        time.sleep(5)
        return api_tool(spec_url)

    print(f"[AUTOGEN] API call safe — proceeding")
    return "API call succeeded"


# ── AutoGen conversation simulation ──────────────────────────────────────
# In production, use:
#   from autogen import AssistantAgent, UserProxyAgent
#   assistant = AssistantAgent("assistant", llm_config=...)
#   user_proxy = UserProxyAgent("user", is_termination_msg=coderifts_termination_check)

def run_autogen_demo(spec_url: str):
    """Simulate an AutoGen conversation with CodeRifts enforcement."""
    print(f"[AUTOGEN] Starting conversation about {spec_url}")

    # Step 1: Agent wants to call an API
    print("[AUTOGEN] Assistant: I'll check the API safety first.")

    # Step 2: CodeRifts guard
    try:
        result = api_tool(spec_url)
        print(f"[AUTOGEN] Tool result: {result}")
    except RuntimeError as e:
        print(f"[AUTOGEN] Tool blocked: {e}")
        print("[AUTOGEN] Conversation terminated by CodeRifts guard")


if __name__ == "__main__":
    run_autogen_demo("https://petstore3.swagger.io/api/v3/openapi.json")