Fix human approve → Slack by resuming from Redis state, not LangGraph…

.env.example

@@ -43,6 +43,9 @@ QUOTA_OVERRIDE_EMAIL=your@gmail.com:20
 # Layer 5 — Durability (reuse existing Upstash creds)
 UPSTASH_REDIS_REST_URL=...
 UPSTASH_REDIS_REST_TOKEN=...
+# Local dev: memory avoids Upstash RedisJSON/FT limits for LangGraph checkpoints.
+# Approve→Slack resume always uses redis_store, not the LangGraph checkpoint.
+OPSCANVAS_CHECKPOINTER=memory
 APPROVAL_TTL_HOURS=10
 
 # Layer 6 — Reliability

backend/app/agents/action.py

@@ -43,6 +43,7 @@
 from app.core.redis_store import is_slack_sent, mark_slack_sent
 from app.graph.state import IncidentState
 from app.guardrails.action_auth import authorize_action
+from app.guardrails.errors import GuardrailError
 from app.guardrails.slack_output import validate_slack_payload
 
 logger = logging.getLogger(__name__)
@@ -194,7 +195,15 @@ def post_to_slack(state: IncidentState) -> bool:
         "attachments": [{"color": _severity_color(severity)}],
     }
 
-    validate_slack_payload(payload)
+    try:
+        validate_slack_payload(payload)
+    except GuardrailError as exc:
+        logger.error(
+            "Slack payload blocked by guardrail — run_id=%s: %s",
+            state.get("run_id"),
+            exc,
+        )
+        return False
 
     try:
         _post_slack_request(webhook_url, payload)

backend/app/api/incidents.py

@@ -157,6 +157,28 @@ def _get_run_status(state: dict) -> str:
     return "running"
 
 
+def _is_feedback_loop_escalation(state: dict) -> bool:
+    """True when escalated because human rejected too many times — not retryable."""
+    err = state.get("_error") or ""
+    if "max_feedback_loops" in err:
+        return True
+    if state.get("_pipeline_phase") == "escalate" and not err.startswith(
+        "action_failed"
+    ):
+        return state.get("retry_count", 0) >= max_feedback_loops()
+    return False
+
+
+def _can_retry_slack_action(state: dict) -> bool:
+    """Approve again when Slack/action failed but feedback-loop cap was not hit."""
+    if state.get("slack_posted"):
+        return False
+    if _is_feedback_loop_escalation(state):
+        return False
+    status = _get_run_status(state)
+    return status in ("failed", "escalated", "running")
+
+
 def _to_incident_response(run_id: str, state: dict) -> IncidentRunResponse:
     costs = cost_from_state(state)
     return IncidentRunResponse(
@@ -353,10 +375,34 @@ def review_incident(
         )
 
     current_status = _get_run_status(state)
+    if body.decision == "approved" and _can_retry_slack_action(state):
+        approver_sub = user_info.get("sub", "")
+        approver_email = user_info.get("email", "")
+        state["approver_sub"] = approver_sub
+        state["approver_email"] = approver_email
+        state["human_decision"] = "approved"
+        state["human_feedback"] = body.feedback
+        state.pop("outcome", None)
+        state.pop("_error", None)
+        state["_status"] = "running"
+        state["_pipeline_phase"] = "action"
+        record_action_audit(state, "review_approved", approver=approver_sub)
+        save_state(run_id, state)
+        dispatch_background("run_action", run_id)
+        return _to_incident_response(run_id, state)
+
     if current_status not in ("awaiting_review", "running"):
+        if body.decision == "approved" and _is_feedback_loop_escalation(state):
+            detail = (
+                f"Run {run_id} exhausted feedback retries — operator review required"
+            )
+        elif body.decision == "approved" and state.get("slack_posted"):
+            detail = f"Run {run_id} already completed — Slack was posted"
+        else:
+            detail = f"Run {run_id} is '{current_status}' — cannot review"
         raise HTTPException(
             status_code=status.HTTP_409_CONFLICT,
-            detail=f"Run {run_id} is '{current_status}' — cannot review",
+            detail=detail,
         )
 
     approver_sub = user_info.get("sub", "")

backend/app/core/graph_checkpointer.py

@@ -16,6 +16,17 @@
 logger = logging.getLogger(__name__)
 
 _setup_complete = False
+_memory_saver = None
+
+
+def _shared_memory_saver():
+    """One in-process checkpointer — used for tests and Upstash fallback."""
+    global _memory_saver
+    if _memory_saver is None:
+        from langgraph.checkpoint.memory import MemorySaver
+
+        _memory_saver = MemorySaver()
+    return _memory_saver
 
 
 def graph_thread_config(run_id: str) -> dict:
@@ -46,15 +57,15 @@ def _checkpoint_ttl_seconds() -> int | None:
 @contextmanager
 def graph_checkpointer():
     """
-    Yield a LangGraph checkpointer for one graph execution.
+      Yield a LangGraph checkpointer for one graph execution.
 
-    OPSCANVAS_CHECKPOINTER=memory forces in-memory checkpoints (tests).
+      OPSCANVAS_CHECKPOINTER=memory forces in-memory checkpoints (local dev/tests).
+      When redis is selected but Upstash cannot run RedisSaver setup, falls back
+    to the shared in-process MemorySaver for the initial graph run only.
     """
     mode = os.environ.get("OPSCANVAS_CHECKPOINTER", "redis").lower()
     if mode == "memory":
-        from langgraph.checkpoint.memory import MemorySaver
-
-        yield MemorySaver()
+        yield _shared_memory_saver()
         return
 
     from langgraph.checkpoint.redis import RedisSaver
@@ -69,17 +80,23 @@ def graph_checkpointer():
     )
 
     global _setup_complete
