From 215d291cfa642e2424b74c9c562e5da0d02a1c7e Mon Sep 17 00:00:00 2001 From: Haruko386 Date: Wed, 15 Jul 2026 10:56:36 +0800 Subject: [PATCH] fix: support email send for go backend (#16895) ### Summary As title image d2aeeb50dd11cc8dc76bc4743425680e --- internal/agent/canvas/scheduler.go | 1 + internal/agent/component/email.go | 251 +++++++++++++++++++++++++ internal/agent/component/email_test.go | 223 ++++++++++++++++++++++ internal/agent/runtime/state.go | 23 ++- internal/agent/runtime/state_test.go | 21 +++ internal/agent/tool/email.go | 155 +++++++++++++-- internal/service/agent.go | 1 + 7 files changed, 659 insertions(+), 16 deletions(-) create mode 100644 internal/agent/component/email.go create mode 100644 internal/agent/component/email_test.go diff --git a/internal/agent/canvas/scheduler.go b/internal/agent/canvas/scheduler.go index 1e4f40fe8b..67dff27e4c 100644 --- a/internal/agent/canvas/scheduler.go +++ b/internal/agent/canvas/scheduler.go @@ -391,6 +391,7 @@ func BuildWorkflow(ctx context.Context, c *Canvas) (*compose.Workflow[map[string } } } + st.EnsureSysDate() return st } diff --git a/internal/agent/component/email.go b/internal/agent/component/email.go new file mode 100644 index 0000000000..0e36395a32 --- /dev/null +++ b/internal/agent/component/email.go @@ -0,0 +1,251 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +package component + +import ( + "context" + "encoding/json" + "fmt" + "ragflow/internal/agent/runtime" + "strconv" + "strings" + + agenttool "ragflow/internal/agent/tool" +) + +const componentNameEmail = "Email" + +// EmailComponent sends a canvas email node through the Go SMTP tool. +// It accepts the same DSL fields as the Python Email tool: +// smtp_server, smtp_port, email, smtp_username, password, sender_name, +// to_email, cc_email, content, and subject. +type EmailComponent struct { + name string + smtpServer string + smtpPort int + email string + smtpUsername string + password string + senderName string + toEmail string + ccEmail string + content string + subject string + tool *agenttool.EmailTool +} + +// NewEmailComponent constructs an Email component from DSL params. +func NewEmailComponent(params map[string]any) (Component, error) { + smtpPort, err := emailIntParam(params, "smtp_port", 465) + if err != nil { + return nil, err + } + return &EmailComponent{ + name: componentNameEmail, + smtpServer: emailStringParam(params, "smtp_server"), + smtpPort: smtpPort, + email: emailStringParam(params, "email"), + smtpUsername: emailStringParam(params, "smtp_username"), + password: emailStringParam(params, "password"), + senderName: emailStringParam(params, "sender_name"), + toEmail: emailStringParam(params, "to_email"), + ccEmail: emailStringParam(params, "cc_email"), + content: emailStringParam(params, "content"), + subject: emailStringParam(params, "subject"), + tool: agenttool.NewEmailTool(), + }, nil +} + +// Name returns the registered component name. +func (e *EmailComponent) Name() string { return e.name } + +// Invoke sends the email and returns success/_ERROR fields, matching the +// Python node's soft-failure behaviour. +func (e *EmailComponent) Invoke(ctx context.Context, inputs map[string]any) (map[string]any, error) { + toEmail := firstString(inputs, "to_email", e.toEmail) + content := firstString(inputs, "content", e.content) + subject := firstString(inputs, "subject", e.subject) + ccEmail := firstString(inputs, "cc_email", e.ccEmail) + + state, _, _ := runtime.GetStateFromContext[*runtime.CanvasState](ctx) + toEmail = runtime.ResolveTemplateForDisplay(toEmail, state) + ccEmail = runtime.ResolveTemplateForDisplay(ccEmail, state) + subject = runtime.ResolveTemplateForDisplay(subject, state) + subject = stripEmailHeaderLineBreaks(subject) + content = runtime.ResolveTemplateForDisplay(content, state) + + args := map[string]any{ + "smtp_host": e.smtpServer, + "smtp_port": e.smtpPort, + "username": firstNonEmpty(e.smtpUsername, e.email), + "password": e.password, + "from_addr": e.email, + "to_addrs": emailRecipients(toEmail, ccEmail), + "subject": subject, + "body": content, + } + argsJSON, err := json.Marshal(args) + if err != nil { + return map[string]any{"success": false, "_ERROR": fmt.Sprintf("email: marshal arguments: %v", err)}, nil + } + + out, err := e.tool.InvokableRun(ctx, string(argsJSON)) + if err != nil { + return map[string]any{"success": false, "_ERROR": err.Error()}, nil + } + var env struct { + OK bool `json:"ok"` + Error string `json:"_ERROR"` + } + if err := json.Unmarshal([]byte(out), &env); err != nil { + return map[string]any{"success": false, "_ERROR": fmt.Sprintf("email: parse result: %v", err)}, nil + } + if env.Error != "" { + return map[string]any{"success": false, "_ERROR": env.Error}, nil + } + return map[string]any{"success": env.OK}, nil +} + +// Stream mirrors Invoke as a single-chunk stream. +func (e *EmailComponent) Stream(ctx context.Context, inputs map[string]any) (<-chan map[string]any, error) { + out, err := e.Invoke(ctx, inputs) + if err != nil { + return nil, err + } + ch := make(chan map[string]any, 1) + ch <- out + close(ch) + return ch, nil +} + +func (e *EmailComponent) GetInputForm() map[string]any { + return map[string]any{ + "to_email": map[string]any{ + "name": "To", + "type": "line", + }, + "subject": map[string]any{ + "name": "Subject", + "type": "line", + }, + "cc_email": map[string]any{ + "name": "CC To", + "type": "line", + }, + } +} + +// Inputs returns the DSL input surface. +func (e *EmailComponent) Inputs() map[string]string { + return map[string]string{ + "to_email": "Recipient email address list.", + "cc_email": "Optional CC recipient list.", + "content": "Email body.", + "subject": "Email subject.", + "smtp_server": "SMTP server hostname.", + "smtp_port": "SMTP server port.", + "email": "Sender email address.", + "smtp_username": "SMTP username; defaults to sender email.", + "password": "SMTP password or authorization code.", + "sender_name": "Sender display name.", + } +} + +// Outputs returns the public output surface. +func (e *EmailComponent) Outputs() map[string]string { + return map[string]string{ + "success": "Whether the email was sent successfully.", + "_ERROR": "SMTP error message when sending fails.", + } +} + +func emailStringParam(params map[string]any, key string) string { + if v, ok := params[key].(string); ok { + return v + } + return "" +} + +func emailIntParam(params map[string]any, key string, fallback int) (int, error) { + switch v := params[key].(type) { + case nil: + return fallback, nil + case int: + return v, nil + case int64: + return int(v), nil + case float64: + return int(v), nil + case json.Number: + n, err := v.Int64() + return int(n), err + case string: + if strings.TrimSpace(v) == "" { + return fallback, nil + } + n, err := strconv.Atoi(strings.TrimSpace(v)) + if err != nil { + return 0, &ParamError{Field: key, Reason: "must be an integer"} + } + return n, nil + default: + return 0, &ParamError{Field: key, Reason: "must be an integer"} + } +} + +func firstString(inputs map[string]any, key, fallback string) string { + if v, ok := inputs[key].(string); ok { + return v + } + return fallback +} + +func firstNonEmpty(values ...string) string { + for _, v := range values { + if strings.TrimSpace(v) != "" { + return v + } + } + return "" +} + +func emailRecipients(toEmail, ccEmail string) []string { + recipients := splitEmailList(toEmail) + recipients = append(recipients, splitEmailList(ccEmail)...) + return recipients +} + +func splitEmailList(value string) []string { + parts := strings.FieldsFunc(value, func(r rune) bool { + return r == ',' || r == ';' || r == '\n' || r == '\r' + }) + out := make([]string, 0, len(parts)) + for _, part := range parts { + if s := strings.TrimSpace(part); s != "" { + out = append(out, s) + } + } + return out +} + +func stripEmailHeaderLineBreaks(value string) string { + return strings.NewReplacer("\r", "", "\n", "").Replace(value) +} + +func init() { + Register(componentNameEmail, NewEmailComponent) +} diff --git a/internal/agent/component/email_test.go b/internal/agent/component/email_test.go new file mode 100644 index 0000000000..a81a04a384 --- /dev/null +++ b/internal/agent/component/email_test.go @@ -0,0 +1,223 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +package component + +import ( + "bufio" + "context" + "fmt" + "net" + "strings" + "testing" + "time" + + "ragflow/internal/agent/runtime" +) + +func TestEmailComponentRegisteredAndSends(t *testing.T) { + t.Parallel() + + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("Listen: %v", err) + } + defer ln.Close() + + var receivedData strings.Builder + done := make(chan struct{}) + go runEmailMockSMTP(t, ln, &receivedData, done) + + _, port, err := net.SplitHostPort(ln.Addr().String()) + if err != nil { + t.Fatalf("SplitHostPort: %v", err) + } + var portInt int + _, _ = fmt.Sscanf(port, "%d", &portInt) + + c, err := New(componentNameEmail, map[string]any{ + "smtp_server": "127.0.0.1", + "smtp_port": portInt, + "email": "alice@example.com", + "smtp_username": "", + "password": "", + "sender_name": "Alice", + "to_email": "bob@example.com", + "subject": "Build status", + "content": "Build succeeded.", + }) + if err != nil { + t.Fatalf("New Email: %v", err) + } + + state := runtime.NewCanvasState("run-email", "task-email") + state.Sys["date"] = "2026-07-14 03:04:05" + ctx := runtime.WithState(context.Background(), state) + + out, err := c.Invoke(ctx, nil) + if err != nil { + t.Fatalf("Invoke: %v", err) + } + if out["success"] != true { + t.Fatalf("success = %v, want true; out=%v", out["success"], out) + } + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("mock SMTP server did not close in time") + } + + data := receivedData.String() + for _, want := range []string{"Subject: Build status", "bob@example.com", "Build succeeded."} { + if !strings.Contains(data, want) { + t.Fatalf("mock SMTP payload missing %q\n--- data ---\n%s\n---", want, data) + } + } +} + +func TestEmailComponentResolvesSysDateInSubject(t *testing.T) { + t.Parallel() + + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("Listen: %v", err) + } + defer ln.Close() + + var receivedData strings.Builder + done := make(chan struct{}) + go runEmailMockSMTP(t, ln, &receivedData, done) + + _, port, err := net.SplitHostPort(ln.Addr().String()) + if err != nil { + t.Fatalf("SplitHostPort: %v", err) + } + var portInt int + _, _ = fmt.Sscanf(port, "%d", &portInt) + + c, err := New(componentNameEmail, map[string]any{ + "smtp_server": "127.0.0.1", + "smtp_port": portInt, + "email": "alice@example.com", + "to_email": "bob@example.com", + "subject": "[noreply]{sys.date}", + "content": "body", + }) + if err != nil { + t.Fatalf("New Email: %v", err) + } + + state := runtime.NewCanvasState("run-email", "task-email") + state.Sys["date"] = "2026-07-14 03:04:05" + out, err := c.Invoke(runtime.WithState(context.Background(), state), nil) + if err != nil { + t.Fatalf("Invoke: %v", err) + } + if out["success"] != true { + t.Fatalf("success = %v, want true; out=%v", out["success"], out) + } + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("mock SMTP server did not close in time") + } + + if strings.Contains(receivedData.String(), "{sys.date}") { + t.Fatalf("subject still contains unresolved sys.date\n--- data ---\n%s\n---", receivedData.String()) + } + if !strings.Contains(receivedData.String(), "Subject: [noreply]2026-07-14 03:04:05") { + t.Fatalf("subject did not resolve sys.date\n--- data ---\n%s\n---", receivedData.String()) + } +} + +func TestEmailComponentSoftFails(t *testing.T) { + t.Parallel() + + c, err := New(componentNameEmail, map[string]any{ + "smtp_port": 465, + "email": "alice@example.com", + "to_email": "bob@example.com", + "subject": "Subject", + }) + if err != nil { + t.Fatalf("New Email: %v", err) + } + out, err := c.Invoke(context.Background(), nil) + if err != nil { + t.Fatalf("Invoke returned hard error: %v", err) + } + if out["success"] != false { + t.Fatalf("success = %v, want false; out=%v", out["success"], out) + } + if _, ok := out["_ERROR"].(string); !ok { + t.Fatalf("_ERROR missing or not string: %v", out) + } +} + +func runEmailMockSMTP(t *testing.T, ln net.Listener, receivedData *strings.Builder, done chan<- struct{}) { + t.Helper() + defer close(done) + + conn, err := ln.Accept() + if err != nil { + return + } + defer conn.Close() + _ = conn.SetReadDeadline(time.Now().Add(2 * time.Second)) + reader := bufio.NewReader(conn) + writer := bufio.NewWriter(conn) + _, _ = writer.WriteString("220 mock-smtp ready\r\n") + _ = writer.Flush() + + inData := false + for { + line, err := reader.ReadString('\n') + if err != nil { + return + } + up := strings.ToUpper(strings.TrimSpace(line)) + switch { + case strings.HasPrefix(up, "EHLO"), strings.HasPrefix(up, "HELO"): + _, _ = writer.WriteString("250-mock-smtp\r\n250-AUTH PLAIN\r\n250 OK\r\n") + _ = writer.Flush() + case strings.HasPrefix(up, "AUTH PLAIN"): + _, _ = writer.WriteString("235 Authentication successful\r\n") + _ = writer.Flush() + case strings.HasPrefix(up, "MAIL FROM:"), strings.HasPrefix(up, "RCPT TO:"): + _, _ = writer.WriteString("250 OK\r\n") + _ = writer.Flush() + case strings.HasPrefix(up, "DATA"): + _, _ = writer.WriteString("354 End data with .\r\n") + _ = writer.Flush() + inData = true + case inData && strings.TrimSpace(line) == ".": + _, _ = writer.WriteString("250 Queued\r\n") + _ = writer.Flush() + inData = false + case inData: + receivedData.WriteString(line) + case strings.HasPrefix(up, "QUIT"): + _, _ = writer.WriteString("221 Bye\r\n") + _ = writer.Flush() + return + default: + _, _ = writer.WriteString("250 OK\r\n") + _ = writer.Flush() + } + } +} diff --git a/internal/agent/runtime/state.go b/internal/agent/runtime/state.go index 7d1ee9d977..e7cc19b9f6 100644 --- a/internal/agent/runtime/state.go +++ b/internal/agent/runtime/state.go @@ -37,6 +37,7 @@ import ( "strings" "sync" "sync/atomic" + "time" "github.com/cloudwego/eino/compose" ) @@ -72,7 +73,7 @@ type CanvasState struct { // The atomic CancelFlag is allocated eagerly so nodes can safely poll it // even before any cancel signal has been wired. func NewCanvasState(runID, taskID string) *CanvasState { - return &CanvasState{ + s := &CanvasState{ Outputs: make(map[string]map[string]any), Sys: make(map[string]any), Env: make(map[string]any), @@ -84,6 +85,26 @@ func NewCanvasState(runID, taskID string) *CanvasState { RunID: runID, TaskID: taskID, } + s.EnsureSysDate() + return s +} + +// EnsureSysDate fills sys.date with the current UTC timestamp when it +// is missing or blank. Python canvas initializes the same variable with +// "%Y-%m-%d %H:%M:%S"; keep that wire format for DSL compatibility. +func (s *CanvasState) EnsureSysDate() { + if s == nil { + return + } + s.mu.Lock() + defer s.mu.Unlock() + if s.Sys == nil { + s.Sys = make(map[string]any) + } + if v, ok := s.Sys["date"]; ok && strings.TrimSpace(fmt.Sprint(v)) != "" { + return + } + s.Sys["date"] = time.Now().UTC().Format("2006-01-02 15:04:05") } // init registers CanvasState with eino's internal type registry so diff --git a/internal/agent/runtime/state_test.go b/internal/agent/runtime/state_test.go index ed8e8f3b0d..43d22212c9 100644 --- a/internal/agent/runtime/state_test.go +++ b/internal/agent/runtime/state_test.go @@ -85,6 +85,27 @@ func TestCanvasState_MarshalJSON_DoesNotLeakMutex(t *testing.T) { } } +func TestCanvasState_EnsureSysDate(t *testing.T) { + t.Parallel() + + state := NewCanvasState("r", "t") + if got, _ := state.Sys["date"].(string); got == "" { + t.Fatal("NewCanvasState did not initialize sys.date") + } + + state.Sys["date"] = "" + state.EnsureSysDate() + if got, _ := state.Sys["date"].(string); got == "" { + t.Fatal("EnsureSysDate left blank sys.date unchanged") + } + + state.Sys["date"] = "custom-date" + state.EnsureSysDate() + if got := state.Sys["date"]; got != "custom-date" { + t.Fatalf("EnsureSysDate overwrote non-empty sys.date: %v", got) + } +} + func TestCanvasState_SetRetrievalReferencesMergesCalls(t *testing.T) { t.Parallel() diff --git a/internal/agent/tool/email.go b/internal/agent/tool/email.go index c9db83a8ac..f7b19a93ca 100644 --- a/internal/agent/tool/email.go +++ b/internal/agent/tool/email.go @@ -18,10 +18,14 @@ package tool import ( "context" + "crypto/tls" "encoding/json" "fmt" + "io" + "net" "net/smtp" "strings" + "time" "github.com/cloudwego/eino/components/tool" "github.com/cloudwego/eino/schema" @@ -31,6 +35,11 @@ const emailToolName = "email" const emailToolDescription = "Send an email via SMTP. Returns success/failure status." +const ( + emailDialTimeout = 10 * time.Second + emailSessionTimeout = 30 * time.Second +) + // emailParams is the JSON shape the model sends into InvokableRun. type emailParams struct { SMTPHost string `json:"smtp_host"` @@ -119,7 +128,7 @@ func buildEmailMessage(from string, to []string, subject, body string) []byte { var b strings.Builder b.WriteString("From: " + from + "\r\n") b.WriteString("To: " + strings.Join(to, ", ") + "\r\n") - b.WriteString("Subject: " + subject + "\r\n") + b.WriteString("Subject: " + stripEmailHeaderLineBreaks(subject) + "\r\n") b.WriteString("MIME-Version: 1.0\r\n") b.WriteString("Content-Type: text/plain; charset=UTF-8\r\n") b.WriteString("\r\n") @@ -128,9 +137,15 @@ func buildEmailMessage(from string, to []string, subject, body string) []byte { return []byte(b.String()) } -// InvokableRun sends the email. We delegate to smtp.SendMail which -// handles EHLO, STARTTLS, and AUTH transparently when an *smtp.Auth is -// supplied; with nil auth it sends unauthenticated. +func stripEmailHeaderLineBreaks(value string) string { + return strings.NewReplacer("\r", "", "\n", "").Replace(value) +} + +type emailSender func(ctx context.Context, p emailParams, msg []byte) error + +var sendEmail = sendEmailSMTP + +// InvokableRun sends the email. func (e *EmailTool) InvokableRun(ctx context.Context, argsJSON string, _ ...tool.Option) (string, error) { var p emailParams if err := json.Unmarshal([]byte(argsJSON), &p); err != nil { @@ -141,28 +156,138 @@ func (e *EmailTool) InvokableRun(ctx context.Context, argsJSON string, _ ...tool return emailErrJSON(err), err } - addr := fmt.Sprintf("%s:%d", p.SMTPHost, p.SMTPPort) msg := buildEmailMessage(p.FromAddr, p.ToAddrs, p.Subject, p.Body) - - var auth smtp.Auth - if p.Username != "" { - auth = smtp.PlainAuth("", p.Username, p.Password, p.SMTPHost) - } - - if err := smtp.SendMail(addr, auth, p.FromAddr, p.ToAddrs, msg); err != nil { + if err := sendEmail(ctx, p, msg); err != nil { return emailErrJSON(fmt.Errorf("email: send: %w", err)), fmt.Errorf("email: send: %w", err) } - // Honor context cancellation if the caller passed a deadline. The - // underlying smtp.SendMail is blocking, so we check after the call; - // a stricter impl would select on ctx.Done() around the call. if err := ctx.Err(); err != nil { return emailErrJSON(err), err } return emailJSON(emailEnvelope{OK: true}), nil } +func sendEmailSMTP(ctx context.Context, p emailParams, msg []byte) error { + if p.SMTPPort == 465 { + return sendEmailSMTPS(ctx, p, msg) + } + return sendEmailSTARTTLS(ctx, p, msg) +} + +func sendEmailSTARTTLS(ctx context.Context, p emailParams, msg []byte) error { + addr := fmt.Sprintf("%s:%d", p.SMTPHost, p.SMTPPort) + dialer := &net.Dialer{Timeout: emailDialTimeout} + conn, err := dialer.DialContext(ctx, "tcp", addr) + if err != nil { + return err + } + defer conn.Close() + setEmailDeadline(ctx, conn) + stopWatch := watchEmailContext(ctx, conn) + defer stopWatch() + + client, err := smtp.NewClient(conn, p.SMTPHost) + if err != nil { + return err + } + defer client.Close() + + if err := client.Hello("localhost"); err != nil { + return err + } + if ok, _ := client.Extension("STARTTLS"); ok { + if err := client.StartTLS(&tls.Config{ServerName: p.SMTPHost, MinVersion: tls.VersionTLS12}); err != nil { + return err + } + } + return submitEmail(ctx, client, p, msg) +} + +func sendEmailSMTPS(ctx context.Context, p emailParams, msg []byte) error { + addr := fmt.Sprintf("%s:%d", p.SMTPHost, p.SMTPPort) + dialer := &net.Dialer{Timeout: emailDialTimeout} + rawConn, err := dialer.DialContext(ctx, "tcp", addr) + if err != nil { + return err + } + setEmailDeadline(ctx, rawConn) + stopWatch := watchEmailContext(ctx, rawConn) + defer stopWatch() + + conn := tls.Client(rawConn, &tls.Config{ServerName: p.SMTPHost, MinVersion: tls.VersionTLS12}) + if err := conn.HandshakeContext(ctx); err != nil { + _ = rawConn.Close() + return err + } + defer conn.Close() + + client, err := smtp.NewClient(conn, p.SMTPHost) + if err != nil { + return err + } + defer client.Close() + + if err := client.Hello("localhost"); err != nil { + return err + } + return submitEmail(ctx, client, p, msg) +} + +func submitEmail(ctx context.Context, client *smtp.Client, p emailParams, msg []byte) error { + if p.Username != "" { + if err := client.Auth(smtp.PlainAuth("", p.Username, p.Password, p.SMTPHost)); err != nil { + return err + } + } + if err := client.Mail(p.FromAddr); err != nil { + return err + } + for _, addr := range p.ToAddrs { + if err := client.Rcpt(addr); err != nil { + return err + } + } + wc, err := client.Data() + if err != nil { + return err + } + if _, err := wc.Write(msg); err != nil { + _ = wc.Close() + return err + } + if err := wc.Close(); err != nil { + return err + } + if err := ctx.Err(); err != nil { + return err + } + if err := client.Quit(); err != nil && err != io.EOF { + return err + } + return nil +} + +func setEmailDeadline(ctx context.Context, conn net.Conn) { + deadline := time.Now().Add(emailSessionTimeout) + if ctxDeadline, ok := ctx.Deadline(); ok && ctxDeadline.Before(deadline) { + deadline = ctxDeadline + } + _ = conn.SetDeadline(deadline) +} + +func watchEmailContext(ctx context.Context, conn net.Conn) func() { + done := make(chan struct{}) + go func() { + select { + case <-ctx.Done(): + _ = conn.SetDeadline(time.Now()) + case <-done: + } + }() + return func() { close(done) } +} + // validateEmailParams guards against obviously broken inputs. The // upstream SMTP server will give a better error for malformed // addresses, but the common case (empty / missing) is caught here. diff --git a/internal/service/agent.go b/internal/service/agent.go index 03174f4be6..14604def2f 100644 --- a/internal/service/agent.go +++ b/internal/service/agent.go @@ -1016,6 +1016,7 @@ func (s *AgentService) buildRunFunc(canvasID string, versionRow *entity.UserCanv } } } + state.EnsureSysDate() state.Sys["query"] = userInput if uid, ok := root["user_id"].(string); ok && uid != "" { state.Sys["user_id"] = uid