debug: (inference) yeah, given up fully part 2

inference.py

@@ -1,116 +1,89 @@
 import asyncio
 import os
-import sys
 import json
-import subprocess
-import time
-import threading
 from pathlib import Path
-
-import httpx
 from dotenv import load_dotenv
 from openai import OpenAI
 
-ROOT = Path(__file__).parent.resolve()
-if str(ROOT) not in sys.path:
-    sys.path.insert(0, str(ROOT))
+from cloudenv.client import ClientEnvironment
+from cloudenv.models import CloudEnvAction
 
-TASK_DIR = ROOT / "tasks"
+load_dotenv()
 
-from cloudenv.client import ClientEnvironment # noqa
-from cloudenv.models import CloudEnvAction, CloudEnvObservation # noqa
+ROOT = Path(__file__).parent.resolve()
+TASK_DIR = ROOT / "tasks"
 
-load_dotenv()
+# ----------------------------
+# SYSTEM PROMPT (FILE)
+# ----------------------------
+SYSTEM_PROMPT = (ROOT / "system_prompt.txt").read_text().strip()
 
 HF_TOKEN = os.getenv("HF_TOKEN") or os.getenv("API_KEY")
-API_BASE_URL = os.getenv("API_BASE_URL", "https://api.openai.com/v1")
 MODEL_NAME = os.getenv("MODEL_NAME", "gpt-4o")
-
-MAX_STEPS = int(os.getenv("MAX_STEPS", "20"))
-TEMPERATURE = float(os.getenv("TEMPERATURE", "0.3"))
-MAX_TOKENS = int(os.getenv("MAX_TOKENS", "300"))
-
-SERVER_URL = "http://127.0.0.1:8000"
-
-TASK_ORDER = ["easy-task", "medium-task", "hard-task"]
+API_BASE_URL = os.getenv("API_BASE_URL", "https://api.openai.com/v1")
 
 client = OpenAI(base_url=API_BASE_URL, api_key=HF_TOKEN)
 
-SYSTEM_PROMPT = '{"service":"","operation":"","instance_id":null,"payload":{}}'
-
-
-def start_ministack():
-    proc = subprocess.Popen(
-        ["uv", "run", "ministack"],
-        cwd=ROOT,
-        stdout=subprocess.DEVNULL,
-        stderr=subprocess.DEVNULL,
-    )
-    time.sleep(3)
-    return proc
-
-
-def start_server():
-    proc = subprocess.Popen(
-        ["uv", "run", "server"],
-        cwd=ROOT,
-        stdout=subprocess.PIPE,
-        stderr=subprocess.STDOUT,
-        text=True,
-        bufsize=1,
-    )
-
-    def stream():
-        for line in proc.stdout:
-            print(f"[SERVER] {line}", end="")
-
-    threading.Thread(target=stream, daemon=True).start()
-
-    for _ in range(40):
-        r = httpx.get(f"{SERVER_URL}/docs", timeout=1)
-        if r.status_code == 200:
-            print("[INFO] server ready")
-            return proc
-        time.sleep(1)
-
-    raise RuntimeError("server failed")
+TASK_ORDER = ["easy-task", "medium-task", "hard-task"]
 
 
+# ----------------------------
+# LOAD TASK
+# ----------------------------
 def load_task(name: str):
     return json.loads((TASK_DIR / f"{name}.json").read_text())
 
 
-def register_pipeline(task):
-    url = f"{SERVER_URL}/api/pipelines"
-    with httpx.Client(timeout=20) as http:
-        r = http.post(url, json=task)
+# ----------------------------
+# SAFE ACTION PARSER (FIXED)
+# ----------------------------
+def parse_action(raw: str, task_id: str, episode_id: str) -> CloudEnvAction:
+    # ALWAYS guarantee validity
+    if not raw or not isinstance(raw, str):
+        return CloudEnvAction(
+            task_id=task_id,
+            episode_id=episode_id,
+            service="internal",
+            operation="NoOp",
+            payload={},
+        )
 
-        if r.status_code == 400:
-            detail = r.json().get("detail", "")
-            if "already exists" in detail:
-                http.delete(f"{url}/{task['pipeline_id']}")
-                http.post(url, json=task)
+    try:
+        raw = raw.strip()
 
