refactor(slack-gateway): recover stale runtime sessions

Author: Onur Solmaz

helm/spritz/templates/operator-deployment.yaml

@@ -53,6 +53,18 @@ spec:
             - name: SPRITZ_HOME_SIZE_LIMIT
               value: {{ .Values.operator.homeSizeLimit | quote }}
             {{- end }}
+            {{- if .Values.operator.lifecycleNotifications.url }}
+            - name: SPRITZ_LIFECYCLE_NOTIFY_URL
+              value: {{ .Values.operator.lifecycleNotifications.url | quote }}
+            {{- end }}
+            {{- if .Values.operator.lifecycleNotifications.authToken }}
+            - name: SPRITZ_LIFECYCLE_NOTIFY_AUTH_TOKEN
+              value: {{ .Values.operator.lifecycleNotifications.authToken | quote }}
+            {{- end }}
+            {{- if .Values.operator.lifecycleNotifications.timeout }}
+            - name: SPRITZ_LIFECYCLE_NOTIFY_TIMEOUT
+              value: {{ .Values.operator.lifecycleNotifications.timeout | quote }}
+            {{- end }}
             - name: SPRITZ_ROUTE_MODEL_TYPE
               value: {{ include "spritz.routeModel.type" . | quote }}
             - name: SPRITZ_ROUTE_HOST

helm/spritz/values.yaml

@@ -105,6 +105,10 @@ operator:
   workspaceSizeLimit: 10Gi
   homeSizeLimit: 5Gi
   podNodeSelector: ""
+  lifecycleNotifications:
+    url: ""
+    authToken: ""
+    timeout: 3s
   sharedMounts:
     enabled: false
     mounts: []

integrations/slack-gateway/backend_client.go

@@ -126,6 +126,17 @@ func channelSessionUnavailableProviderAuth(err error) (slackInstallation, bool)
 	return unavailableErr.providerAuth, true
 }
 
