kin: KIN-135-backend_dev

agents/runner.py

@@ -1127,6 +1127,97 @@ def _save_tech_debt_output(
     return {"created": True, "task_id": new_task_id}
 
 
+# ---------------------------------------------------------------------------
+# Return analyst output: handle escalation pipeline creation (KIN-135)
+# ---------------------------------------------------------------------------
+
+# Mapping from task category to escalation dept_head (KIN-135)
+_CATEGORY_TO_DEPT_HEAD = {
+    "SEC": "security_head",
+    "UI": "frontend_head",
+    "API": "backend_head",
+    "BIZ": "backend_head",
+    "DB": "backend_head",
+    "PERF": "backend_head",
+    "ARCH": "cto_advisor",
+    "FIX": "cto_advisor",
+    "INFRA": "infra_head",
+    "OBS": "infra_head",
+    "TEST": "qa_head",
+    "DOCS": "cto_advisor",
+}
+
+
+def _save_return_analyst_output(
+    conn: sqlite3.Connection,
+    task_id: str,
+    project_id: str,
+    result: dict,
+    parent_pipeline_id: int | None = None,
+) -> dict:
+    """Parse return_analyst output and create escalation pipeline if needed.
+
+    If escalate_to_dept_head=true in analyst output:
+    - Determines target dept_head from task.category (defaults to cto_advisor)
+    - Creates a new escalation pipeline with pipeline_type='escalation'
+    - The dept_head step receives analyst output as initial context
+
+    Returns {"escalated": bool, "escalation_pipeline_id": int | None, "dept_head": str | None}.
+    Never raises — escalation errors must never block the current pipeline.
+    """
+    raw = result.get("raw_output") or result.get("output") or ""
+    if isinstance(raw, (dict, list)):
+        raw = json.dumps(raw, ensure_ascii=False)
+
+    parsed = _try_parse_json(raw)
+    if not isinstance(parsed, dict):
+        return {"escalated": False, "escalation_pipeline_id": None, "dept_head": None}
+
+    if not parsed.get("escalate_to_dept_head"):
+        return {"escalated": False, "escalation_pipeline_id": None, "dept_head": None}
+
+    # Determine target dept_head from task category
+    task = models.get_task(conn, task_id)
+    category = (task or {}).get("category") or ""
+    dept_head = _CATEGORY_TO_DEPT_HEAD.get(category.upper(), "cto_advisor")
+
+    # Build escalation pipeline steps: dept_head with analyst context as brief
+    analyst_summary = parsed.get("root_cause_analysis", "")
+    refined_brief = parsed.get("refined_brief", "")
+    escalation_brief = (
+        f"[ESCALATION from return_analyst]\n"
+        f"Root cause: {analyst_summary}\n"
+        f"Refined brief: {refined_brief}"
+    )
+    escalation_steps = [{"role": dept_head, "model": "opus", "brief": escalation_brief}]
+
+    try:
+        esc_pipeline = models.create_pipeline(
+            conn, task_id, project_id,
+            route_type="escalation",
+            steps=escalation_steps,
+            parent_pipeline_id=parent_pipeline_id,
+        )
+        # Mark pipeline_type as escalation so return tracking is skipped inside it
+        conn.execute(
+            "UPDATE pipelines SET pipeline_type = 'escalation' WHERE id = ?",
+            (esc_pipeline["id"],),
+        )
+        conn.commit()
+        _logger.info(
+            "KIN-135: escalation pipeline %s created for task %s → %s",
+            esc_pipeline["id"], task_id, dept_head,
+        )
+        return {
+            "escalated": True,
+            "escalation_pipeline_id": esc_pipeline["id"],
+            "dept_head": dept_head,
+        }
+    except Exception as exc:
+        _logger.warning("KIN-135: escalation pipeline creation failed for %s: %s", task_id, exc)
+        return {"escalated": False, "escalation_pipeline_id": None, "dept_head": None}
+
+
 # ---------------------------------------------------------------------------
 # Auto-learning: extract decisions from pipeline results
 # ---------------------------------------------------------------------------
@@ -1978,6 +2069,16 @@ def run_pipeline(
             except Exception:
                 pass  # Never block pipeline on decomposer save errors
 
+        # Return analyst: create escalation pipeline if escalate_to_dept_head=true (KIN-135)
+        if role == "return_analyst" and result["success"] and not dry_run:
+            try:
+                _save_return_analyst_output(
+                    conn, task_id, project_id, result,
+                    parent_pipeline_id=pipeline["id"] if pipeline else None,
+                )
+            except Exception:
+                pass  # Never block pipeline on analyst escalation errors
+
         # Smoke tester: parse result and escalate if cannot_confirm (KIN-128)
         if role == "smoke_tester" and result["success"] and not dry_run:
             smoke_output = result.get("output") or result.get("raw_output") or ""
@@ -2423,6 +2524,20 @@ def run_pipeline(
                         total_tokens=total_tokens,
                         total_duration_seconds=total_duration,
                     )
+                # KIN-135: record gate return — skip for escalation pipelines to avoid loops
+                _pipeline_type = (pipeline or {}).get("pipeline_type", "standard")
+                if _pipeline_type != "escalation":
+                    try:
+                        models.record_task_return(
+                            conn,
+                            task_id=task_id,
+                            reason_category="recurring_quality_fail",
+                            reason_text=f"Gate {effective_last_role}: {_block_reason[:200]}",
+                            returned_by=effective_last_role,
+                            pipeline_id=pipeline["id"] if pipeline else None,
+                        )
+                    except Exception:
+                        pass  # Never block pipeline on return tracking errors
                 try:
                     models.write_log(
                         conn, pipeline["id"] if pipeline else None,