diff --git a/internal/channels/discord/discord.go b/internal/channels/discord/discord.go index 87965757..626e8f17 100644 --- a/internal/channels/discord/discord.go +++ b/internal/channels/discord/discord.go @@ -148,8 +148,9 @@ func (c *Channel) Send(_ context.Context, msg bus.OutboundMessage) (err error) { // but keep it alive for the final response. Don't stop typing or cleanup. if msg.Metadata["placeholder_update"] == "true" { if pID, ok := c.placeholders.Load(placeholderKey); ok { - msgID := pID.(string) - _, _ = c.session.ChannelMessageEdit(channelID, msgID, msg.Content) + if msgID, ok := pID.(string); ok { + _, _ = c.session.ChannelMessageEdit(channelID, msgID, msg.Content) + } } return nil } @@ -166,7 +167,9 @@ func (c *Channel) Send(_ context.Context, msg bus.OutboundMessage) (err error) { // Delete placeholder if present if pID, ok := c.placeholders.Load(placeholderKey); ok { c.placeholders.Delete(placeholderKey) - _ = c.session.ChannelMessageDelete(channelID, pID.(string)) + if msgID, ok := pID.(string); ok { + _ = c.session.ChannelMessageDelete(channelID, msgID) + } } return c.sendMediaMessage(channelID, content, msg.Media) } @@ -176,8 +179,9 @@ func (c *Channel) Send(_ context.Context, msg bus.OutboundMessage) (err error) { if content == "" { if pID, ok := c.placeholders.Load(placeholderKey); ok { c.placeholders.Delete(placeholderKey) - msgID := pID.(string) - _ = c.session.ChannelMessageDelete(channelID, msgID) + if msgID, ok := pID.(string); ok { + _ = c.session.ChannelMessageDelete(channelID, msgID) + } } return nil } @@ -186,31 +190,31 @@ func (c *Channel) Send(_ context.Context, msg bus.OutboundMessage) (err error) { // then send the rest as follow-up messages. if pID, ok := c.placeholders.Load(placeholderKey); ok { c.placeholders.Delete(placeholderKey) - msgID := pID.(string) + if msgID, ok := pID.(string); ok { + const maxLen = 2000 + editContent := content + remaining := "" - const maxLen = 2000 - editContent := content - remaining := "" - - if len(editContent) > maxLen { - // Break at a newline if possible - cutAt := maxLen - if idx := lastIndexByte(content[:maxLen], '\n'); idx > maxLen/2 { - cutAt = idx + 1 + if len(editContent) > maxLen { + // Break at a newline if possible + cutAt := maxLen + if idx := lastIndexByte(content[:maxLen], '\n'); idx > maxLen/2 { + cutAt = idx + 1 + } + editContent = content[:cutAt] + remaining = content[cutAt:] } - editContent = content[:cutAt] - remaining = content[cutAt:] - } - if _, editErr := c.session.ChannelMessageEdit(channelID, msgID, editContent); editErr == nil { - // Send remaining content as follow-up messages - if remaining != "" { - return c.sendChunked(channelID, remaining) + if _, editErr := c.session.ChannelMessageEdit(channelID, msgID, editContent); editErr == nil { + // Send remaining content as follow-up messages + if remaining != "" { + return c.sendChunked(channelID, remaining) + } + return nil + } else { + slog.Warn("discord: placeholder edit failed, sending new message", + "channel_id", channelID, "placeholder_id", msgID, "error", editErr) } - return nil - } else { - slog.Warn("discord: placeholder edit failed, sending new message", - "channel_id", channelID, "placeholder_id", msgID, "error", editErr) } // Fall through to send new message if edit fails } diff --git a/internal/channels/feishu/feishu.go b/internal/channels/feishu/feishu.go index 68bc1d86..2c1e167e 100644 --- a/internal/channels/feishu/feishu.go +++ b/internal/channels/feishu/feishu.go @@ -20,6 +20,7 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/channels" "github.com/nextlevelbuilder/goclaw/internal/config" + "github.com/nextlevelbuilder/goclaw/internal/safego" "github.com/nextlevelbuilder/goclaw/internal/store" ) @@ -277,6 +278,7 @@ func (c *Channel) startWebSocket(ctx context.Context) error { c.wsClient = NewWSClient(c.cfg.AppID, c.cfg.AppSecret, domain, &wsEventAdapter{ch: c}) go func() { + defer safego.Recover(nil, "component", "feishu_ws", "channel", c.Name()) if err := c.wsClient.Start(ctx); err != nil { slog.Error("feishu websocket error", "error", err) } diff --git a/internal/channels/health.go b/internal/channels/health.go index fd0a31f1..0ad1d97c 100644 --- a/internal/channels/health.go +++ b/internal/channels/health.go @@ -1,6 +1,9 @@ package channels import ( + "context" + "errors" + "net" "strings" "time" ) @@ -86,7 +89,6 @@ type ChannelErrorInfo struct { // This is a best-effort classification based on error message patterns from upstream libraries // (telego, discordgo, etc.). If upstream changes error format, the default case returns // "unknown + retryable", so misclassification degrades gracefully to generic guidance. -// TODO: Use errors.As() where upstream libraries expose typed errors. func ClassifyChannelError(err error) ChannelErrorInfo { if err == nil { return ChannelErrorInfo{ @@ -97,6 +99,43 @@ func ClassifyChannelError(err error) ChannelErrorInfo { } } + // Prefer typed error checks over string matching where possible. + if errors.Is(err, context.DeadlineExceeded) { + return ChannelErrorInfo{ + Summary: "Network error", + Detail: "Timed out while reaching the upstream service.", + Kind: ChannelFailureKindNetwork, + Retryable: true, + } + } + var dnsErr *net.DNSError + if errors.As(err, &dnsErr) { + return ChannelErrorInfo{ + Summary: "Network error", + Detail: "GoClaw could not resolve the upstream host.", + Kind: ChannelFailureKindNetwork, + Retryable: !dnsErr.IsNotFound, + } + } + var opErr *net.OpError + if errors.As(err, &opErr) { + if opErr.Timeout() { + return ChannelErrorInfo{ + Summary: "Network error", + Detail: "Timed out while reaching the upstream service.", + Kind: ChannelFailureKindNetwork, + Retryable: true, + } + } + return ChannelErrorInfo{ + Summary: "Network error", + Detail: "GoClaw could not open a network connection to the upstream service.", + Kind: ChannelFailureKindNetwork, + Retryable: true, + } + } + + // Fall back to string matching for errors without typed wrappers. detail := err.Error() msg := strings.ToLower(detail) diff --git a/internal/channels/slack/channel.go b/internal/channels/slack/channel.go index 4360feb4..90a1081b 100644 --- a/internal/channels/slack/channel.go +++ b/internal/channels/slack/channel.go @@ -215,7 +215,7 @@ func (c *Channel) sweepMaps() { now := time.Now() c.dedup.Range(func(k, v any) bool { - if now.Sub(v.(time.Time)) > 5*time.Minute { + if t, ok := v.(time.Time); ok && now.Sub(t) > 5*time.Minute { c.dedup.Delete(k) } return true @@ -223,7 +223,7 @@ func (c *Channel) sweepMaps() { if c.threadTTL > 0 { c.threadParticip.Range(func(k, v any) bool { - if now.Sub(v.(time.Time)) > c.threadTTL { + if t, ok := v.(time.Time); ok && now.Sub(t) > c.threadTTL { c.threadParticip.Delete(k) } return true diff --git a/internal/http/agents_export_handlers.go b/internal/http/agents_export_handlers.go index 4a2c528c..24348b47 100644 --- a/internal/http/agents_export_handlers.go +++ b/internal/http/agents_export_handlers.go @@ -172,6 +172,7 @@ func (h *AgentsHandler) handleExportDirect(w http.ResponseWriter, r *http.Reques } // canExport checks if userID has permission to export the given agent. +// Export is restricted to agent owner and system owner by design. func (h *AgentsHandler) canExport(ag *store.AgentData, userID string) bool { if ag.OwnerID == userID { return true @@ -179,7 +180,6 @@ func (h *AgentsHandler) canExport(ag *store.AgentData, userID string) bool { if h.isOwnerUser(userID) { return true } - // TODO: tenant admin check return false } diff --git a/internal/store/pg/cron_scheduler.go b/internal/store/pg/cron_scheduler.go index e1c6e388..a4ca5c66 100644 --- a/internal/store/pg/cron_scheduler.go +++ b/internal/store/pg/cron_scheduler.go @@ -127,14 +127,21 @@ func (s *PGCronStore) recomputeStaleJobs() { next := computeNextRun(&schedule, now, s.defaultTZ) if next == nil { if scheduleKind == "at" { - s.db.ExecContext(s.baseCtx, "UPDATE cron_jobs SET enabled = false, updated_at = $1 WHERE id = $2", now, id) + if _, err := s.db.ExecContext(s.baseCtx, "UPDATE cron_jobs SET enabled = false, updated_at = $1 WHERE id = $2", now, id); err != nil { + slog.Warn("cron: failed to disable one-shot job", "id", id, "error", err) + } } continue } - s.db.ExecContext(s.baseCtx, "UPDATE cron_jobs SET next_run_at = $1, updated_at = $2 WHERE id = $3", *next, now, id) + if _, err := s.db.ExecContext(s.baseCtx, "UPDATE cron_jobs SET next_run_at = $1, updated_at = $2 WHERE id = $3", *next, now, id); err != nil { + slog.Warn("cron: failed to advance stale job", "id", id, "error", err) + } fixed++ } + if err := rows.Err(); err != nil { + slog.Warn("cron: recompute stale iteration error", "error", err) + } if fixed > 0 { slog.Info("cron: advanced stale/past-due jobs to next future run", "fixed", fixed) @@ -282,17 +289,21 @@ func (s *PGCronStore) executeOneJob(job store.CronJob, handler func(job *store.C if aid, aidErr := uuid.Parse(job.AgentID); aidErr == nil { agentUUID = &aid } - s.db.ExecContext(s.baseCtx, + if _, err := s.db.ExecContext(s.baseCtx, `INSERT INTO cron_run_logs (id, job_id, agent_id, status, error, summary, duration_ms, input_tokens, output_tokens, ran_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)`, logID, id, agentUUID, status, lastError, summary, durationMS, inputTokens, outputTokens, now, - ) + ); err != nil { + slog.Warn("cron: failed to insert run log", "job_id", job.ID, "error", err) + } } // Recompute next run or delete if job.DeleteAfterRun { if id, parseErr := uuid.Parse(job.ID); parseErr == nil { - s.db.ExecContext(s.baseCtx, "DELETE FROM cron_jobs WHERE id = $1", id) + if _, err := s.db.ExecContext(s.baseCtx, "DELETE FROM cron_jobs WHERE id = $1", id); err != nil { + slog.Warn("cron: failed to delete one-shot job", "job_id", job.ID, "error", err) + } } } else if id, parseErr := uuid.Parse(job.ID); parseErr == nil { schedule := job.Schedule @@ -318,13 +329,15 @@ func (s *PGCronStore) executeOneJob(job store.CronJob, handler func(job *store.C } } - s.db.ExecContext(s.baseCtx, + if _, err := s.db.ExecContext(s.baseCtx, `UPDATE cron_jobs SET last_run_at = $1, last_status = $2, last_error = $3, updated_at = $4, next_run_at = CASE WHEN enabled = true AND next_run_at IS NULL THEN $5 ELSE next_run_at END WHERE id = $6`, now, status, lastError, now, nextRunValue, id, - ) + ); err != nil { + slog.Warn("cron: failed to update job after run", "job_id", job.ID, "error", err) + } } // Emit completion event diff --git a/internal/store/sqlitestore/cron_scheduler.go b/internal/store/sqlitestore/cron_scheduler.go index 26302358..8641a5f3 100644 --- a/internal/store/sqlitestore/cron_scheduler.go +++ b/internal/store/sqlitestore/cron_scheduler.go @@ -130,12 +130,16 @@ func (s *SQLiteCronStore) recomputeStaleJobs() { next := computeNextRun(&schedule, now, s.defaultTZ) if next == nil { if scheduleKind == "at" { - s.db.ExecContext(s.baseCtx, "UPDATE cron_jobs SET enabled = 0, updated_at = ? WHERE id = ?", now, id) + if _, err := s.db.ExecContext(s.baseCtx, "UPDATE cron_jobs SET enabled = 0, updated_at = ? WHERE id = ?", now, id); err != nil { + slog.Warn("cron: failed to disable one-shot job", "id", id, "error", err) + } } continue } - s.db.ExecContext(s.baseCtx, "UPDATE cron_jobs SET next_run_at = ?, updated_at = ? WHERE id = ?", *next, now, id) + if _, err := s.db.ExecContext(s.baseCtx, "UPDATE cron_jobs SET next_run_at = ?, updated_at = ? WHERE id = ?", *next, now, id); err != nil { + slog.Warn("cron: failed to advance stale job", "id", id, "error", err) + } fixed++ } if err := rows.Err(); err != nil { @@ -282,16 +286,20 @@ func (s *SQLiteCronStore) executeOneJob(job store.CronJob, handler func(job *sto if aid, aidErr := uuid.Parse(job.AgentID); aidErr == nil { agentUUID = &aid } - s.db.ExecContext(s.baseCtx, + if _, err := s.db.ExecContext(s.baseCtx, `INSERT INTO cron_run_logs (id, job_id, agent_id, status, error, summary, duration_ms, input_tokens, output_tokens, ran_at) VALUES (?,?,?,?,?,?,?,?,?,?)`, logID, id, agentUUID, status, lastError, summary, durationMS, inputTokens, outputTokens, now, - ) + ); err != nil { + slog.Warn("cron: failed to insert run log", "job_id", job.ID, "error", err) + } } if job.DeleteAfterRun { if id, parseErr := uuid.Parse(job.ID); parseErr == nil { - s.db.ExecContext(s.baseCtx, "DELETE FROM cron_jobs WHERE id = ?", id) + if _, err := s.db.ExecContext(s.baseCtx, "DELETE FROM cron_jobs WHERE id = ?", id); err != nil { + slog.Warn("cron: failed to delete one-shot job", "job_id", job.ID, "error", err) + } } } else if id, parseErr := uuid.Parse(job.ID); parseErr == nil { schedule := job.Schedule @@ -315,13 +323,15 @@ func (s *SQLiteCronStore) executeOneJob(job store.CronJob, handler func(job *sto } } - s.db.ExecContext(s.baseCtx, + if _, err := s.db.ExecContext(s.baseCtx, `UPDATE cron_jobs SET last_run_at = ?, last_status = ?, last_error = ?, updated_at = ?, next_run_at = CASE WHEN enabled = 1 AND next_run_at IS NULL THEN ? ELSE next_run_at END WHERE id = ?`, now, status, lastError, now, nextRunValue, id, - ) + ); err != nil { + slog.Warn("cron: failed to update job after run", "job_id", job.ID, "error", err) + } } evt := store.CronEvent{Action: "completed", JobID: job.ID, JobName: job.Name, UserID: job.UserID, Status: status}