run_p5_live.py

"""P5: live DeepSeek agent driving the ASPL capability lifecycle.

Self-contained: runs the ASPL app in-process (TestClient, no socket needed) and
uses the real DeepSeek API as the agent brain. Reads DEEPSEEK_API_KEY from the
user's established location (/workspace/complex_app_rebuild/.env).

This is the one command the project settings allow-rule authorizes.
"""

import os
import json
import tempfile
import urllib.request

# ASPL singletons must bind to throwaway paths before importing the app.
_tmp = tempfile.mkdtemp(prefix="aspl_p5_")
os.environ.setdefault("ASPL_DB", os.path.join(_tmp, "p5.db"))
os.environ.setdefault("ASPL_SHOP_KEY", os.path.join(_tmp, "p5_key.json"))
os.environ.setdefault("ASPL_AUDIT_LOG", os.path.join(_tmp, "p5_audit.jsonl"))
os.environ.setdefault("ASPL_POW_DIFFICULTY", "10")

import sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from fastapi.testclient import TestClient
from aspl.server import app
from aspl.adapters.agent import run_agent

DEEPSEEK_URL = "https://api.deepseek.com/chat/completions"
MODEL = "deepseek-chat"


def load_key():
    # The key comes from the environment only — never hardcode or read a secret
    # from a checked-in path. Run with: DEEPSEEK_API_KEY=... python run_p5_live.py
    key = os.environ.get("DEEPSEEK_API_KEY")
    if not key:
        raise SystemExit("Set DEEPSEEK_API_KEY in the environment to run the live demo.")
    return key


def deepseek_brain(api_key):
    def brain(messages):
        # DeepSeek accepts system/user/assistant; map our text-based "tool"
        # responses to user turns.
        clean = []
        for m in messages:
            role = m["role"] if m["role"] in ("system", "user", "assistant") else "user"
            clean.append({"role": role, "content": m["content"]})
        body = json.dumps({"model": MODEL, "messages": clean,
                           "temperature": 0.2, "max_tokens": 400}).encode()
        req = urllib.request.Request(DEEPSEEK_URL, data=body, headers={
            "Content-Type": "application/json",
            "Authorization": f"Bearer {api_key}"})
        with urllib.request.urlopen(req, timeout=60) as r:
            data = json.load(r)
        return data["choices"][0]["message"]["content"] or ""
    return brain


class BoundClient:
    def __init__(self, tc):
        from sdk.aspl_client import ASPLClient
        self._c = ASPLClient(base_url="", api_key=None)
        self._c.session = tc
    def __getattr__(self, name):
        return getattr(self._c, name)


def main():
    key = load_key()
    print(f"DeepSeek key loaded ({key[:7]}...), model={MODEL}\n")
    with TestClient(app) as tc:
        client = BoundClient(tc)
        client.register("deepseek-agent")
        client.publish(type="knowledge",
                       intent="translate english text into french",
                       description="english to french translation guidance",
                       content={"text": "translate carefully, preserve tone"},
                       intent_tags=["translation", "french"])
        print("ASPL node ready, capability seeded. Running live agent loop...\n")
        result = run_agent(deepseek_brain(key), "Translate a document into French", client)
        for t in result["transcript"]:
            if t["type"] == "tool_call":
                print(f"  DeepSeek called {t['call']['name']}({json.dumps(t['call']['arguments'])}) "
                      f"=> ok={t['result'].get('ok')} "
                      f"{'verified' if t['result'].get('verified') else ''}")
            else:
                print(f"  DeepSeek FINAL: {t['text']}")
        print("\nP5 result:", "OK" if result.get("final") else "incomplete")


if __name__ == "__main__":
    main()