-        r.raise_for_status()
+        start = raw.find("{")
+        end = raw.rfind("}")
 
+        if start == -1 or end == -1 or end <= start:
+            raise ValueError("No JSON found")
 
-def parse_action(raw: str, task_id: str, episode_id: str) -> CloudEnvAction:
-    raw = raw.strip()
-    raw = raw[raw.find("{") : raw.rfind("}") + 1]
-    d = json.loads(raw)
+        raw_json = raw[start : end + 1]
+        d = json.loads(raw_json)
 
-    return CloudEnvAction(
-        task_id=task_id,
-        episode_id=episode_id,
-        service=d.get("service", "internal"),
-        instance_id=d.get("instance_id"),
-        operation=d.get("operation", "GetCallerIdentity"),
-        payload=d.get("payload", {}),
-        principal_arn=d.get("principal_arn"),
-        intent=d.get("intent"),
-    )
+        return CloudEnvAction(
+            task_id=task_id,
+            episode_id=episode_id,
+            service=d.get("service", "internal"),
+            instance_id=d.get("instance_id"),
+            operation=d.get("operation", "NoOp"),
+            payload=d.get("payload", {}),
+            principal_arn=d.get("principal_arn"),
+            intent=d.get("intent"),
+        )
+
+    except Exception:
+        return CloudEnvAction(
+            task_id=task_id,
+            episode_id=episode_id,
+            service="internal",
+            operation="NoOp",
+            payload={},
+        )
 
 
+# ----------------------------
+# LLM CALL
+# ----------------------------
 def call_model(step, state, broken, ideal, history):
     prompt = f"""
 STEP {step}
@@ -126,32 +99,30 @@ HISTORY {history[-5:]}
             {"role": "system", "content": SYSTEM_PROMPT},
             {"role": "user", "content": prompt},
         ],
-        temperature=TEMPERATURE,
-        max_tokens=MAX_TOKENS,
+        temperature=0.0,
     )
 
     return (res.choices[0].message.content or "").strip()
 
 
-def run_episode(task):
+# ----------------------------
+# EPISODE RUN
+# ----------------------------
+async def run_episode(task, env: ClientEnvironment):
     pipeline_id = task["pipeline_id"]
     task_id = pipeline_id
     episode_id = f"{pipeline_id}-episode-1"
 
-    env = ClientEnvironment(base_url=SERVER_URL)
-
-    register_pipeline(task)
-
-    result = asyncio.run(
-        env.initialize_environment(task_id=task_id, episode_id=episode_id)
+    result = await env.reset(
+        task_id=task_id,
+        episode_id=episode_id,
     )
 
     obs = result.observation
-
     history = []
     rewards = []
 
-    for step in range(MAX_STEPS):
+    for step in range(20):
         if obs.done:
             break
 
@@ -165,27 +136,28 @@ def run_episode(task):
 
         action = parse_action(raw, task_id, episode_id)
 
-        result = asyncio.run(env.step(action))
+        result = await env.step(action)
         obs = result.observation
 
-        rewards.append(float(obs.reward or 0))
+        rewards.append(float(obs.reward or 0.0))
         history.append(raw)
 
     score = sum(rewards) / (len(rewards) or 1)
     print(f"[END] task={task_id} score={score:.3f}")
 
 
-def main():
-    miniproc = start_ministack()
-    serverproc = start_server()
+# ----------------------------
+# MAIN
+# ----------------------------
+async def main():
+    env = await ClientEnvironment.from_env(
+        "Vishvam10/cloudenv", use_docker=False
+    )
 
     for name in TASK_ORDER:
         task = load_task(name)
-        run_episode(task)
-
-    miniproc.terminate()
-    serverproc.terminate()
+        await run_episode(task, env)
 
 
 if __name__ == "__main__":
-    main()
+    asyncio.run(main())

local_log_server.py

@@ -1,9 +0,0 @@
-from fastapi import FastAPI, Request
-
-app = FastAPI()
-
-@app.post("/log")
-async def receive_log(req: Request):
-    data = await req.json()
-    print(data["line"], flush=True)
-    return {"ok": True}