From 23aacf65b19151d0409e925f563a8ed1de17910f Mon Sep 17 00:00:00 2001 From: Atti Ur Rehman Date: Thu, 18 Jun 2026 19:12:28 +0300 Subject: [PATCH 1/2] =?UTF-8?q?Fix=20human=20approve=20=E2=86=92=20Slack?= =?UTF-8?q?=20by=20resuming=20from=20Redis=20state,=20not=20LangGraph=20ch?= =?UTF-8?q?eckpoint.=20Upstash=20lacks=20RedisJSON/FT=20required=20by=20Re?= =?UTF-8?q?disSaver;=20in-memory=20checkpoints=20also=20did=20not=20surviv?= =?UTF-8?q?e=20the=20approve=20background=20thread.=20execute=5Faction=20n?= =?UTF-8?q?ow=20calls=20run=5Faction=20on=20opscanvas:run:{run=5Fid},=20wi?= =?UTF-8?q?th=20checkpointer=20fallback,=20approve=20retry=20API,=20and=20?= =?UTF-8?q?a=20Retry=20Slack=20post=20UI=20when=20delivery=20fails.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 3 + backend/app/agents/action.py | 11 +++- backend/app/api/incidents.py | 46 ++++++++++++- backend/app/core/graph_checkpointer.py | 31 +++++++-- backend/app/core/graph_runner.py | 75 +++++++++++---------- frontend/src/components/ApprovalPanel.tsx | 13 ++-- frontend/src/hooks/usePollIncident.ts | 8 ++- frontend/src/pages/DashboardPage.tsx | 52 ++++++++++++++- frontend/src/types/api.ts | 3 +- tests/test_action_retry.py | 80 +++++++++++++++++++++++ tests/test_durability.py | 25 ++----- 11 files changed, 275 insertions(+), 72 deletions(-) create mode 100644 tests/test_action_retry.py diff --git a/.env.example b/.env.example index 2993689..6542b7f 100644 --- a/.env.example +++ b/.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 diff --git a/backend/app/agents/action.py b/backend/app/agents/action.py index 3eff678..e025f96 100644 --- a/backend/app/agents/action.py +++ b/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) diff --git a/backend/app/api/incidents.py b/backend/app/api/incidents.py index eabe133..eb4430e 100644 --- a/backend/app/api/incidents.py +++ b/backend/app/api/incidents.py @@ -157,6 +157,26 @@ 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 +373,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", "") diff --git a/backend/app/core/graph_checkpointer.py b/backend/app/core/graph_checkpointer.py index 1b2e734..093edd3 100644 --- a/backend/app/core/graph_checkpointer.py +++ b/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: @@ -48,13 +59,13 @@ def graph_checkpointer(): """ 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: diff --git a/backend/app/core/graph_runner.py b/backend/app/core/graph_runner.py index 036bfa3..b19ec46 100644 --- a/backend/app/core/graph_runner.py +++ b/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,37 @@ 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() diff --git a/frontend/src/components/ApprovalPanel.tsx b/frontend/src/components/ApprovalPanel.tsx index 129cd72..bf274ae 100644 --- a/frontend/src/components/ApprovalPanel.tsx +++ b/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 = { @@ -29,7 +30,7 @@ const SEV_LABEL: Record = { }; 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}