feat: Phase 2+4 — agent_loop engine + project agent config

pipeline_service/agent_loop.py

@@ -0,0 +1,74 @@
+"""Agent loop — fetches tasks, executes with LLM, delivers results."""
+import asyncio
+import json
+import logging
+
+logger = logging.getLogger("pipeline.agent_loop")
+
+
+async def run_agent_loop(project_id, role_name, model_name=None):
+    """Single iteration: fetch pending task, execute, deliver."""
+    from sqlor.dbpools import DBPools
+
+    db = DBPools()
+    async with db.sqlorContext("pipeline") as sor:
+        recs = await sor.sqlExe(
+            "SELECT id, title, params, pipeline_id FROM pipeline_tasks "
+            "WHERE state='submitted' AND tenant_id=${pid}$ LIMIT 1",
+            {"pid": project_id},
+        )
+        if not recs:
+            return {"status": "idle", "message": "No pending tasks"}
+
+        task = recs[0]
+        task_id = task.id
+        await sor.sqlExe(
+            "UPDATE pipeline_tasks SET state='running' WHERE id=${tid}$",
+            {"tid": task_id},
+        )
+
+        from pipeline_service.llm_bridge import llm_call
+
+        title = getattr(task, "title", "")
+        params_str = getattr(task, "params", "{}")
+        prompt = (
+            f"Execute this development task:\n"
+            f"Title: {title}\n"
+            f"Params: {params_str}\n\n"
+            f"Provide the result."
+        )
+        try:
+            result = await llm_call(prompt, model=model_name)
+        except Exception as e:
+            await sor.sqlExe(
+                "UPDATE pipeline_tasks SET state='failed' WHERE id=${tid}$",
+                {"tid": task_id},
+            )
+            return {"status": "failed", "task_id": task_id, "error": str(e)[:200]}
+
+        from appPublic.uniqueID import getID
+
+        did = getID()
+        await sor.C("pipeline_deliverables", {
+            "id": did, "project_id": project_id, "task_id": task_id,
+            "deliverable_type": "code", "title": title, "content": result,
+            "quality_score": 80, "review_status": "pending",
+            "created_by": role_name,
+        })
+        await sor.sqlExe(
+            "UPDATE pipeline_tasks SET state='completed' WHERE id=${tid}$",
+            {"tid": task_id},
+        )
+        return {"status": "completed", "task_id": task_id, "deliverable_id": did}
+
+
+async def agent_loop(project_id, role_name, model_name=None, max_iterations=10):
+    """Run agent loop until idle or max reached."""
+    results = []
+    for _ in range(max_iterations):
+        r = await run_agent_loop(project_id, role_name, model_name)
+        results.append(r)
+        if r["status"] == "idle":
+            break
+        await asyncio.sleep(2)
+    return results

pipeline_service/init.py

@@ -370,8 +370,11 @@ def load_pipeline_service():
     # Register intent classifier + LLM bridge (shared across all pipelines)
     from .intent_classifier import intent_classify
     from .llm_bridge import llm_call
+    from .agent_loop import agent_loop, run_agent_loop
     env.intent_classify = intent_classify
     env.pipeline_llm_call = llm_call
+    env.agent_loop = agent_loop
+    env.run_agent_loop = run_agent_loop
 
     debug(f"[{MODULE_NAME}] v{MODULE_VERSION} loaded — pipeline engine with human-in-the-loop support")
     return True