[Haik]: ckpt, abstracted utils into their own folder

backend/apps/app_builder/app_builder.py

@@ -16,9 +16,9 @@ from backend.apps.app_builder.App import (
     WorkspaceSeedRequest,
 )
 from backend.config.paths import DB_ROOT
-from backend.apps.app_builder.executor import execute_backend_code
 from backend.apps.app_builder.templates.templates import APP_BUILDER_SKILL, APP_BUILDER_TEMPLATE_FILES
-from backend.apps.app_builder.helpers import walk_directory
+from backend.apps.app_builder.utils.execute_backend_code import execute_backend_code, BackendExecResult
+from backend.apps.app_builder.utils.walk_directory import walk_directory
 
 APP_BUILDER_DIR = os.path.join(DB_ROOT, "app_builder")
 APP_BUILDER_WORKSPACE_DIR = os.path.join(APP_BUILDER_DIR, "workspace")
@@ -199,7 +199,7 @@ async def execute_app(body: AppExecute):
     error = None
     if backend_code:
         try:
-            exec_result = await execute_backend_code(backend_code)
+            exec_result: BackendExecResult = await execute_backend_code(backend_code)
             backend_result = exec_result.result
             stdout_text = exec_result.stdout
             stderr_text = exec_result.stderr

backend/apps/app_builder/utils/execute_backend_code.py

@@ -1,21 +1,18 @@
 import asyncio
 import json
-import logging
 import sys
-from dataclasses import dataclass
-
-logger = logging.getLogger(__name__)
+from typing import Dict, Any
+from pydantic import BaseModel
+from typeguard import typechecked
 
 TIMEOUT_SECONDS = 30
 
-
-@dataclass
-class BackendExecResult:
-    result: dict
+class BackendExecResult(BaseModel):
+    result: Dict[str, Any]
     stdout: str
     stderr: str
 
-
+@typechecked
 async def execute_backend_code(code: str) -> BackendExecResult:
     """Execute user-provided Python code in a subprocess.
 
@@ -24,26 +21,28 @@ async def execute_backend_code(code: str) -> BackendExecResult:
     an in-process StringIO redirect.
     """
 
-    preamble = (
+    preamble: str = (
         "import json, sys, io\n"
         "_orig_stdout = sys.stdout\n"
         "_capture = io.StringIO()\n"
         "sys.stdout = _capture\n"
         "result = {}\n"
     )
-    postamble = (
+    postamble: str = (
         "\nsys.stdout = _orig_stdout\n"
         'json.dump({"__stdout__": _capture.getvalue(), "__result__": result}, sys.stdout)\n'
     )
-    wrapper = preamble + code + postamble
+    wrapper: str = preamble + code + postamble
 
-    proc = await asyncio.create_subprocess_exec(
+    proc: asyncio.subprocess.Process = await asyncio.create_subprocess_exec(
         sys.executable, "-c", wrapper,
         stdout=asyncio.subprocess.PIPE,
         stderr=asyncio.subprocess.PIPE,
     )
 
     try:
+        stdout: bytes
+        stderr: bytes
         stdout, stderr = await asyncio.wait_for(
             proc.communicate(),
             timeout=TIMEOUT_SECONDS,
@@ -53,20 +52,20 @@ async def execute_backend_code(code: str) -> BackendExecResult:
         await proc.wait()
         raise RuntimeError(f"Backend code execution timed out after {TIMEOUT_SECONDS}s")
 
-    stderr_text = stderr.decode(errors="replace").strip()
+    stderr_text: str = stderr.decode(errors="replace").strip()
 
     if proc.returncode != 0:
         raise RuntimeError(f"Backend code error (exit {proc.returncode}): {stderr_text}")
 
     try:
-        parsed = json.loads(stdout.decode())
+        parsed: Dict[str, Any] = json.loads(stdout.decode())
         return BackendExecResult(
             result=parsed.get("__result__", {}),
             stdout=parsed.get("__stdout__", ""),
             stderr=stderr_text,
         )
     except json.JSONDecodeError:
-        raw = stdout.decode(errors="replace").strip()
+        raw: str = stdout.decode(errors="replace").strip()
         raise RuntimeError(
             f"Backend code did not produce valid JSON. Raw output: {raw[:500]}"
         )

backend/apps/app_builder/utils/walk_directory.py

@@ -1,12 +1,10 @@
-"""Pure helpers for the app builder."""
-
-from __future__ import annotations
-
 import os
+from typing import Dict
+from typeguard import typechecked
 
-
-def walk_directory(folder: str) -> dict[str, str]:
-    files: dict[str, str] = {}
+@typechecked
+def walk_directory(folder: str) -> Dict[str, str]:
+    files: Dict[str, str] = {}
     if not os.path.isdir(folder):
         return files
     for root, _dirs, filenames in os.walk(folder):