+    use_saver = saver
     if not _setup_complete:
         try:
             saver.setup()
             _setup_complete = True
         except Exception as exc:
-            logger.warning("Redis checkpointer setup failed (will retry): %s", exc)
+            logger.warning(
+                "Redis checkpointer unavailable (%s) — using in-process memory "
+                "(human-approve resume uses redis_store, not this checkpoint)",
+                exc,
+            )
+            use_saver = _shared_memory_saver()
 
     try:
-        yield saver
+        yield use_saver
     finally:
-        if saver._owns_its_client:
+        if use_saver is saver and saver._owns_its_client:
             saver._redis.close()
             pool = getattr(saver._redis, "connection_pool", None)
             if pool is not None:

backend/app/core/graph_runner.py

@@ -11,8 +11,6 @@
 import logging
 from typing import Any
 
-from langgraph.types import Command
-
 from app.core.approval_ttl import expire_pending_approval, is_approval_expired
 from app.core.cost_meter import CostBreakerError
 from app.core.dead_letter import dead_letter_incident
@@ -29,6 +27,20 @@
 logger = logging.getLogger(__name__)
 
 
+def _mark_action_failed(run_id: str, exc: Exception, root: Any) -> None:
+    """Slack/action failure — failed status, retryable via POST /review approve."""
+    try:
+        current = load_state(run_id) or {}
+        current["_status"] = "failed"
+        current["_pipeline_phase"] = "action"
+        current["_error"] = f"action_failed: {exc}"
+        # Keep human_decision=approved so the UI can retry posting to Slack.
+        save_state(run_id, current)
+        finalize_root_span(root, current)
+    except Exception:
+        pass
+
+
 def _mark_run_failed(
     run_id: str,
     exc: Exception,
@@ -207,10 +219,14 @@ def execute_synthesiser_retry(run_id: str) -> None:
 
 def execute_action(run_id: str, *, approver: str | None = None) -> None:
     """
-    Resume a checkpointed graph after human approval.
+    Run the action node after human approval.
 
-    Runs the action node on a new Lambda/process; Slack delivery is idempotent.
+    Resume uses opscanvas:run:{run_id} in Redis — not LangGraph RedisSaver.
+    Upstash lacks RedisJSON/FT commands RedisSaver needs, and in-memory
+    checkpoints do not survive a second thread/Lambda invocation.
     """
+    from app.agents.action import run_action
+
     with incident_root_span(run_id, "run_action") as root:
         try:
             current = load_state(run_id)
@@ -233,48 +249,36 @@ def execute_action(run_id: str, *, approver: str | None = None) -> None:
                 finalize_root_span(root, current)
                 return
 
+            current["run_id"] = run_id
+            current["human_decision"] = "approved"
+            if approver and not current.get("approver_sub"):
+                current["approver_sub"] = approver
             current["_pipeline_phase"] = "action"
             current["_status"] = "running"
             save_state(run_id, current)
 
-            with graph_checkpointer() as checkpointer:
-                graph = build_graph(checkpointer)
-                config = graph_thread_config(run_id)
-
-                graph.update_state(
-                    config,
-                    {
-                        "human_decision": "approved",
-                        "human_feedback": current.get("human_feedback"),
-                        "_pipeline_phase": "action",
-                        "_status": "running",
-                    },
+            updates = run_action(current)
+            final = {**current, **updates}
+            final["_status"] = "completed" if final.get("slack_posted") else "failed"
+            final["_pipeline_phase"] = "action"
+            if not final.get("slack_posted"):
+                final["_error"] = (
+                    final.get("_error") or "action_failed: slack post returned false"
                 )
-
-                graph.invoke(
-                    Command(
-                        resume={
-                            "approval": current.get("approver_sub")
-                            or approver
-                            or "engineer"
-                        }
-                    ),
-                    config,
-                )
-
-                final = _merge_checkpoint_to_redis(run_id, graph, config)
-                final["_status"] = "completed"
-                final["_pipeline_phase"] = "action"
-                save_state(run_id, final)
+            save_state(run_id, final)
 
             finalize_root_span(root, final)
-            logger.info("Action complete — run_id=%s", run_id)
+            logger.info(
+                "Action complete — run_id=%s slack_posted=%s",
+                run_id,
+                final.get("slack_posted"),
+            )
         except CostBreakerError as exc:
             logger.error("Cost breaker on action — run_id=%s: %s", run_id, exc)
-            _mark_run_failed(run_id, exc, root, cost_breaker=True)
+            _mark_run_failed(run_id, exc, root, cost_breaker=True, dead_letter=False)
         except Exception as exc:
             logger.error("Action failed — run_id=%s: %s", run_id, exc)
-            _mark_run_failed(run_id, exc, root)
+            _mark_action_failed(run_id, exc, root)
     flush_langfuse()
 
 

frontend/src/components/ApprovalPanel.tsx

@@ -7,10 +7,11 @@ import { useState } from "react";
 import type { IncidentRun } from "../types/api";
 
 interface ApprovalPanelProps {
-  run:       IncidentRun;
-  onApprove: () => void;
-  onReject:  (feedback: string) => void;
-  loading:   boolean;
+  run:          IncidentRun;
+  onApprove:    () => void;
+  onReject:     (feedback: string) => void;
+  loading:      boolean;
+  approveLabel?: string;
 }
 
 const SEV_BADGE: Record<string, string> = {
@@ -29,7 +30,7 @@ const SEV_LABEL: Record<string, string> = {
 };
 
 export default function ApprovalPanel({
-  run, onApprove, onReject, loading,
+  run, onApprove, onReject, loading, approveLabel = "Approve — post to Slack",
 }: ApprovalPanelProps) {
   const [mode, setMode]         = useState<"idle" | "rejecting">("idle");
   const [feedback, setFeedback] = useState("");
@@ -139,7 +140,7 @@ export default function ApprovalPanel({
             disabled={loading}
             className="flex-1 px-4 py-3 rounded-xl bg-green-50 border border-green-200 text-green-700 text-sm font-semibold hover:bg-green-100 active:scale-[0.99] disabled:opacity-40 disabled:cursor-not-allowed transition-all duration-150"
           >
-            {loading ? "Processing…" : "Approve — post to Slack"}
+            {loading ? "Processing…" : approveLabel}
           </button>
           <button
             onClick={() => setMode("rejecting")}

frontend/src/hooks/usePollIncident.ts

@@ -7,7 +7,13 @@ import { api, ApiError } from "../api/client";
 import type { IncidentRun, RunStatus } from "../types/api";
 
 const POLL_INTERVAL_MS = 3_000;
-const TERMINAL_STATUSES: RunStatus[] = ["awaiting_review", "completed", "failed"];
+const TERMINAL_STATUSES: RunStatus[] = [
+  "awaiting_review",
+  "completed",
+  "failed",
+  "escalated",
+  "expired",
+];
 
 interface PollState {
   run:     IncidentRun | null;