From d7342fbd6eaad207c3d3b1085d6421e434599218 Mon Sep 17 00:00:00 2001 From: viettranx Date: Fri, 17 Apr 2026 18:48:44 +0700 Subject: [PATCH] feat(agents): 3-layer recovery for agents stuck in summoning state Add ResetStuckSummoning store method (PG + SQLite) + startup sweep to flip ghost rows to summon_failed on boot. Drop 409 guards on resummon and regenerate to allow retry. Add POST /v1/agents/{id}/cancel-summon endpoint + audit event agent.summon_cancelled + i18n (en/vi/zh). --- cmd/gateway.go | 11 ++++++ internal/http/agents.go | 1 + internal/http/agents_sharing.go | 52 +++++++++++++++++++++++----- internal/i18n/catalog_en.go | 2 ++ internal/i18n/catalog_vi.go | 2 ++ internal/i18n/catalog_zh.go | 2 ++ internal/i18n/keys.go | 2 ++ internal/store/agent_store.go | 3 ++ internal/store/pg/agents.go | 12 +++++++ internal/store/sqlitestore/agents.go | 12 +++++++ 10 files changed, 91 insertions(+), 8 deletions(-) diff --git a/cmd/gateway.go b/cmd/gateway.go index 88904675..6c700af4 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -143,6 +143,17 @@ func runGateway() { } pgStores, traceCollector, snapshotWorker := setupStoresAndTracing(cfg, dataDir, msgBus) + + // Recover from crashes: flip ghost 'summoning' rows to 'summon_failed'. + // Summon goroutines don't survive process restart; stale DB rows would trap the UI. + if pgStores.Agents != nil { + if n, err := pgStores.Agents.ResetStuckSummoning(context.Background()); err != nil { + slog.Warn("agents.reset_stuck_summoning_failed", "err", err) + } else if n > 0 { + slog.Info("agents.reset_stuck_summoning", "count", n) + } + } + if traceCollector != nil { defer traceCollector.Stop() // OTel OTLP export: compiled via build tags. Build with 'go build -tags otel' to enable. diff --git a/internal/http/agents.go b/internal/http/agents.go index 8e96dc27..1e485e90 100644 --- a/internal/http/agents.go +++ b/internal/http/agents.go @@ -145,6 +145,7 @@ func (h *AgentsHandler) RegisterRoutes(mux *http.ServeMux) { // Agent operations (admin+) mux.HandleFunc("POST /v1/agents/{id}/regenerate", h.adminMiddleware(h.handleRegenerate)) mux.HandleFunc("POST /v1/agents/{id}/resummon", h.adminMiddleware(h.handleResummon)) + mux.HandleFunc("POST /v1/agents/{id}/cancel-summon", h.adminMiddleware(h.handleCancelSummon)) // Export (agent owner or system owner) mux.HandleFunc("GET /v1/agents/{id}/system-prompt-preview", h.adminMiddleware(h.handleSystemPromptPreview)) mux.HandleFunc("GET /v1/agents/{id}/export/preview", h.authMiddleware(h.handleExportPreview)) diff --git a/internal/http/agents_sharing.go b/internal/http/agents_sharing.go index f0b35d46..334df9cb 100644 --- a/internal/http/agents_sharing.go +++ b/internal/http/agents_sharing.go @@ -5,8 +5,10 @@ import ( "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/i18n" "github.com/nextlevelbuilder/goclaw/internal/store" + "github.com/nextlevelbuilder/goclaw/pkg/protocol" ) func (h *AgentsHandler) handleListShares(w http.ResponseWriter, r *http.Request) { @@ -139,10 +141,6 @@ func (h *AgentsHandler) handleRegenerate(w http.ResponseWriter, r *http.Request) writeJSON(w, http.StatusForbidden, map[string]string{"error": i18n.T(locale, i18n.MsgOwnerOnly, "regenerate agent")}) return } - if ag.Status == store.AgentStatusSummoning { - writeJSON(w, http.StatusConflict, map[string]string{"error": i18n.T(locale, i18n.MsgAlreadySummoning)}) - return - } if h.summoner == nil { writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": i18n.T(locale, i18n.MsgSummoningUnavailable)}) return @@ -191,10 +189,6 @@ func (h *AgentsHandler) handleResummon(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusForbidden, map[string]string{"error": i18n.T(locale, i18n.MsgOwnerOnly, "resummon agent")}) return } - if ag.Status == store.AgentStatusSummoning { - writeJSON(w, http.StatusConflict, map[string]string{"error": i18n.T(locale, i18n.MsgAlreadySummoning)}) - return - } if h.summoner == nil { writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": i18n.T(locale, i18n.MsgSummoningUnavailable)}) return @@ -217,4 +211,46 @@ func (h *AgentsHandler) handleResummon(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusAccepted, map[string]string{"ok": "true", "status": store.AgentStatusSummoning}) } +// handleCancelSummon force-transitions a stuck 'summoning' agent to 'summon_failed'. +// Used when user wants to abort a hanging summon (UI Cancel button after 60s). +func (h *AgentsHandler) handleCancelSummon(w http.ResponseWriter, r *http.Request) { + userID := store.UserIDFromContext(r.Context()) + locale := store.LocaleFromContext(r.Context()) + id, err := uuid.Parse(r.PathValue("id")) + if err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": i18n.T(locale, i18n.MsgInvalidID, "agent")}) + return + } + + ag, err := h.agents.GetByID(r.Context(), id) + if err != nil { + writeJSON(w, http.StatusNotFound, map[string]string{"error": i18n.T(locale, i18n.MsgNotFound, "agent", id.String())}) + return + } + if userID != "" && ag.OwnerID != userID && !h.isOwnerUser(userID) { + writeJSON(w, http.StatusForbidden, map[string]string{"error": i18n.T(locale, i18n.MsgOwnerOnly, "cancel summon")}) + return + } + if ag.Status != store.AgentStatusSummoning { + writeJSON(w, http.StatusConflict, map[string]string{"error": i18n.T(locale, i18n.MsgCannotCancel)}) + return + } + + if err := h.agents.Update(r.Context(), id, map[string]any{"status": store.AgentStatusSummonFailed}); err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()}) + return + } + + if h.msgBus != nil { + bus.BroadcastForTenant(h.msgBus, protocol.EventAgentSummoning, store.TenantIDFromContext(r.Context()), map[string]any{ + "agent_id": id.String(), + "type": "failed", + "error": i18n.T(locale, i18n.MsgSummonCancelled), + }) + } + + emitAudit(h.msgBus, r, "agent.summon_cancelled", "agent", id.String()) + writeJSON(w, http.StatusAccepted, map[string]string{"ok": "true", "status": store.AgentStatusSummonFailed}) +} + // writeJSON moved to response_helpers.go diff --git a/internal/i18n/catalog_en.go b/internal/i18n/catalog_en.go index e25c8e4a..ec4b7944 100644 --- a/internal/i18n/catalog_en.go +++ b/internal/i18n/catalog_en.go @@ -73,6 +73,8 @@ func init() { MsgAlreadySummoning: "agent is already being summoned", MsgSummoningUnavailable: "summoning not available", MsgNoDescription: "agent has no description to resummon from", + MsgSummonCancelled: "summon cancelled by user", + MsgCannotCancel: "agent is not being summoned", MsgInvalidPath: "invalid path", // Tenant backup / restore diff --git a/internal/i18n/catalog_vi.go b/internal/i18n/catalog_vi.go index a3f795af..8ddd0527 100644 --- a/internal/i18n/catalog_vi.go +++ b/internal/i18n/catalog_vi.go @@ -73,6 +73,8 @@ func init() { MsgAlreadySummoning: "agent đang được triệu hồi", MsgSummoningUnavailable: "triệu hồi không khả dụng", MsgNoDescription: "agent không có mô tả để triệu hồi lại", + MsgSummonCancelled: "đã huỷ triệu hồi", + MsgCannotCancel: "agent không trong trạng thái đang triệu hồi", MsgInvalidPath: "đường dẫn không hợp lệ", // Tenant backup / restore diff --git a/internal/i18n/catalog_zh.go b/internal/i18n/catalog_zh.go index 9ade79c8..7ccf1d00 100644 --- a/internal/i18n/catalog_zh.go +++ b/internal/i18n/catalog_zh.go @@ -73,6 +73,8 @@ func init() { MsgAlreadySummoning: "Agent正在被召唤中", MsgSummoningUnavailable: "召唤功能不可用", MsgNoDescription: "Agent没有可供重新召唤的描述", + MsgSummonCancelled: "已取消召唤", + MsgCannotCancel: "Agent 未处于召唤状态", MsgInvalidPath: "路径无效", // Tenant backup / restore diff --git a/internal/i18n/keys.go b/internal/i18n/keys.go index 5c6c7a67..2c1e308c 100644 --- a/internal/i18n/keys.go +++ b/internal/i18n/keys.go @@ -74,6 +74,8 @@ const ( MsgAlreadySummoning = "error.already_summoning" // "agent is already being summoned" MsgSummoningUnavailable = "error.summoning_unavailable" // "summoning not available" MsgNoDescription = "error.no_description" // "agent has no description to resummon from" + MsgSummonCancelled = "info.summon_cancelled" // "summon cancelled by user" + MsgCannotCancel = "error.cannot_cancel_summon" // "agent is not being summoned" MsgInvalidPath = "error.invalid_path" // "invalid path" // --- Tenant backup / restore --- diff --git a/internal/store/agent_store.go b/internal/store/agent_store.go index edefd911..64c2c30a 100644 --- a/internal/store/agent_store.go +++ b/internal/store/agent_store.go @@ -608,6 +608,9 @@ type AgentCRUDStore interface { Delete(ctx context.Context, id uuid.UUID) error List(ctx context.Context, ownerID string) ([]AgentData, error) GetDefault(ctx context.Context) (*AgentData, error) // agent with is_default=true, or first available + // ResetStuckSummoning flips rows with status='summoning' to 'summon_failed'. + // Called at startup to recover from crashes where summon goroutine died mid-flight. + ResetStuckSummoning(ctx context.Context) (int64, error) } // AgentAccessStore manages agent sharing and access control. diff --git a/internal/store/pg/agents.go b/internal/store/pg/agents.go index 1074a7fd..25aff349 100644 --- a/internal/store/pg/agents.go +++ b/internal/store/pg/agents.go @@ -609,6 +609,18 @@ func joinStrings(s []string, sep string) string { return result.String() } +// ResetStuckSummoning flips rows with status='summoning' to 'summon_failed'. +// Called at startup to recover from crashes where summon goroutines died mid-flight. +func (s *PGAgentStore) ResetStuckSummoning(ctx context.Context) (int64, error) { + const q = `UPDATE agents SET status = $1 WHERE status = $2` + res, err := s.db.ExecContext(ctx, q, store.AgentStatusSummonFailed, store.AgentStatusSummoning) + if err != nil { + return 0, err + } + n, _ := res.RowsAffected() + return n, nil +} + func replaceIDX(s, replacement string) string { var result strings.Builder for i := 0; i < len(s); i++ { diff --git a/internal/store/sqlitestore/agents.go b/internal/store/sqlitestore/agents.go index 9e12e307..45f7ff80 100644 --- a/internal/store/sqlitestore/agents.go +++ b/internal/store/sqlitestore/agents.go @@ -327,3 +327,15 @@ func scanAgentRows(rows *sql.Rows) ([]store.AgentData, error) { } return result, rows.Err() } + +// ResetStuckSummoning flips rows with status='summoning' to 'summon_failed'. +// Called at startup to recover from crashes where summon goroutines died mid-flight. +func (s *SQLiteAgentStore) ResetStuckSummoning(ctx context.Context) (int64, error) { + const q = `UPDATE agents SET status = ? WHERE status = ?` + res, err := s.db.ExecContext(ctx, q, store.AgentStatusSummonFailed, store.AgentStatusSummoning) + if err != nil { + return 0, err + } + n, _ := res.RowsAffected() + return n, nil +}