+func isSpritzRuntimeMissingError(err error) bool {
+	var statusErr *httpStatusError
+	if !errors.As(err, &statusErr) {
+		return false
+	}
+	if statusErr.statusCode != http.StatusNotFound {
+		return false
+	}
+	return strings.Contains(strings.ToLower(statusErr.body), "spritz not found")
+}
+
 func (g *slackGateway) exchangeChannelSession(ctx context.Context, teamID string) (channelSession, error) {
 	body := map[string]any{
 		"principalId":       g.cfg.PrincipalID,

integrations/slack-gateway/gateway_test.go

@@ -2176,6 +2176,219 @@ func TestProcessMessageEventPostsStatusMessageWhileSessionRecoveryIsInFlight(t *
 	}
 }
 
+func TestProcessMessageEventRecoversAfterRuntimeDisappearsMidFlight(t *testing.T) {
+	var slackPayloads struct {
+		sync.Mutex
+		items []map[string]any
+	}
+	slackAPI := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/chat.postMessage" {
+			t.Fatalf("unexpected slack path %s", r.URL.Path)
+		}
+		var payload map[string]any
+		if err := json.NewDecoder(r.Body).Decode(&payload); err != nil {
+			t.Fatalf("decode slack post body: %v", err)
+		}
+		slackPayloads.Lock()
+		slackPayloads.items = append(slackPayloads.items, payload)
+		slackPayloads.Unlock()
+		writeJSON(w, http.StatusOK, map[string]any{"ok": true, "ts": fmt.Sprintf("1711387375.00010%d", len(slackPayloads.items))})
+	}))
+	defer slackAPI.Close()
+
+	var sessionExchangeCalls atomic.Int32
+	backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/internal/v1/spritz/channel-sessions/exchange" {
+			t.Fatalf("unexpected backend path %s", r.URL.Path)
+		}
+		call := sessionExchangeCalls.Add(1)
+		switch call {
+		case 1, 2:
+			writeJSON(w, http.StatusOK, map[string]any{
+				"status": "resolved",
+				"session": map[string]any{
+					"accessToken": "owner-token-old",
+					"ownerAuthId": "owner-123",
+					"namespace":   "spritz-staging",
+					"instanceId":  "zeno-old",
+					"providerAuth": map[string]any{
+						"providerInstallRef": "cred_slack_workspace_1",
+						"apiAppId":           "A_app_1",
+						"teamId":             "T_workspace_1",
+						"botUserId":          "U_bot",
+						"botAccessToken":     "xoxb-installed",
+					},
+				},
+			})
+		case 3:
+			writeJSON(w, http.StatusServiceUnavailable, map[string]any{
+				"status": "unavailable",
+				"providerAuth": map[string]any{
+					"providerInstallRef": "cred_slack_workspace_1",
+					"apiAppId":           "A_app_1",
+					"teamId":             "T_workspace_1",
+					"botUserId":          "U_bot",
+					"botAccessToken":     "xoxb-installed",
+				},
+			})
+		default:
+			writeJSON(w, http.StatusOK, map[string]any{
+				"status": "resolved",
+				"session": map[string]any{
+					"accessToken": "owner-token-new",
+					"ownerAuthId": "owner-123",
+					"namespace":   "spritz-staging",
+					"instanceId":  "zeno-new",
+					"providerAuth": map[string]any{
+						"providerInstallRef": "cred_slack_workspace_1",
+						"apiAppId":           "A_app_1",
+						"teamId":             "T_workspace_1",
+						"botUserId":          "U_bot",
+						"botAccessToken":     "xoxb-installed",
+					},
+				},
+			})
+		}
+	}))
+	defer backend.Close()
+
+	var upsertCalls atomic.Int32
+	upgrader := websocket.Upgrader{CheckOrigin: func(r *http.Request) bool { return true }}
+	spritz := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch r.URL.Path {
+		case "/api/channel-conversations/upsert":
+			call := upsertCalls.Add(1)
+			if call == 1 {
+				http.Error(w, `{"status":"error","message":"spritz not found"}`, http.StatusNotFound)
+				return
+			}
+			writeJSON(w, http.StatusCreated, map[string]any{
+				"status": "success",
+				"data": map[string]any{
+					"created": true,
+					"conversation": map[string]any{
+						"metadata": map[string]any{"name": "conv-1"},
+						"spec":     map[string]any{"cwd": "/home/dev"},
+					},
+				},
+			})
+		case "/api/acp/conversations/conv-1/bootstrap":
+			writeJSON(w, http.StatusOK, map[string]any{
+				"status": "success",
+				"data": map[string]any{
+					"effectiveSessionId": "session-1",
+					"conversation": map[string]any{
+						"metadata": map[string]any{"name": "conv-1"},
+						"spec":     map[string]any{"sessionId": "session-1", "cwd": "/home/dev"},
+					},
+				},
+			})
+		case "/api/acp/conversations/conv-1/connect":
+			conn, err := upgrader.Upgrade(w, r, nil)
+			if err != nil {
+				t.Fatalf("upgrade failed: %v", err)
+			}
+			defer conn.Close()
+			for {
+				_, payload, err := conn.ReadMessage()
+				if err != nil {
+					return
+				}
+				var message map[string]any
+				if err := json.Unmarshal(payload, &message); err != nil {
+					t.Fatalf("decode ws payload: %v", err)
+				}
+				switch message["method"] {
+				case "initialize":
+					_ = conn.WriteJSON(map[string]any{"jsonrpc": "2.0", "id": message["id"], "result": map[string]any{"protocolVersion": 1}})
+				case "session/load":
+					_ = conn.WriteJSON(map[string]any{"jsonrpc": "2.0", "id": message["id"], "result": map[string]any{}})
+				case "session/prompt":
+					_ = conn.WriteJSON(map[string]any{
+						"jsonrpc": "2.0",
+						"method":  "session/update",
+						"params": map[string]any{
+							"update": map[string]any{
+								"sessionUpdate": "agent_message_chunk",
+								"content": []map[string]any{{
+									"type": "text",
+									"text": "Hello from recovered concierge",
+								}},
+							},
+						},
+					})
+					_ = conn.WriteJSON(map[string]any{"jsonrpc": "2.0", "id": message["id"], "result": map[string]any{}})
+					return
+				default:
+					t.Fatalf("unexpected ACP method %#v", message["method"])
+				}
+			}
+		default:
+			t.Fatalf("unexpected spritz path %s", r.URL.Path)
+		}
+	}))
+	defer spritz.Close()
+
+	cfg := config{
+		SlackAPIBaseURL:      slackAPI.URL,
+		BackendBaseURL:       backend.URL,
+		BackendInternalToken: "backend-internal-token",
+		SpritzBaseURL:        spritz.URL,
+		SpritzServiceToken:   "spritz-service-token",
+		PrincipalID:          "shared-slack-gateway",
+		HTTPTimeout:          200 * time.Millisecond,
+		DedupeTTL:            time.Minute,
+		StatusMessageDelay:   5 * time.Millisecond,
+		SessionRetryInterval: 10 * time.Millisecond,
+		ProcessingTimeout:    200 * time.Millisecond,
+	}
+	gateway := newSlackGateway(cfg, slog.New(slog.NewTextHandler(io.Discard, nil)))
+
+	envelope := slackEnvelope{
+		APIAppID: "A_app_1",
+		TeamID:   "T_workspace_1",
+		Event: slackEventInner{
+			Type:        "app_mention",
+			User:        "U_user",
+			Text:        "<@U_bot> hello",
+			Channel:     "C_channel_1",
+			ChannelType: "channel",
+			TS:          "1711387375.000100",
+		},
+	}
+	delivery, process, err := gateway.beginMessageEventDelivery(envelope)
+	if err != nil {
+		t.Fatalf("beginMessageEventDelivery returned error: %v", err)
+	}
+	if !process {
+		t.Fatal("expected app mention to be processed")
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
+	defer cancel()
+	if err := gateway.processMessageEventWithDelivery(ctx, envelope, delivery); err != nil {
+		t.Fatalf("expected missing-runtime recovery flow to succeed, got %v", err)
+	}
+
+	slackPayloads.Lock()
+	defer slackPayloads.Unlock()
+	if len(slackPayloads.items) != 2 {
+		t.Fatalf("expected wake-up message and final reply, got %#v", slackPayloads.items)
+	}
+	if got := slackPayloads.items[0]["text"]; got != "Still waking up. I will continue here shortly." {
+		t.Fatalf("expected wake-up status text, got %#v", got)
+	}
+	if got := slackPayloads.items[1]["text"]; got != "Hello from recovered concierge" {
+		t.Fatalf("expected final reply text, got %#v", got)
+	}
+	if sessionExchangeCalls.Load() != 4 {
+		t.Fatalf("expected 4 session exchange attempts, got %d", sessionExchangeCalls.Load())
+	}
+	if upsertCalls.Load() != 3 {
+		t.Fatalf("expected recovery retry plus alias persistence, got %d", upsertCalls.Load())
+	}
+}
+
 func TestProcessMessageEventPostsTerminalErrorAfterRecoveryTimeout(t *testing.T) {
 	var slackPayloads struct {
 		sync.Mutex