kin is now less dumb

Author: Udayscode

.github/workflows/ci.yml

@@ -27,7 +27,7 @@ jobs:
       - name: Lint and Format Check
         run: |
           uv run black --check .
-          uv run flake8 . --exclude .venv,migrations --ignore E501,F401
+          uv run flake8 . --exclude .venv,migrations --ignore E501,F401,W503,W504
 
       - name: Verify CLI Entrypoint
         run: PYTHONPATH=src uv run python -m kin.cli --help

setup.cfg

@@ -0,0 +1,4 @@
+[flake8]
+exclude = .venv,migrations,__pycache__,.git
+max-line-length = 120
+ignore = E501,F401,W503,W504

src/kin/agent_runner.py

@@ -31,15 +31,33 @@ async def run_agent():
                 if task_data.agent_type == agent.agent_type:
                     print(f"Processing task: {task_data.msg_id}")
 
-                    result_data = await agent.process(task_data)
-
-                    res = TaskResult(
-                        workflow_id=task_data.workflow_id,
-                        node_id=task_data.node_id,
-                        status="COMPLETED",
-                        output=result_data.get("data", result_data),
+                    # publish RUNNING so CLI shows it immediately
+                    await r.xadd(
+                        f"results:{task_data.workflow_id}",
+                        {
+                            "data": TaskResult(
+                                node_id=task_data.node_id,
+                                status="RUNNING",
+                                output={"agent_type": agent.agent_type},
+                            ).model_dump_json()
+                        },
                     )
 
+                    try:
+                        result_data = await agent.process(task_data)
+                        res = TaskResult(
+                            node_id=task_data.node_id,
+                            status="COMPLETED",
+                            output=result_data,
+                        )
+                    except Exception as e:
+                        res = TaskResult(
+                            node_id=task_data.node_id,
+                            status="FAILED",
+                            error=str(e),
+                            output={"agent_type": agent.agent_type},
+                        )
+
                     await r.xadd(
                         f"results:{task_data.workflow_id}",
                         {"data": res.model_dump_json()},

src/kin/agents/researcher/main.py

@@ -33,13 +33,9 @@ def __init__(self, api_key: str, **kwargs):
     async def process(self, task: TaskMessage) -> dict:
         print(f"[*] RESEARCHER: Processing task -> {task.task_description[:50]}...")
 
-        # 1. Action: Search the web for real-time data
-        queries = [
-            "silicon wafer shipment growth 2026",
-            "AI chip demand statistics 2026",
-            "memory price forecast Q3 2026 semiconductor",
-        ]
-        search_data = "\n\n".join(web_search(q) for q in queries)
+        # Use task description as the search query (first 120 chars as a focused query)
+        search_query = task.task_description[:120].strip()
+        search_data = web_search(search_query, max_results=5)
         print(f"[DEBUG] DDG returned {len(search_data)} chars")
 
         system_prompt = (
@@ -48,18 +44,23 @@ async def process(self, task: TaskMessage) -> dict:
         )
 
         try:
-            # 2. Reasoning: Synthesis via Gemini
-            prompt = f"{system_prompt}\n\nTASK: {task.task_description}\n\nSEARCH DATA:\n{search_data}"
             response = await asyncio.to_thread(
                 self.client.chat.completions.create,
                 model=self.model,
                 messages=[
                     {"role": "system", "content": system_prompt},
-                    {"role": "user", "content": prompt},
+                    {
+                        "role": "user",
+                        "content": (
+                            f"TASK: {task.task_description}\n\n"
+                            f"SEARCH DATA:\n{search_data or 'No search results available.'}"
+                        ),
+                    },
                 ],
             )
 
             return {
+                "agent_type": "researcher",
                 "data": {
                     "research_content": response.choices[0].message.content,
                     "sources": ["DuckDuckGo Live Search", "Groq AI"],
@@ -72,6 +73,7 @@ async def process(self, task: TaskMessage) -> dict:
                 raise  # let Temporal retry after backoff
             # only fallback for non-quota errors
             return {
+                "agent_type": "researcher",
                 "data": {"research_content": search_data, "sources": ["DuckDuckGo"]},
                 "status": "partial_success",
             }

src/kin/gateway/main.py

@@ -1,37 +1,72 @@
 import os
+from contextlib import asynccontextmanager
+
 import redis.asyncio as redis
 from dotenv import load_dotenv
 from fastapi import FastAPI, HTTPException
 from pydantic import BaseModel
 from temporalio.client import Client
 from temporalio.contrib.pydantic import pydantic_data_converter
+
+from kin.models.schemas import TaskResult
+from kin.observability.logging import get_logger, setup_logging
 from kin.orchestrator.executor.dag_workflow import KinDAGWorkflow
-from kin.models.schemas import DAGSpec, TaskNode, TaskResult
 from kin.orchestrator.planner import Planner
 
 load_dotenv()
+setup_logging()
+log = get_logger("kin.gateway")
+
+RESULTS_TTL_SEC = 3600
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    log.info("Gateway starting up...")
+    app.state.temporal = await Client.connect(
+        "localhost:7233", data_converter=pydantic_data_converter
+    )
+    app.state.redis = redis.Redis(host="localhost", port=6379, decode_responses=True)
+    log.info("Connected to Temporal and Redis")
+    yield
+    await app.state.redis.aclose()
+    log.info("Gateway shut down cleanly")
 
-app = FastAPI(title="Kin AI Gateway")
+
+app = FastAPI(title="Kin AI Gateway", lifespan=lifespan)
 
 
 class WorkflowRequest(BaseModel):
     prompt: str
 
 
+# ---------------------------------------------------------------------------
+# Health check
+# ---------------------------------------------------------------------------
+@app.get("/healthz", tags=["ops"])
+async def health():
+    """Liveness probe — returns 200 when gateway is up and Redis is reachable."""
+    try:
+        await app.state.redis.ping()
+        redis_ok = True
+    except Exception:
+        redis_ok = False
+    return {"status": "ok", "redis": redis_ok}
+
+
+# ---------------------------------------------------------------------------
+# Workflow endpoints
+# ---------------------------------------------------------------------------
 @app.post("/v1/workflows")
 async def start_workflow(request: WorkflowRequest):
     try:
-        client = await Client.connect(
-            "localhost:7233", data_converter=pydantic_data_converter
-        )
-
-        # NEW: use planner
         planner = Planner(api_key=os.getenv("GROQ_API_KEY"))
         dag = planner.plan(request.prompt)
-
         dag_id = str(dag.workflow_id)
 
-        await client.start_workflow(
+        log.info("Starting workflow dag_id=%s nodes=%d", dag_id, len(dag.nodes))
+
+        await app.state.temporal.start_workflow(
             KinDAGWorkflow.run,
             dag,
             id=dag_id,
@@ -44,29 +79,22 @@ async def start_workflow(request: WorkflowRequest):
         }
 
     except Exception as e:
-        print(f"Error starting workflow: {e}")
+        log.error("Failed to start workflow: %s", e, exc_info=True)
         raise HTTPException(status_code=500, detail=str(e))
 
 
-# Initialize Redis client (ideally outside the function or in app state)
-redis_client = redis.Redis(host="localhost", port=6379, decode_responses=True)
-
-
 @app.get("/v1/workflows/{workflow_id}")
 async def get_status(workflow_id: str):
     try:
-        client = await Client.connect("localhost:7233")
-        handle = client.get_workflow_handle(workflow_id)
+        handle = app.state.temporal.get_workflow_handle(workflow_id)
         desc = await handle.describe()
 
         stream_key = f"results:{workflow_id}"
-        raw_entries = await redis_client.xrange(stream_key)
-
-        final_results = {}
+        raw_entries = await app.state.redis.xrange(stream_key)
 
+        final_results: dict = {}
         for _, entry in raw_entries:
             result = TaskResult.model_validate_json(entry["data"])
-
             final_results[result.node_id] = {
                 "status": result.status,
                 "agent_type": (
@@ -76,18 +104,24 @@ async def get_status(workflow_id: str):
                 "error": result.error,
             }
 
+        overall_status = desc.status.name
+
+        # Set TTL on result stream once workflow is terminal
+        if overall_status in ("COMPLETED", "FAILED", "TERMINATED", "TIMED_OUT"):
+            await app.state.redis.expire(stream_key, RESULTS_TTL_SEC)
+
         return {
             "workflow_id": workflow_id,
-            "status": desc.status.name,
+            "status": overall_status,
             "results": final_results,
         }
 
     except Exception as e:
+        log.warning("get_status error for %s: %s", workflow_id, e)
         raise HTTPException(status_code=404, detail=str(e))
 
 
 if __name__ == "__main__":
     import uvicorn
 
-    # Use the string "kin.gateway.main:app" for hot-reloading support
-    uvicorn.run(app, host="0.0.0.0", port=8000)
+    uvicorn.run("kin.gateway.main:app", host="0.0.0.0", port=8000, reload=True)

src/kin/models/schemas.py

@@ -17,7 +17,7 @@ class TaskNode(BaseModel):
     task_description: str
     input_from: List[str] = []
     dependencies: list[str] = []
-    timeout_sec: int = 120
+    timeout_sec: int = 600
     max_retries: int = 2
 
 

src/kin/observability/__init__.py

@@ -0,0 +1,3 @@
+from kin.observability.logging import get_logger, setup_logging
+
+__all__ = ["get_logger", "setup_logging"]

src/kin/observability/logging.py

@@ -0,0 +1,28 @@
+"""
+Centralized structured logging for Kin.
+Call setup_logging() once at process startup.
+"""
+
+import logging
+import sys
+
+
+def setup_logging(level: str = "INFO") -> None:
+    fmt = "%(asctime)s | %(levelname)-8s | %(name)s | %(message)s"
+    datefmt = "%Y-%m-%dT%H:%M:%S"
+
+    handler = logging.StreamHandler(sys.stdout)
+    handler.setFormatter(logging.Formatter(fmt, datefmt=datefmt))
+
+    root = logging.getLogger()
+    root.handlers.clear()
+    root.addHandler(handler)
+    root.setLevel(getattr(logging, level.upper(), logging.INFO))
+
+    # suppress noisy third-party loggers
+    for noisy in ("httpx", "httpcore", "temporalio", "uvicorn.access"):
+        logging.getLogger(noisy).setLevel(logging.WARNING)
+
+
+def get_logger(name: str) -> logging.Logger:
+    return logging.getLogger(name)

src/kin/orchestrator/executor/activities.py

@@ -88,12 +88,13 @@ async def dispatch_task(self, node: TaskNode, workflow_id: str) -> dict:
                             if result.node_id == node.id:
                                 if result.status == "FAILED":
                                     raise Exception(result.error)
-
-                                return {
-                                    "node_id": node.id,
-                                    "agent_type": node.agent_type,
-                                    "data": result.output,
-                                }
+                                if result.status == "COMPLETED":
+                                    return {
+                                        "node_id": node.id,
+                                        "agent_type": node.agent_type,
+                                        "data": result.output,
+                                    }
+                                # RUNNING — keep polling
 
             except asyncio.CancelledError:
                 activity.logger.warning(

src/kin/orchestrator/executor/dag_workflow.py

@@ -42,12 +42,21 @@ async def run(self, dag: DAGSpec) -> dict:
             for nid in ready:
                 node = node_map[nid]
                 if node.input_from:
-                    context = "\n".join(
-                        f"--- Output from {dep} ---\n{str(results.get(dep, ''))}"
-                        for dep in node.input_from
-                    )
+                    parts = []
+                    for dep in node.input_from:
+                        dep_result = results.get(dep, {})
+                        if isinstance(dep_result, dict):
+                            content = (
+                                dep_result.get("data", {}).get("research_content")
+                                or dep_result.get("markdown")
+                                or dep_result.get("research_content")
+                                or str(dep_result)
+                            )
+                        else:
+                            content = str(dep_result)
+                        parts.append("--- Research from " + dep + " ---\n" + content)
                     node.task_description += (
-                        f"\n\nCONTEXT FROM PREVIOUS STEPS:\n{context}"
+                        "\n\nCONTEXT FROM PREVIOUS RESEARCH:\n" + "\n\n".join(parts)
                     )
 
             # dispatch ready nodes in parallel
@@ -56,7 +65,11 @@ async def run(self, dag: DAGSpec) -> dict:
                     "dispatch_task",
                     args=[node_map[nid], str(dag.workflow_id)],
                     start_to_close_timeout=timedelta(seconds=node_map[nid].timeout_sec),
-                    retry_policy=RetryPolicy(maximum_attempts=1),
+                    retry_policy=RetryPolicy(
+                        maximum_attempts=3,
+                        initial_interval=timedelta(seconds=30),
+                        maximum_interval=timedelta(minutes=3),
+                    ),
                 )
                 for nid in ready
             ]