fix(relay): prevent agent session corruption on timeout (#405)

* fix(feishu): respond to @all messages in group chats

Feishu @all sends {"text":"@_all"} with 0 mentions, which was being
ignored when group_reply_all=false. Add respond_to_at_everyone_and_here
option to allow bots to respond to @all messages.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix(relay): prevent agent session corruption on timeout

Three issues caused agents to become unresponsive after a relay timeout:

1. StartSession received the relay timeout context -- when the deadline
   fired, exec.CommandContext killed the Claude process mid-turn,
   corrupting the session state. Now uses the engine context instead.

2. defer agentSession.Close() killed the process immediately on return.
   Now the session drains in a background goroutine (drainRelaySession)
   so the agent can finish its turn and save state cleanly.

3. No fallback when resuming a corrupted session. Added retry logic that
   clears the stale session ID and starts fresh, matching the existing
   interactive message flow.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

---------

Co-authored-by: pinxi_mei <pinxi_mei@lebo.cn>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

core/engine.go

@@ -8990,7 +8990,7 @@ func (e *Engine) sendTTSReply(p Platform, replyCtx any, text string) {
 
 // HandleRelay processes a relay message synchronously: starts or resumes a
 // dedicated relay session, sends the message to the agent, and blocks until
-// the complete response is collected.
+// the complete response is collected (or the relay context times out).
 func (e *Engine) HandleRelay(ctx context.Context, fromProject, chatID, message string) (string, error) {
 	relaySessionKey := "relay:" + fromProject + ":" + chatID
 	session := e.sessions.GetOrCreateActive(relaySessionKey)
@@ -9009,23 +9009,41 @@ func (e *Engine) HandleRelay(ctx context.Context, fromProject, chatID, message s
 		inj.SetSessionEnv(envVars)
 	}
 
-	agentSession, err := e.agent.StartSession(ctx, session.GetAgentSessionID())
+	// Use the engine context (not the relay timeout context) so that the
+	// agent process is not killed when the relay deadline fires. The relay
+	// timeout only controls how long we *wait* for the response.
+	agentSession, err := e.agent.StartSession(e.ctx, session.GetAgentSessionID())
 	if err != nil {
-		return "", fmt.Errorf("start relay session: %w", err)
+		// Resume failed — fall back to a fresh session so the relay is not
+		// permanently broken by a corrupted/stale session ID.
+		if session.GetAgentSessionID() != "" {
+			slog.Warn("relay: session resume failed, trying fresh session",
+				"relay_key", relaySessionKey, "error", err)
+			session.SetAgentSessionID("", e.agent.Name())
+			e.sessions.Save()
+			agentSession, err = e.agent.StartSession(e.ctx, "")
+		}
+		if err != nil {
+			return "", fmt.Errorf("start relay session: %w", err)
+		}
 	}
-	defer agentSession.Close()
 
 	if session.CompareAndSetAgentSessionID(agentSession.CurrentSessionID(), e.agent.Name()) {
 		e.sessions.Save()
 	}
 
 	if err := agentSession.Send(message, nil, nil); err != nil {
+		agentSession.Close()
 		return "", fmt.Errorf("send relay message: %w", err)
 	}
 
 	var textParts []string
 	for event := range agentSession.Events() {
 		if ctx.Err() != nil {
+			// Relay timed out. Let the agent finish its turn in the
+			// background so the session state is saved cleanly and the
+			// session remains resumable for the next relay call.
+			go e.drainRelaySession(agentSession, session, relaySessionKey)
 			return relayPartialResponseOrError(ctx.Err(), textParts, fromProject, e.name)
 		}
 		switch event.Type {
@@ -9063,8 +9081,10 @@ func (e *Engine) HandleRelay(ctx context.Context, fromProject, chatID, message s
 				resp = "(empty response)"
 			}
 			slog.Info("relay: turn complete", "from", fromProject, "to", e.name, "response_len", len(resp))
+			agentSession.Close()
 			return resp, nil
 		case EventError:
+			agentSession.Close()
 			if event.Error != nil {
 				return "", event.Error
 			}
@@ -9078,6 +9098,9 @@ func (e *Engine) HandleRelay(ctx context.Context, fromProject, chatID, message s
 		}
 	}
 
+	// Event channel closed without EventResult.
+	agentSession.Close()
+
 	if ctx.Err() != nil {
 		return relayPartialResponseOrError(ctx.Err(), textParts, fromProject, e.name)
 	}
@@ -9103,6 +9126,54 @@ func relayPartialResponseOrError(ctxErr error, textParts []string, fromProject,
 	return resp, nil
 }
 
+// drainRelaySession runs in a goroutine after a relay timeout. It lets the
+// agent finish its current turn (saving the session ID for future resumption),
+// auto-approves any permission requests, and then closes the session. A 10-minute
+// safety timeout prevents the goroutine from leaking if the agent hangs.
+func (e *Engine) drainRelaySession(agentSession AgentSession, session *Session, relaySessionKey string) {
+	timer := time.NewTimer(10 * time.Minute)
+	defer timer.Stop()
+
+	for {
+		select {
+		case ev, ok := <-agentSession.Events():
+			if !ok {
+				// Event channel closed — session ended naturally.
+				return
+			}
+			if ev.SessionID != "" {
+				session.SetAgentSessionID(ev.SessionID, e.agent.Name())
+				e.sessions.Save()
+			}
+			switch ev.Type {
+			case EventResult:
+				slog.Info("relay: background drain completed (agent finished turn)",
+					"relay_key", relaySessionKey)
+				agentSession.Close()
+				return
+			case EventError:
+				slog.Warn("relay: background drain got error",
+					"relay_key", relaySessionKey, "error", ev.Error)
+				agentSession.Close()
+				return
+			case EventPermissionRequest:
+				_ = agentSession.RespondPermission(ev.RequestID, PermissionResult{
+					Behavior:     "allow",
+					UpdatedInput: ev.ToolInputRaw,
+				})
+			}
+		case <-timer.C:
+			slog.Warn("relay: background drain timed out, closing session",
+				"relay_key", relaySessionKey)
+			agentSession.Close()
+			return
+		case <-e.ctx.Done():
+			agentSession.Close()
+			return
+		}
+	}
+}
+
 // cmdBind handles /bind — establishes a relay binding between bots in a group chat.
 //
 // Usage:

core/relay.go

@@ -289,10 +289,15 @@ func (rm *RelayManager) relayContext(ctx context.Context) (context.Context, cont
 
 func parseSessionKeyParts(sessionKey string) (platform, chatID string, err error) {
 	// Format: "platform:chatID:userID"
+	// Relay session format: "relay:sourceProject:chatID"
 	parts := strings.SplitN(sessionKey, ":", 3)
 	if len(parts) < 2 {
 		return "", "", fmt.Errorf("invalid session key format: %q", sessionKey)
 	}
+	if parts[0] == "relay" && len(parts) == 3 {
+		// For relay sessions, chatID is the third part: "relay:sourceProject:chatID"
+		return parts[0], parts[2], nil
+	}
 	return parts[0], parts[1], nil
 }
 

core/relay_test.go

@@ -3,6 +3,7 @@ package core
 import (
 	"context"
 	"errors"
+	"fmt"
 	"testing"
 	"time"
 )
@@ -67,7 +68,12 @@ func TestHandleRelay_ReturnsPartialOnTimeout(t *testing.T) {
 
 	session.events <- Event{Type: EventText, Content: "partial response", SessionID: "relay-session"}
 	time.Sleep(40 * time.Millisecond)
+	// After timeout, HandleRelay consumes the next event from the channel to
+	// unblock the for-range loop, then checks ctx.Err() and spawns the drain
+	// goroutine. We need two events: one to unblock HandleRelay, and one
+	// EventResult for the drain goroutine to close the session cleanly.
 	session.events <- Event{Type: EventThinking, Content: "still working"}
+	session.events <- Event{Type: EventResult, Content: "done"}
 
 	got := <-done
 	if got.err != nil {
@@ -76,6 +82,13 @@ func TestHandleRelay_ReturnsPartialOnTimeout(t *testing.T) {
 	if got.resp != "partial response" {
 		t.Fatalf("HandleRelay() response = %q, want %q", got.resp, "partial response")
 	}
+
+	// Wait for the background drain goroutine to close the session.
+	select {
+	case <-session.closed:
+	case <-time.After(2 * time.Second):
+		t.Fatal("background drain goroutine did not close the session")
+	}
 }
 
 func TestHandleRelay_TimeoutWithoutTextReturnsContextError(t *testing.T) {
@@ -97,7 +110,9 @@ func TestHandleRelay_TimeoutWithoutTextReturnsContextError(t *testing.T) {
 	}()
 
 	time.Sleep(40 * time.Millisecond)
+	// One event to unblock HandleRelay's for-range, one for the drain goroutine.
 	session.events <- Event{Type: EventThinking, Content: "still working"}
+	session.events <- Event{Type: EventResult, Content: "done"}
 
 	got := <-done
 	if got.resp != "" {
@@ -106,4 +121,69 @@ func TestHandleRelay_TimeoutWithoutTextReturnsContextError(t *testing.T) {
 	if !errors.Is(got.err, context.DeadlineExceeded) {
 		t.Fatalf("HandleRelay() error = %v, want context deadline exceeded", got.err)
 	}
+
+	select {
+	case <-session.closed:
+	case <-time.After(2 * time.Second):
+		t.Fatal("background drain goroutine did not close the session")
+	}
+}
+
+// relayFallbackAgent fails the first StartSession call (simulating a corrupt
+// resume) and returns freshSession on the second call (fresh start).
+type relayFallbackAgent struct {
+	callCount    int
+	freshSession AgentSession
+}
+
+func (a *relayFallbackAgent) Name() string { return "fallback" }
+func (a *relayFallbackAgent) StartSession(_ context.Context, sessionID string) (AgentSession, error) {
+	a.callCount++
+	if a.callCount == 1 && sessionID != "" {
+		return nil, fmt.Errorf("simulated resume failure")
+	}
+	return a.freshSession, nil
+}
+func (a *relayFallbackAgent) ListSessions(_ context.Context) ([]AgentSessionInfo, error) {
+	return nil, nil
+}
+func (a *relayFallbackAgent) Stop() error { return nil }
+
+func TestHandleRelay_ResumeFailureFallsBackToFreshSession(t *testing.T) {
+	e := newTestEngine()
+	freshSession := newControllableSession("fresh-session")
+
+	e.agent = &relayFallbackAgent{freshSession: freshSession}
+
+	// Pre-set a stale session ID so that the first StartSession tries to resume.
+	relaySessionKey := "relay:source:chat-1"
+	sess := e.sessions.GetOrCreateActive(relaySessionKey)
+	sess.SetAgentSessionID("stale-id", "fallback")
+	e.sessions.Save()
+
+	ctx := context.Background()
+	done := make(chan string, 1)
+	go func() {
+		resp, err := e.HandleRelay(ctx, "source", "chat-1", "hello")
+		if err != nil {
+			done <- "error: " + err.Error()
+			return
+		}
+		done <- resp
+	}()
+
+	// The fresh session should receive the message and respond.
+	freshSession.events <- Event{Type: EventResult, Content: "recovered", SessionID: "fresh-session"}
+
+	got := <-done
+	if got != "recovered" {
+		t.Fatalf("HandleRelay() = %q, want %q", got, "recovered")
+	}
+
+	// Session should be closed after EventResult.
+	select {
+	case <-freshSession.closed:
+	case <-time.After(2 * time.Second):
+		t.Fatal("session was not closed after EventResult")
+	}
 }

platform/feishu/feishu.go

@@ -115,7 +115,8 @@ type Platform struct {
 	self                  core.Platform
 	reactionEmoji         string
 	allowFrom             string
-	groupReplyAll         bool
+	groupReplyAll              bool
+	respondToAtEveryoneAndHere bool
 	shareSessionInChannel bool
 	threadIsolation       bool
 	// noReplyToTrigger: when true, send via Create instead of Im.Message.Reply (no quote to the user's message).
@@ -174,6 +175,7 @@ func newPlatform(name, domain string, opts map[string]any) (core.Platform, error
 	allowFrom, _ := opts["allow_from"].(string)
 	core.CheckAllowFrom(name, allowFrom)
 	groupReplyAll, _ := opts["group_reply_all"].(bool)
+	respondToAtEveryoneAndHere, _ := opts["respond_to_at_everyone_and_here"].(bool)
 	shareSessionInChannel, _ := opts["share_session_in_channel"].(bool)
 	threadIsolation, _ := opts["thread_isolation"].(bool)
 	noReplyToTrigger := false
@@ -222,7 +224,8 @@ func newPlatform(name, domain string, opts map[string]any) (core.Platform, error
 		useInteractiveCard:    useInteractiveCard,
 		reactionEmoji:         reactionEmoji,
 		allowFrom:             allowFrom,
-		groupReplyAll:         groupReplyAll,
+		groupReplyAll:              groupReplyAll,
+		respondToAtEveryoneAndHere: respondToAtEveryoneAndHere,
 		shareSessionInChannel: shareSessionInChannel,
 		threadIsolation:       threadIsolation,
 		noReplyToTrigger:      noReplyToTrigger,
@@ -672,8 +675,13 @@ func (p *Platform) onMessage(event *larkim.P2MessageReceiveV1) error {
 
 	if chatType == "group" && !p.groupReplyAll && p.botOpenID != "" {
 		if !isBotMentioned(msg.Mentions, p.botOpenID) {
-			slog.Debug(p.tag()+": ignoring group message without bot mention", "chat_id", chatID)
-			return nil
+			// Feishu @all sends {"text":"@_all"} with 0 mentions.
+			if p.respondToAtEveryoneAndHere && msg.Content != nil && strings.Contains(*msg.Content, "@_all") {
+				slog.Debug(p.tag()+": responding to @all message", "chat_id", chatID)
+			} else {
+				slog.Debug(p.tag()+": ignoring group message without bot mention", "chat_id", chatID)
+				return nil
+			}
 		}
 	}