diff --git a/core/engine.go b/core/engine.go index e5c4846..476ab75 100644 --- a/core/engine.go +++ b/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: diff --git a/core/relay.go b/core/relay.go index 9aaa3c3..b16aaf0 100644 --- a/core/relay.go +++ b/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 } diff --git a/core/relay_test.go b/core/relay_test.go index 12ed483..15f3ebf 100644 --- a/core/relay_test.go +++ b/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") + } } diff --git a/platform/feishu/feishu.go b/platform/feishu/feishu.go index 9ae8c2f..f684788 100644 --- a/platform/feishu/feishu.go +++ b/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 + } } }