diff --git a/server/internal/handler/runtime.go b/server/internal/handler/runtime.go index 6e569d523a..7023fabdf4 100644 --- a/server/internal/handler/runtime.go +++ b/server/internal/handler/runtime.go @@ -760,6 +760,14 @@ func (h *Handler) DeleteAgentRuntime(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusInternalServerError, "failed to clean up agent invocation targets") return } + // Same app-layer cleanup for channel installations: channel_* has no + // workspace/agent FK (MUL-3515 §4), so an archived agent's bot installations + // would otherwise survive the hard-delete as orphans and keep occupying their + // (channel_type, app_id) routing slots, making those bots un-rebindable (#4810). + if err := qtx.DeleteChannelInstallationsByArchivedRuntimeAgents(r.Context(), rt.ID); err != nil { + writeError(w, http.StatusInternalServerError, "failed to clean up channel installations") + return + } if err := qtx.DeleteArchivedAgentsByRuntime(r.Context(), rt.ID); err != nil { writeError(w, http.StatusInternalServerError, "failed to clean up archived agents") return @@ -1004,6 +1012,14 @@ func (h *Handler) ArchiveAgentsAndDeleteRuntime(w http.ResponseWriter, r *http.R writeError(w, http.StatusInternalServerError, "failed to clean up agent invocation targets") return } + // Same app-layer cleanup for channel installations: channel_* has no + // workspace/agent FK (MUL-3515 §4), so an archived agent's bot installations + // would otherwise survive the hard-delete as orphans and keep occupying their + // (channel_type, app_id) routing slots, making those bots un-rebindable (#4810). + if err := qtx.DeleteChannelInstallationsByArchivedRuntimeAgents(r.Context(), rt.ID); err != nil { + writeError(w, http.StatusInternalServerError, "failed to clean up channel installations") + return + } if err := qtx.DeleteArchivedAgentsByRuntime(r.Context(), rt.ID); err != nil { writeError(w, http.StatusInternalServerError, "failed to clean up archived agents") return diff --git a/server/internal/handler/runtime_profile.go b/server/internal/handler/runtime_profile.go index b66964ffe0..5b9292f0c2 100644 --- a/server/internal/handler/runtime_profile.go +++ b/server/internal/handler/runtime_profile.go @@ -458,6 +458,15 @@ func (h *Handler) DeleteRuntimeProfile(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusInternalServerError, "failed to clean up agent invocation targets") return } + // Same app-layer cleanup for channel installations: channel_* has no + // workspace/agent FK (MUL-3515 §4), so an archived agent's bot + // installations would otherwise survive the hard-delete as orphans and + // keep occupying their (channel_type, app_id) routing slots, making those + // bots un-rebindable (#4810). + if err := qtx.DeleteChannelInstallationsByArchivedRuntimeAgents(r.Context(), rid); err != nil { + writeError(w, http.StatusInternalServerError, "failed to clean up channel installations") + return + } if err := qtx.DeleteArchivedAgentsByRuntime(r.Context(), rid); err != nil { writeError(w, http.StatusInternalServerError, "failed to clean up archived agents") return diff --git a/server/internal/handler/slack.go b/server/internal/handler/slack.go index b79710d6b5..527e547645 100644 --- a/server/internal/handler/slack.go +++ b/server/internal/handler/slack.go @@ -145,8 +145,12 @@ func (h *Handler) RegisterSlackBYO(w http.ResponseWriter, r *http.Request) { switch { case errors.Is(err, slack.ErrInvalidBotToken), errors.Is(err, slack.ErrInvalidAppToken), errors.Is(err, slack.ErrTokenAppMismatch): writeError(w, http.StatusBadRequest, err.Error()) + case errors.Is(err, slack.ErrTeamOwnedBySameWorkspace): + writeError(w, http.StatusConflict, "this Slack app is already connected to another agent in this workspace — disconnect it there first, then connect it here") + case errors.Is(err, slack.ErrTeamOwnedByArchivedAgent): + writeError(w, http.StatusConflict, "this Slack app is connected to an archived agent in this workspace — restore that agent, or disconnect its bot, before connecting it here") case errors.Is(err, slack.ErrTeamOwnedByAnotherWorkspace): - writeError(w, http.StatusConflict, "this Slack app is already connected to a different Multica workspace") + writeError(w, http.StatusConflict, "this Slack app is already connected to a different Multica workspace — disconnect it there before connecting it here") default: // The dominant non-sentinel failure here is auth.test rejecting the // pasted bot token (a user error), so guide the user to recheck the diff --git a/server/internal/integrations/lark/channel_cleanup_test.go b/server/internal/integrations/lark/channel_cleanup_test.go new file mode 100644 index 0000000000..72cd696ed5 --- /dev/null +++ b/server/internal/integrations/lark/channel_cleanup_test.go @@ -0,0 +1,201 @@ +package lark + +import ( + "context" + "testing" + + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/multica-ai/multica/server/internal/util" + db "github.com/multica-ai/multica/server/pkg/db/generated" +) + +// Delete-time channel cleanup fixtures. These pin the #4810 fix on the OTHER +// half of "auto-reclaim on delete": deleting a workspace, or hard-deleting an +// archived agent on runtime teardown, must sweep the channel_installation rows +// (and every dependent row) their owners left behind — channel_* has no FK to +// workspace/agent (MUL-3515 §4), so nothing else would. +const ( + ccWS = "cc000000-0000-4000-8000-000000000001" + ccRuntime = "cc000000-0000-4000-8000-000000000003" + ccAgentArch = "cc000000-0000-4000-8000-00000000000a" + ccAgentLive = "cc000000-0000-4000-8000-00000000000b" + ccInstaller = "cc000000-0000-4000-8000-000000000005" + ccUser = "cc000000-0000-4000-8000-000000000006" + ccChatArch = "cc000000-0000-4000-8000-0000000000c1" + ccChatLive = "cc000000-0000-4000-8000-0000000000c2" + ccTokenArch = "cc_token_archived" + ccTokenLive = "cc_token_live" + ccAuditArch = "ev_cc_archived" + ccAuditLive = "ev_cc_live" + ccAppArchive = "cli_cc_archived" + ccAppLive = "cli_cc_live" + + // Workspace-delete fixture (its own workspace so the DeleteWorkspace can + // remove it, cascading the runtime + agent). + ccWSDel = "cc000000-0000-4000-8000-000000000002" + ccRuntimeDel = "cc000000-0000-4000-8000-000000000004" + ccAgentDel = "cc000000-0000-4000-8000-00000000000d" + ccChatWs = "cc000000-0000-4000-8000-0000000000c3" + ccTokenWs = "cc_token_ws" + ccAuditWs = "ev_cc_ws" + ccAppWs = "cli_cc_ws" +) + +// seedFullInstallation inserts one active installation plus the full spread of +// dependents (member link, chat-session binding, pending binding token, inbound +// audit) so a cleanup path can be shown to sweep all of them. +func seedFullInstallation(t *testing.T, ctx context.Context, pool *pgxpool.Pool, ws, agent, app, chatSess, tokenHash, auditEvent string) string { + t.Helper() + var id string + if err := pool.QueryRow(ctx, ` +INSERT INTO channel_installation (workspace_id, agent_id, channel_type, config, installer_user_id, status) +VALUES ($1, $2, 'feishu', jsonb_build_object('app_id', $3::text), $4, 'active') +RETURNING id +`, ws, agent, app, ccInstaller).Scan(&id); err != nil { + t.Fatalf("seed installation app=%s: %v", app, err) + } + exec := func(q string, args ...any) { + if _, err := pool.Exec(ctx, q, args...); err != nil { + t.Fatalf("seed dependent for app=%s: %v", app, err) + } + } + exec(`INSERT INTO channel_user_binding (workspace_id, multica_user_id, installation_id, channel_type, channel_user_id) +VALUES ($1, $2, $3, 'feishu', 'ou_cc_user')`, ws, ccUser, id) + exec(`INSERT INTO channel_chat_session_binding (chat_session_id, installation_id, channel_type, channel_chat_id, chat_type) +VALUES ($1, $2, 'feishu', 'oc_cc_chat', 'p2p')`, chatSess, id) + exec(`INSERT INTO channel_binding_token (token_hash, workspace_id, installation_id, channel_type, channel_user_id, expires_at) +VALUES ($1, $2, $3, 'feishu', 'ou_cc_user', now() + interval '10 minutes')`, tokenHash, ws, id) + exec(`INSERT INTO channel_inbound_audit (installation_id, channel_type, event_type, channel_event_id, drop_reason) +VALUES ($1, 'feishu', 'im.message.receive_v1', $2, 'test')`, id, auditEvent) + return id +} + +func ccCount(t *testing.T, ctx context.Context, pool *pgxpool.Pool, q string, args ...any) int { + t.Helper() + var n int + if err := pool.QueryRow(ctx, q, args...).Scan(&n); err != nil { + t.Fatalf("count: %v", err) + } + return n +} + +// assertInstallationSwept checks an installation and all its dependents are gone, +// with the inbound-audit row preserved but detached (installation_id NULL). +func assertInstallationSwept(t *testing.T, ctx context.Context, pool *pgxpool.Pool, id, auditEvent string) { + t.Helper() + if n := ccCount(t, ctx, pool, `SELECT count(*) FROM channel_installation WHERE id = $1`, id); n != 0 { + t.Fatalf("installation not swept: %d rows remain (its bot's app_id slot stays occupied)", n) + } + if n := ccCount(t, ctx, pool, `SELECT count(*) FROM channel_user_binding WHERE installation_id = $1`, id); n != 0 { + t.Fatalf("member links not swept: %d dangling rows", n) + } + if n := ccCount(t, ctx, pool, `SELECT count(*) FROM channel_chat_session_binding WHERE installation_id = $1`, id); n != 0 { + t.Fatalf("chat-session bindings not swept: %d dangling rows", n) + } + if n := ccCount(t, ctx, pool, `SELECT count(*) FROM channel_binding_token WHERE installation_id = $1`, id); n != 0 { + t.Fatalf("binding tokens not swept: %d dangling rows", n) + } + if n := ccCount(t, ctx, pool, `SELECT count(*) FROM channel_inbound_audit WHERE channel_event_id = $1 AND installation_id IS NULL`, auditEvent); n != 1 { + t.Fatalf("audit row should survive detached (installation_id NULL), got %d", n) + } +} + +func assertInstallationIntact(t *testing.T, ctx context.Context, pool *pgxpool.Pool, id, auditEvent string) { + t.Helper() + if n := ccCount(t, ctx, pool, `SELECT count(*) FROM channel_installation WHERE id = $1`, id); n != 1 { + t.Fatalf("a live-owner installation was swept: %d rows (want 1)", n) + } + if n := ccCount(t, ctx, pool, `SELECT count(*) FROM channel_user_binding WHERE installation_id = $1`, id); n != 1 { + t.Fatalf("live-owner member link was swept: %d (want 1)", n) + } + if n := ccCount(t, ctx, pool, `SELECT count(*) FROM channel_inbound_audit WHERE channel_event_id = $1 AND installation_id = $2`, auditEvent, id); n != 1 { + t.Fatalf("live-owner audit ref was detached: %d (want 1)", n) + } +} + +// TestDeleteChannelInstallationsByArchivedRuntimeAgents: runtime teardown hard- +// deletes archived agents; this cleanup must sweep exactly those agents' +// installations (and dependents), leaving live agents on the runtime untouched. +func TestDeleteChannelInstallationsByArchivedRuntimeAgents(t *testing.T) { + pool := channelScopeTestDB(t) + ctx := context.Background() + q := db.New(pool) + + clean := func() { + _, _ = pool.Exec(ctx, `DELETE FROM channel_installation WHERE config->>'app_id' = ANY($1)`, []string{ccAppArchive, ccAppLive}) + _, _ = pool.Exec(ctx, `DELETE FROM channel_user_binding WHERE multica_user_id = $1`, ccUser) + _, _ = pool.Exec(ctx, `DELETE FROM channel_chat_session_binding WHERE chat_session_id = ANY($1)`, []string{ccChatArch, ccChatLive}) + _, _ = pool.Exec(ctx, `DELETE FROM channel_binding_token WHERE token_hash = ANY($1)`, []string{ccTokenArch, ccTokenLive}) + _, _ = pool.Exec(ctx, `DELETE FROM channel_inbound_audit WHERE channel_event_id = ANY($1)`, []string{ccAuditArch, ccAuditLive}) + _, _ = pool.Exec(ctx, `DELETE FROM workspace WHERE id = $1`, ccWS) + } + clean() + t.Cleanup(clean) + + exec := func(query string, args ...any) { + if _, err := pool.Exec(ctx, query, args...); err != nil { + t.Fatalf("seed owner: %v", err) + } + } + exec(`INSERT INTO workspace (id, name, slug, description) VALUES ($1, 'cc ws', 'cc-ws', '')`, ccWS) + exec(`INSERT INTO agent_runtime (id, workspace_id, name, runtime_mode, provider) +VALUES ($1, $2, 'cc runtime', 'local', 'multica_daemon')`, ccRuntime, ccWS) + exec(`INSERT INTO agent (id, workspace_id, name, runtime_mode, runtime_id, archived_at) +VALUES ($1, $2, 'cc archived agent', 'local', $3, now())`, ccAgentArch, ccWS, ccRuntime) + exec(`INSERT INTO agent (id, workspace_id, name, runtime_mode, runtime_id) +VALUES ($1, $2, 'cc live agent', 'local', $3)`, ccAgentLive, ccWS, ccRuntime) + + archivedID := seedFullInstallation(t, ctx, pool, ccWS, ccAgentArch, ccAppArchive, ccChatArch, ccTokenArch, ccAuditArch) + liveID := seedFullInstallation(t, ctx, pool, ccWS, ccAgentLive, ccAppLive, ccChatLive, ccTokenLive, ccAuditLive) + + if err := q.DeleteChannelInstallationsByArchivedRuntimeAgents(ctx, util.MustParseUUID(ccRuntime)); err != nil { + t.Fatalf("DeleteChannelInstallationsByArchivedRuntimeAgents: %v", err) + } + + assertInstallationSwept(t, ctx, pool, archivedID, ccAuditArch) + assertInstallationIntact(t, ctx, pool, liveID, ccAuditLive) +} + +// TestDeleteWorkspace_SweepsChannelInstallations: deleting a workspace must sweep +// its channel installations (and dependents) so no orphan keeps occupying its +// bot's routing slot after the workspace — and its cascade-deleted agents — are +// gone. +func TestDeleteWorkspace_SweepsChannelInstallations(t *testing.T) { + pool := channelScopeTestDB(t) + ctx := context.Background() + q := db.New(pool) + + clean := func() { + _, _ = pool.Exec(ctx, `DELETE FROM channel_installation WHERE config->>'app_id' = $1`, ccAppWs) + _, _ = pool.Exec(ctx, `DELETE FROM channel_user_binding WHERE multica_user_id = $1`, ccUser) + _, _ = pool.Exec(ctx, `DELETE FROM channel_chat_session_binding WHERE chat_session_id = $1`, ccChatWs) + _, _ = pool.Exec(ctx, `DELETE FROM channel_binding_token WHERE token_hash = $1`, ccTokenWs) + _, _ = pool.Exec(ctx, `DELETE FROM channel_inbound_audit WHERE channel_event_id = $1`, ccAuditWs) + _, _ = pool.Exec(ctx, `DELETE FROM workspace WHERE id = $1`, ccWSDel) + } + clean() + t.Cleanup(clean) + + exec := func(query string, args ...any) { + if _, err := pool.Exec(ctx, query, args...); err != nil { + t.Fatalf("seed owner: %v", err) + } + } + exec(`INSERT INTO workspace (id, name, slug, description) VALUES ($1, 'cc ws del', 'cc-ws-del', '')`, ccWSDel) + exec(`INSERT INTO agent_runtime (id, workspace_id, name, runtime_mode, provider) +VALUES ($1, $2, 'cc runtime del', 'local', 'multica_daemon')`, ccRuntimeDel, ccWSDel) + exec(`INSERT INTO agent (id, workspace_id, name, runtime_mode, runtime_id) +VALUES ($1, $2, 'cc agent del', 'local', $3)`, ccAgentDel, ccWSDel, ccRuntimeDel) + + id := seedFullInstallation(t, ctx, pool, ccWSDel, ccAgentDel, ccAppWs, ccChatWs, ccTokenWs, ccAuditWs) + + if err := q.DeleteWorkspace(ctx, util.MustParseUUID(ccWSDel)); err != nil { + t.Fatalf("DeleteWorkspace: %v", err) + } + + if n := ccCount(t, ctx, pool, `SELECT count(*) FROM workspace WHERE id = $1`, ccWSDel); n != 0 { + t.Fatalf("workspace not deleted: %d rows", n) + } + assertInstallationSwept(t, ctx, pool, id, ccAuditWs) +} diff --git a/server/internal/integrations/lark/channel_store.go b/server/internal/integrations/lark/channel_store.go index 74dd07bc04..351653bdf5 100644 --- a/server/internal/integrations/lark/channel_store.go +++ b/server/internal/integrations/lark/channel_store.go @@ -142,61 +142,39 @@ func (s *ChannelStore) UpsertLarkInstallation(ctx context.Context, arg UpsertIns return installationFromRow(row) } -// RemoveRevokedInstallationByAppID clears a revoked installation that belongs to -// a DIFFERENT agent in the same workspace and holds the (channel_type, -// config->>'app_id') unique slot, so the caller can re-install the same -// Lark/Feishu app against a new agent without tripping the functional unique -// index. The guarded DELETE is the atomic gate: it removes the row only when it -// is still (this workspace, another agent, revoked) at delete time and RETURNS -// its id. Only when a row is actually claimed do we clean its dependents. -// -// - the SAME agent's revoked row (agent_id = agentID) never matches, so the -// upsert reactivates it in place (installation_id + all bindings preserved); -// - an ACTIVE row (bot still connected) never matches, so the install surfaces -// as a conflict instead of silently stealing it; -// - a row in ANOTHER workspace never matches. -// -// Keying cleanup off the RETURNING id — rather than a prior read — closes the -// read-then-delete race: a concurrent same-agent reconnect that flips the row -// to 'active' between a read and the delete would previously still trigger the -// dependent cleanup on a stale id and wipe a since-reactivated installation's -// bindings. Under READ COMMITTED the DELETE re-checks status='revoked' against -// the live row, so it claims nothing in that case and we touch no dependents. -// -// channel_* has no FK/cascade (MUL-3515 §4), so every application-owned row that -// referenced the deleted installation is cleared explicitly — chat-session -// bindings, pending binding tokens, member links — and inbound-audit rows are -// detached by NULLing installation_id (the app-layer stand-in for the old -// ON DELETE SET NULL). With no FK the order is free, so cleanup runs after the -// claiming delete; all of it is on the caller's transaction-bound queries, so -// delete + cleanup + the follow-up upsert commit (or roll back) atomically. -func (s *ChannelStore) RemoveRevokedInstallationByAppID(ctx context.Context, workspaceID, agentID pgtype.UUID, appID string) error { - installID, err := s.Queries.DeleteRevokedChannelInstallationByAppID(ctx, db.DeleteRevokedChannelInstallationByAppIDParams{ +// ReclaimDeadInstallationByAppID frees the (feishu, config->>'app_id') routing +// slot before a rebind by removing a DEAD prior owner of the same Lark/Feishu +// app — a revoked placeholder left by a DIFFERENT agent in this workspace, or an +// ORPHAN whose owning workspace/agent has been deleted (#4810) — together with +// every dependent row of that installation, in a single statement. A live owner +// is deliberately left in place: the SAME agent's own revoked row (reactivated by +// the follow-up upsert), and any ACTIVE owner whose agent still exists — including +// an ARCHIVED agent, since archiving is reversible — so the upsert surfaces a +// conflict instead of silently stealing the bot. See the full contract, and the +// TOCTOU / EvalPlanQual reasoning, on ReclaimDeadChannelInstallationByAppID. +func (s *ChannelStore) ReclaimDeadInstallationByAppID(ctx context.Context, workspaceID, agentID pgtype.UUID, appID string) error { + _, err := s.Queries.ReclaimDeadChannelInstallationByAppID(ctx, db.ReclaimDeadChannelInstallationByAppIDParams{ ChannelType: channelTypeFeishu, AppID: appID, WorkspaceID: workspaceID, AgentID: agentID, }) - if err != nil { - if errors.Is(err, pgx.ErrNoRows) { - return nil // no blocking row claimed (same agent / active / other workspace / none) - } + if err != nil && !errors.Is(err, pgx.ErrNoRows) { + // pgx.ErrNoRows just means nothing was dead — a no-op, not a failure. return err } + return nil +} - if err := s.Queries.DeleteChannelChatSessionBindingsByInstallation(ctx, db.DeleteChannelChatSessionBindingsByInstallationParams{ - InstallationID: installID, - ChannelType: channelTypeFeishu, - }); err != nil { - return err - } - if err := s.Queries.DeleteChannelBindingTokensByInstallation(ctx, installID); err != nil { - return err - } - if err := s.Queries.DeleteChannelUserBindingsByInstallation(ctx, installID); err != nil { - return err - } - return s.Queries.NullChannelInboundAuditInstallationID(ctx, installID) +// InstallationOwnerByAppID returns the current owner of the (feishu, app_id) +// routing slot so the caller can build an accurate rebind-conflict message. +// Called after ReclaimDeadInstallationByAppID, so a row here is a live owner; +// pgx.ErrNoRows means the slot is free. +func (s *ChannelStore) InstallationOwnerByAppID(ctx context.Context, appID string) (db.GetChannelInstallationOwnerByAppIDRow, error) { + return s.Queries.GetChannelInstallationOwnerByAppID(ctx, db.GetChannelInstallationOwnerByAppIDParams{ + ChannelType: channelTypeFeishu, + AppID: appID, + }) } func (s *ChannelStore) SetLarkInstallationStatus(ctx context.Context, arg SetInstallationStatusParams) error { diff --git a/server/internal/integrations/lark/channel_store_rebind_test.go b/server/internal/integrations/lark/channel_store_rebind_test.go index f46ea0fbd9..0003dfeef8 100644 --- a/server/internal/integrations/lark/channel_store_rebind_test.go +++ b/server/internal/integrations/lark/channel_store_rebind_test.go @@ -8,6 +8,7 @@ import ( "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgtype" + "github.com/jackc/pgx/v5/pgxpool" "github.com/multica-ai/multica/server/internal/util" db "github.com/multica-ai/multica/server/pkg/db/generated" @@ -18,27 +19,74 @@ import ( // so these rows need no parent records; the test cleans up by deterministic key // before and after (a killed prior run must not leave colliding rows behind). const ( - rbWS = "5c09e100-0000-4000-8000-000000000001" - rbWS2 = "5c09e100-0000-4000-8000-000000000002" - rbAgentA = "5c09e100-0000-4000-8000-00000000000a" - rbAgentB = "5c09e100-0000-4000-8000-00000000000b" - rbInstaller = "5c09e100-0000-4000-8000-000000000005" - rbUser = "5c09e100-0000-4000-8000-000000000006" - rbChatSess = "5c09e100-0000-4000-8000-000000000007" + rbWS = "5c09e100-0000-4000-8000-000000000001" + rbWS2 = "5c09e100-0000-4000-8000-000000000002" + rbRuntime = "5c09e100-0000-4000-8000-000000000003" + rbAgentA = "5c09e100-0000-4000-8000-00000000000a" + rbAgentB = "5c09e100-0000-4000-8000-00000000000b" + rbAgentArch = "5c09e100-0000-4000-8000-00000000000c" + rbInstaller = "5c09e100-0000-4000-8000-000000000005" + rbUser = "5c09e100-0000-4000-8000-000000000006" + rbChatSess = "5c09e100-0000-4000-8000-000000000007" + rbGhostWS = "5c09e100-0000-4000-8000-0000000000f1" // never seeded: a deleted workspace + rbGhostAgent = "5c09e100-0000-4000-8000-0000000000f2" // never seeded: a hard-deleted agent - rbAppSame = "cli_rb_same" - rbAppDiff = "cli_rb_diff" - rbAppActive = "cli_rb_active" - rbAppWsFence = "cli_rb_wsfence" - rbAppReactivate = "cli_rb_reactivate" - rbAppMove = "cli_rb_move" + rbAppSame = "cli_rb_same" + rbAppDiff = "cli_rb_diff" + rbAppActive = "cli_rb_active" + rbAppWsFence = "cli_rb_wsfence" + rbAppReactivate = "cli_rb_reactivate" + rbAppMove = "cli_rb_move" + rbAppOrphanWS = "cli_rb_orphan_ws" + rbAppOrphanAgent = "cli_rb_orphan_agent" + rbAppArchived = "cli_rb_archived" + rbAppLive = "cli_rb_live" ) -// TestChannelStore_RemoveRevokedInstallationByAppID guards the WHERE clause of -// the DeleteRevokedChannelInstallationByAppID gate: it must claim ONLY a revoked -// row that belongs to a DIFFERENT agent in the SAME workspace. The same agent's -// own revoked row, any active row, and rows in another workspace must survive. -func TestChannelStore_RemoveRevokedInstallationByAppID(t *testing.T) { +// seedRebindOwners inserts the workspace/runtime/agent rows the rebind fixtures +// reference. ReclaimDeadChannelInstallationByAppID now treats an installation +// whose owning workspace OR agent row is gone as a DEAD orphan to reclaim, so +// these tests must give their rows real owners — otherwise the orphan branch, +// not the revoked/same-agent/other-workspace fences, would decide every case. +// rbAgentArch is archived (a live-but-reversible owner). Idempotent; the matching +// teardown drops the workspaces, which cascades to the runtime and agents. +func seedRebindOwners(t *testing.T, ctx context.Context, pool *pgxpool.Pool) { + t.Helper() + exec := func(q string, args ...any) { + if _, err := pool.Exec(ctx, q, args...); err != nil { + t.Fatalf("seed rebind owner: %v", err) + } + } + for _, ws := range []string{rbWS, rbWS2} { + exec(`INSERT INTO workspace (id, name, slug, description) VALUES ($1, $2, $3, '') ON CONFLICT (id) DO NOTHING`, + ws, "rebind "+ws, "rebind-"+ws) + } + exec(`INSERT INTO agent_runtime (id, workspace_id, name, runtime_mode, provider) +VALUES ($1, $2, 'rebind runtime', 'local', 'multica_daemon') ON CONFLICT (id) DO NOTHING`, rbRuntime, rbWS) + // Names must be unique per workspace (agent_workspace_name_unique), so key + // each on its id. + for _, agent := range []string{rbAgentA, rbAgentB} { + exec(`INSERT INTO agent (id, workspace_id, name, runtime_mode, runtime_id) +VALUES ($1, $2, $3, 'local', $4) ON CONFLICT (id) DO NOTHING`, agent, rbWS, "rebind agent "+agent, rbRuntime) + } + exec(`INSERT INTO agent (id, workspace_id, name, runtime_mode, runtime_id, archived_at) +VALUES ($1, $2, $3, 'local', $4, now()) ON CONFLICT (id) DO NOTHING`, rbAgentArch, rbWS, "rebind archived agent "+rbAgentArch, rbRuntime) +} + +// cleanRebindOwners drops the seeded workspaces; the FK ON DELETE CASCADE takes +// the runtime and agents with them. channel_installation has no such FK — the +// bug under test — so those rows are cleaned by app_id separately. +func cleanRebindOwners(ctx context.Context, pool *pgxpool.Pool) { + _, _ = pool.Exec(ctx, `DELETE FROM workspace WHERE id = ANY($1)`, []string{rbWS, rbWS2}) +} + +// TestChannelStore_ReclaimDeadRevokedFences guards the REVOKED branch of the +// ReclaimDeadChannelInstallationByAppID gate: with live owners seeded, it must +// claim ONLY a revoked row that belongs to a DIFFERENT agent in the SAME +// workspace. The same agent's own revoked row, any active row, and a revoked row +// in another (still-live) workspace must survive. (The orphan branch is covered +// by TestChannelStore_ReclaimDeadReclaimsOrphansRefusesLiveOwners.) +func TestChannelStore_ReclaimDeadRevokedFences(t *testing.T) { pool := channelScopeTestDB(t) ctx := context.Background() store := NewChannelStore(db.New(pool)) @@ -50,6 +98,8 @@ func TestChannelStore_RemoveRevokedInstallationByAppID(t *testing.T) { _, _ = pool.Exec(ctx, `DELETE FROM channel_chat_session_binding WHERE chat_session_id = $1`, rbChatSess) } clean() + seedRebindOwners(t, ctx, pool) + t.Cleanup(func() { cleanRebindOwners(ctx, pool) }) t.Cleanup(clean) // insert an installation and return its id. @@ -83,8 +133,8 @@ RETURNING id t.Run("same agent revoked row is preserved", func(t *testing.T) { clean() id := insert(rbAppSame, rbWS, rbAgentA, "revoked") - if err := store.RemoveRevokedInstallationByAppID(ctx, wsUUID, agentAUUID, rbAppSame); err != nil { - t.Fatalf("RemoveRevokedInstallationByAppID: %v", err) + if err := store.ReclaimDeadInstallationByAppID(ctx, wsUUID, agentAUUID, rbAppSame); err != nil { + t.Fatalf("ReclaimDeadInstallationByAppID: %v", err) } if !exists(id) { t.Fatal("same agent's own revoked row was deleted; it must be reactivated in place by the upsert, not orphaned") @@ -94,8 +144,8 @@ RETURNING id t.Run("different agent revoked row is deleted", func(t *testing.T) { clean() id := insert(rbAppDiff, rbWS, rbAgentA, "revoked") - if err := store.RemoveRevokedInstallationByAppID(ctx, wsUUID, agentBUUID, rbAppDiff); err != nil { - t.Fatalf("RemoveRevokedInstallationByAppID: %v", err) + if err := store.ReclaimDeadInstallationByAppID(ctx, wsUUID, agentBUUID, rbAppDiff); err != nil { + t.Fatalf("ReclaimDeadInstallationByAppID: %v", err) } if exists(id) { t.Fatal("a different agent's revoked row was not deleted; it would keep blocking the app_id unique slot") @@ -105,8 +155,8 @@ RETURNING id t.Run("active row is never deleted", func(t *testing.T) { clean() id := insert(rbAppActive, rbWS, rbAgentA, "active") - if err := store.RemoveRevokedInstallationByAppID(ctx, wsUUID, agentBUUID, rbAppActive); err != nil { - t.Fatalf("RemoveRevokedInstallationByAppID: %v", err) + if err := store.ReclaimDeadInstallationByAppID(ctx, wsUUID, agentBUUID, rbAppActive); err != nil { + t.Fatalf("ReclaimDeadInstallationByAppID: %v", err) } if !exists(id) { t.Fatal("an active installation was deleted through the revoked-cleanup path") @@ -116,8 +166,8 @@ RETURNING id t.Run("other workspace revoked row is preserved", func(t *testing.T) { clean() id := insert(rbAppWsFence, rbWS2, rbAgentA, "revoked") - if err := store.RemoveRevokedInstallationByAppID(ctx, wsUUID, agentBUUID, rbAppWsFence); err != nil { - t.Fatalf("RemoveRevokedInstallationByAppID: %v", err) + if err := store.ReclaimDeadInstallationByAppID(ctx, wsUUID, agentBUUID, rbAppWsFence); err != nil { + t.Fatalf("ReclaimDeadInstallationByAppID: %v", err) } if !exists(id) { t.Fatal("a revoked row in another workspace was deleted; the delete must stay workspace-scoped") @@ -145,6 +195,8 @@ func TestChannelStore_ReinstallReactivationSemantics(t *testing.T) { _, _ = pool.Exec(ctx, `DELETE FROM channel_chat_session_binding WHERE chat_session_id = $1`, rbChatSess) } clean() + seedRebindOwners(t, ctx, pool) + t.Cleanup(func() { cleanRebindOwners(ctx, pool) }) t.Cleanup(clean) insertRevoked := func(app, agent string) pgtype.UUID { @@ -207,7 +259,7 @@ VALUES ($1, $2, 'feishu', 'oc_rb_chat', 'p2p') // finishSuccess order: cleanup for the current agent (a no-op for the // same agent), then upsert. - if err := store.RemoveRevokedInstallationByAppID(ctx, util.MustParseUUID(rbWS), util.MustParseUUID(rbAgentA), rbAppReactivate); err != nil { + if err := store.ReclaimDeadInstallationByAppID(ctx, util.MustParseUUID(rbWS), util.MustParseUUID(rbAgentA), rbAppReactivate); err != nil { t.Fatalf("cleanup: %v", err) } inst := upsert(rbAgentA, rbAppReactivate) @@ -228,7 +280,7 @@ VALUES ($1, $2, 'feishu', 'oc_rb_chat', 'p2p') oldID := insertRevoked(rbAppMove, rbAgentA) attachBindings(oldID) - if err := store.RemoveRevokedInstallationByAppID(ctx, util.MustParseUUID(rbWS), util.MustParseUUID(rbAgentB), rbAppMove); err != nil { + if err := store.ReclaimDeadInstallationByAppID(ctx, util.MustParseUUID(rbWS), util.MustParseUUID(rbAgentB), rbAppMove); err != nil { t.Fatalf("cleanup: %v", err) } inst := upsert(rbAgentB, rbAppMove) @@ -269,6 +321,8 @@ func TestChannelStore_RebindCleansDependentRows(t *testing.T) { _, _ = pool.Exec(ctx, `DELETE FROM channel_inbound_audit WHERE channel_event_id = $1`, auditEvent) } clean() + seedRebindOwners(t, ctx, pool) + t.Cleanup(func() { cleanRebindOwners(ctx, pool) }) t.Cleanup(clean) // A revoked installation for agent A carrying the full spread of dependents. @@ -295,8 +349,8 @@ VALUES ($1, $2, $3, 'feishu', 'ou_rb_user', now() + interval '10 minutes')`, tok VALUES ($1, 'feishu', 'im.message.receive_v1', $2, 'revoked_installation')`, oldID, auditEvent) // Rebind the app to a DIFFERENT agent. - if err := store.RemoveRevokedInstallationByAppID(ctx, util.MustParseUUID(rbWS), util.MustParseUUID(rbAgentB), app); err != nil { - t.Fatalf("RemoveRevokedInstallationByAppID: %v", err) + if err := store.ReclaimDeadInstallationByAppID(ctx, util.MustParseUUID(rbWS), util.MustParseUUID(rbAgentB), app); err != nil { + t.Fatalf("ReclaimDeadInstallationByAppID: %v", err) } count := func(q string, args ...any) int { @@ -335,7 +389,7 @@ VALUES ($1, 'feishu', 'im.message.receive_v1', $2, 'revoked_installation')`, old // - txReconnect (agent A reconnecting to the SAME agent) reactivates the row // to 'active' but holds the row lock uncommitted; // - txRebind (agent B rebinding to a DIFFERENT agent) runs the full cleanup -// via RemoveRevokedInstallationByAppID. +// via ReclaimDeadInstallationByAppID. // // The old read-then-clean-then-delete shape would read the still-committed // 'revoked' row, wipe its dependents, then no-op on the fenced delete — losing @@ -363,6 +417,8 @@ func TestChannelStore_RebindGuardedDeleteRaceWithReactivation(t *testing.T) { _, _ = pool.Exec(ctx, `DELETE FROM channel_inbound_audit WHERE channel_event_id = $1`, auditEvent) } clean() + seedRebindOwners(t, ctx, pool) + t.Cleanup(func() { cleanRebindOwners(ctx, pool) }) t.Cleanup(clean) // A revoked installation for agent A, with the full spread of dependents. @@ -408,7 +464,7 @@ VALUES ($1, 'feishu', 'im.message.receive_v1', $2, 'revoked_installation')`, idS return } defer txRebind.Rollback(ctx) - if err := store.WithTx(txRebind).RemoveRevokedInstallationByAppID(ctx, util.MustParseUUID(rbWS), util.MustParseUUID(rbAgentB), app); err != nil { + if err := store.WithTx(txRebind).ReclaimDeadInstallationByAppID(ctx, util.MustParseUUID(rbWS), util.MustParseUUID(rbAgentB), app); err != nil { done <- err return } @@ -452,3 +508,110 @@ VALUES ($1, 'feishu', 'im.message.receive_v1', $2, 'revoked_installation')`, idS t.Fatalf("audit reference detached by the racing rebind: got %d, want 1", n) } } + +// TestChannelStore_ReclaimDeadReclaimsOrphansRefusesLiveOwners covers the ORPHAN +// branch of the reclaim gate — the #4810 fix. An installation whose owning +// workspace or agent has been hard-deleted is a dead orphan still occupying the +// (channel_type, app_id) routing slot; the reclaim must clear it so the bot can +// be rebound. A live owner — an active agent, INCLUDING an archived one — must be +// left in place so the follow-up upsert refuses the rebind rather than stealing +// the bot, and InstallationOwnerByAppID must report who holds it (so the caller +// can build an accurate conflict message). +func TestChannelStore_ReclaimDeadReclaimsOrphansRefusesLiveOwners(t *testing.T) { + pool := channelScopeTestDB(t) + ctx := context.Background() + store := NewChannelStore(db.New(pool)) + + apps := []string{rbAppOrphanWS, rbAppOrphanAgent, rbAppLive, rbAppArchived} + clean := func() { + _, _ = pool.Exec(ctx, `DELETE FROM channel_installation WHERE config->>'app_id' = ANY($1)`, apps) + } + clean() + seedRebindOwners(t, ctx, pool) + t.Cleanup(func() { cleanRebindOwners(ctx, pool) }) + t.Cleanup(clean) + + insert := func(app, ws, agent, status string) pgtype.UUID { + var id string + if err := pool.QueryRow(ctx, ` +INSERT INTO channel_installation (workspace_id, agent_id, channel_type, config, installer_user_id, status) +VALUES ($1, $2, 'feishu', jsonb_build_object('app_id', $3::text), $4, $5) +RETURNING id +`, ws, agent, app, rbInstaller, status).Scan(&id); err != nil { + t.Fatalf("insert installation app=%s: %v", app, err) + } + return util.MustParseUUID(id) + } + exists := func(id pgtype.UUID) bool { + _, err := store.GetLarkInstallation(ctx, id) + if err == nil { + return true + } + if errors.Is(err, pgx.ErrNoRows) { + return false + } + t.Fatalf("GetLarkInstallation: %v", err) + return false + } + // A NEW agent B, in a live workspace, is the one rebinding each app. + wsUUID := util.MustParseUUID(rbWS) + agentBUUID := util.MustParseUUID(rbAgentB) + + t.Run("orphan from a deleted workspace is reclaimed", func(t *testing.T) { + clean() + id := insert(rbAppOrphanWS, rbGhostWS, rbAgentA, "active") + if err := store.ReclaimDeadInstallationByAppID(ctx, wsUUID, agentBUUID, rbAppOrphanWS); err != nil { + t.Fatalf("ReclaimDeadInstallationByAppID: %v", err) + } + if exists(id) { + t.Fatal("an installation whose workspace no longer exists was not reclaimed; the bot would stay un-rebindable") + } + }) + + t.Run("orphan from a hard-deleted agent is reclaimed", func(t *testing.T) { + clean() + id := insert(rbAppOrphanAgent, rbWS, rbGhostAgent, "active") + if err := store.ReclaimDeadInstallationByAppID(ctx, wsUUID, agentBUUID, rbAppOrphanAgent); err != nil { + t.Fatalf("ReclaimDeadInstallationByAppID: %v", err) + } + if exists(id) { + t.Fatal("an installation whose agent was hard-deleted was not reclaimed") + } + }) + + t.Run("live active owner is refused, not stolen", func(t *testing.T) { + clean() + id := insert(rbAppLive, rbWS, rbAgentA, "active") + if err := store.ReclaimDeadInstallationByAppID(ctx, wsUUID, agentBUUID, rbAppLive); err != nil { + t.Fatalf("ReclaimDeadInstallationByAppID: %v", err) + } + if !exists(id) { + t.Fatal("a live active owner was reclaimed; agent B would silently steal the bot") + } + owner, err := store.InstallationOwnerByAppID(ctx, rbAppLive) + if err != nil { + t.Fatalf("InstallationOwnerByAppID: %v", err) + } + if owner.WorkspaceID != wsUUID || owner.AgentID != util.MustParseUUID(rbAgentA) || owner.AgentArchivedAt.Valid { + t.Fatalf("owner mismatch: ws=%v agent=%v archived=%v", owner.WorkspaceID, owner.AgentID, owner.AgentArchivedAt.Valid) + } + }) + + t.Run("archived agent owner is refused and reported archived", func(t *testing.T) { + clean() + id := insert(rbAppArchived, rbWS, rbAgentArch, "active") + if err := store.ReclaimDeadInstallationByAppID(ctx, wsUUID, agentBUUID, rbAppArchived); err != nil { + t.Fatalf("ReclaimDeadInstallationByAppID: %v", err) + } + if !exists(id) { + t.Fatal("an archived agent's installation was reclaimed; archiving is reversible so the bot must stay owned") + } + owner, err := store.InstallationOwnerByAppID(ctx, rbAppArchived) + if err != nil { + t.Fatalf("InstallationOwnerByAppID: %v", err) + } + if !owner.AgentArchivedAt.Valid { + t.Fatal("owner lookup did not report the agent as archived; the conflict message could not distinguish it") + } + }) +} diff --git a/server/internal/integrations/lark/registration_service.go b/server/internal/integrations/lark/registration_service.go index 9aba3426c1..2dd640869d 100644 --- a/server/internal/integrations/lark/registration_service.go +++ b/server/internal/integrations/lark/registration_service.go @@ -11,12 +11,18 @@ import ( "sync" "time" + "github.com/jackc/pgx/v5/pgconn" "github.com/jackc/pgx/v5/pgtype" "github.com/multica-ai/multica/server/internal/events" db "github.com/multica-ai/multica/server/pkg/db/generated" "github.com/multica-ai/multica/server/pkg/protocol" ) +// pgUniqueViolation is the Postgres SQLSTATE for a unique-constraint violation. +// A rebind upsert that trips the (channel_type, config->>'app_id') index after +// the dead-owner reclaim ran means a LIVE owner still holds the slot. +const pgUniqueViolation = "23505" + // RegistrationSessionStatus is the discriminated state a `begin` // session lives in. The HTTP status endpoint serializes the underlying // string verbatim so the frontend can pattern-match without parsing @@ -571,17 +577,17 @@ func (s *RegistrationService) finishSuccess(ctx context.Context, sess *registrat defer tx.Rollback(ctx) qtx := s.queries.WithTx(tx) - // If the same Feishu app (app_id) was previously bound to a DIFFERENT - // agent in this workspace and later revoked, that revoked row still - // holds the (channel_type, config->>'app_id') unique index slot and - // blocks the UpsertChannelInstallation INSERT below. Remove the - // revoked placeholder first — the transaction wraps both the delete - // and the upsert so a failure between them rolls back cleanly. The - // current agent (sess.agentID) is excluded: re-connecting the SAME - // agent reactivates its own revoked row in place via the upsert's - // ON CONFLICT, keeping its installation_id and every binding intact. - if err := qtx.RemoveRevokedInstallationByAppID(ctx, sess.workspaceID, sess.agentID, res.ClientID); err != nil { - s.cfg.Logger.Warn("lark registration: cleanup revoked installation", + // If the same Feishu app (app_id) is held by a DEAD prior owner — a revoked + // placeholder left by a DIFFERENT agent in this workspace, or an orphan whose + // workspace/agent was deleted (#4810) — that row still occupies the + // (channel_type, config->>'app_id') unique slot and blocks the + // UpsertChannelInstallation INSERT below. Reclaim it first — the transaction + // wraps both the delete and the upsert so a failure between them rolls back + // cleanly. A live owner is left in place (the SAME agent's own revoked row is + // reactivated in place by the upsert; an active/archived agent stays owned), + // so the upsert surfaces the conflict below instead of stealing the bot. + if err := qtx.ReclaimDeadInstallationByAppID(ctx, sess.workspaceID, sess.agentID, res.ClientID); err != nil { + s.cfg.Logger.Warn("lark registration: reclaim dead installation", "session_id", sess.id, "err", err) sess.markError(RegistrationReasonInternalError, err.Error(), s.gcDeadline()) return @@ -600,7 +606,16 @@ func (s *RegistrationService) finishSuccess(ctx context.Context, sess *registrat if err != nil { s.cfg.Logger.Warn("lark registration: upsert installation", "session_id", sess.id, "err", err) - sess.markError(RegistrationReasonInstallationConflict, err.Error(), s.gcDeadline()) + // A unique violation here means the app_id slot is held by a LIVE owner + // (the reclaim above already cleared every dead one). Surface who holds + // it — another agent in this workspace, an archived agent, or a different + // workspace — instead of leaking the raw Postgres error. + msg := err.Error() + var pgErr *pgconn.PgError + if errors.As(err, &pgErr) && pgErr.Code == pgUniqueViolation { + msg = s.liveOwnerConflictMessage(ctx, sess.workspaceID, res.ClientID) + } + sess.markError(RegistrationReasonInstallationConflict, msg, s.gcDeadline()) return } @@ -634,6 +649,28 @@ func (s *RegistrationService) finishSuccess(ctx context.Context, sess *registrat "installation_id", uuidString(inst.ID)) } +// liveOwnerConflictMessage builds the user-facing copy for a rebind refused +// because the Feishu app's routing slot is held by a LIVE owner. It names which +// kind of owner so the user knows how to recover, instead of the old catch-all +// "connected to a different Multica workspace" that lied when the real owner sat +// in the SAME workspace (#4810). Looked up on the base pool, not the aborted +// upsert tx. If the slot turns out free (a concurrent disconnect between the +// upsert and this read), a generic message is enough — the user can just retry. +func (s *RegistrationService) liveOwnerConflictMessage(ctx context.Context, requestingWorkspaceID pgtype.UUID, appID string) string { + owner, err := s.queries.InstallationOwnerByAppID(ctx, appID) + if err != nil { + return "This Feishu app is already connected to another agent. Disconnect it there first, then connect it here." + } + switch { + case owner.WorkspaceID != requestingWorkspaceID: + return "This Feishu app is already connected to a different Multica workspace. Disconnect it there before connecting it here." + case owner.AgentArchivedAt.Valid: + return "This Feishu app is connected to an archived agent in this workspace. Restore that agent, or disconnect its bot, before connecting it here." + default: + return "This Feishu app is already connected to another agent in this workspace. Disconnect it there first, then connect it here." + } +} + func (s *RegistrationService) gcDeadline() time.Time { return s.cfg.Now().Add(s.cfg.SessionTTL) } diff --git a/server/internal/integrations/slack/byo_install.go b/server/internal/integrations/slack/byo_install.go index d6fd0cbc9c..cb0412582b 100644 --- a/server/internal/integrations/slack/byo_install.go +++ b/server/internal/integrations/slack/byo_install.go @@ -123,6 +123,7 @@ func (s *InstallService) RegisterBYO(ctx context.Context, p RegisterBYOParams) ( wsID: p.WorkspaceID, agentID: p.AgentID, installerID: p.InitiatorID, + appIDKey: appID, configJSON: cfgJSON, }) } diff --git a/server/internal/integrations/slack/byo_install_test.go b/server/internal/integrations/slack/byo_install_test.go index fd79282d47..3f565886c8 100644 --- a/server/internal/integrations/slack/byo_install_test.go +++ b/server/internal/integrations/slack/byo_install_test.go @@ -196,15 +196,17 @@ func TestRegisterBYO_AuthTestFailure(t *testing.T) { } } -func TestRegisterBYO_AppAlreadyConnected_Rejected(t *testing.T) { +func TestRegisterBYO_AppConnectedToAnotherWorkspace_Rejected(t *testing.T) { srv := authTestServer(t, true) defer srv.Close() - // The pasted app is already connected to another agent / workspace, so the - // (channel_type, app_id) routing index rejects the upsert (unique violation). - // We must refuse, not steal it. + // The pasted app is live-owned by an agent in a DIFFERENT Multica workspace, + // so after the dead-owner reclaim the (channel_type, app_id) routing index + // still rejects the upsert. We must refuse, not steal it — and name the real + // case (another workspace), not the old catch-all. q := &fakeInstallQueries{ - rowID: mustUUID(t, "44444444-4444-4444-4444-444444444444"), - appIDTaken: true, + rowID: mustUUID(t, "44444444-4444-4444-4444-444444444444"), + appIDTaken: true, + ownerWorkspaceID: mustUUID(t, "99999999-9999-9999-9999-999999999999"), } svc := newTestInstallService(t, q) svc.apiURL = srv.URL + "/" @@ -215,6 +217,55 @@ func TestRegisterBYO_AppAlreadyConnected_Rejected(t *testing.T) { )); err != ErrTeamOwnedByAnotherWorkspace { t.Fatalf("app already connected = %v, want ErrTeamOwnedByAnotherWorkspace", err) } + if !q.reclaimCalled { + t.Error("install must attempt a dead-owner reclaim before the upsert") + } +} + +func TestRegisterBYO_AppConnectedToAnotherAgentSameWorkspace_Rejected(t *testing.T) { + srv := authTestServer(t, true) + defer srv.Close() + // The pasted app is live-owned by a DIFFERENT (non-archived) agent in the + // SAME workspace. The old catch-all wrongly blamed "another workspace"; this + // must surface the same-workspace sentinel so the UI points at the Disconnect + // the user can actually reach (#4810). + q := &fakeInstallQueries{ + rowID: mustUUID(t, "44444444-4444-4444-4444-444444444444"), + appIDTaken: true, + ownerWorkspaceID: mustUUID(t, "11111111-1111-1111-1111-111111111111"), + } + svc := newTestInstallService(t, q) + svc.apiURL = srv.URL + "/" + + if _, err := svc.RegisterBYO(context.Background(), byoParams( + "11111111-1111-1111-1111-111111111111", + "22222222-2222-2222-2222-222222222222", + )); err != ErrTeamOwnedBySameWorkspace { + t.Fatalf("app owned by another agent in this workspace = %v, want ErrTeamOwnedBySameWorkspace", err) + } +} + +func TestRegisterBYO_AppConnectedToArchivedAgent_Rejected(t *testing.T) { + srv := authTestServer(t, true) + defer srv.Close() + // The pasted app's owning agent is archived — a live-but-reversible owner — + // so the reclaim leaves it in place and the upsert is refused. The user is + // told to restore the agent or disconnect its bot, not that it's gone. + q := &fakeInstallQueries{ + rowID: mustUUID(t, "44444444-4444-4444-4444-444444444444"), + appIDTaken: true, + ownerWorkspaceID: mustUUID(t, "11111111-1111-1111-1111-111111111111"), + ownerArchived: true, + } + svc := newTestInstallService(t, q) + svc.apiURL = srv.URL + "/" + + if _, err := svc.RegisterBYO(context.Background(), byoParams( + "11111111-1111-1111-1111-111111111111", + "22222222-2222-2222-2222-222222222222", + )); err != ErrTeamOwnedByArchivedAgent { + t.Fatalf("app owned by an archived agent = %v, want ErrTeamOwnedByArchivedAgent", err) + } } func TestRegisterBYO_ReconnectSameAgent_UpdatesRowInPlace(t *testing.T) { diff --git a/server/internal/integrations/slack/install.go b/server/internal/integrations/slack/install.go index 95a89f7089..6a08b0bfcd 100644 --- a/server/internal/integrations/slack/install.go +++ b/server/internal/integrations/slack/install.go @@ -28,11 +28,20 @@ var ( // ErrInstallationNotFound surfaces "no row matches in this workspace". ErrInstallationNotFound = errors.New("slack installation not found") // ErrTeamOwnedByAnotherWorkspace is returned when the pasted Slack app is - // already connected to a DIFFERENT agent or Multica workspace — it would - // collide with the (channel_type, app_id) routing index. A Slack app is one - // bot identity and maps to one agent; reusing it elsewhere requires - // disconnecting it there first. - ErrTeamOwnedByAnotherWorkspace = errors.New("slack: this Slack app is already connected to another agent or Multica workspace") + // already connected to a live owner in a DIFFERENT Multica workspace — it + // would collide with the (channel_type, app_id) routing index. A Slack app is + // one bot identity and maps to one agent; reusing it here requires + // disconnecting it in the other workspace first. + ErrTeamOwnedByAnotherWorkspace = errors.New("slack: this Slack app is already connected to a different Multica workspace") + // ErrTeamOwnedBySameWorkspace is returned when the app is already connected to + // a DIFFERENT (live, non-archived) agent in the SAME workspace. The old + // catch-all wrongly blamed "another workspace"; naming the same-workspace case + // points the user at the Disconnect they can actually reach (#4810). + ErrTeamOwnedBySameWorkspace = errors.New("slack: this Slack app is already connected to another agent in this workspace") + // ErrTeamOwnedByArchivedAgent is returned when the app's owning agent is + // archived (and so still holds the bot, since archiving is reversible). The + // user recovers by restoring that agent or disconnecting its bot. + ErrTeamOwnedByArchivedAgent = errors.New("slack: this Slack app is connected to an archived agent in this workspace") ) // installQueries is the slice of generated queries InstallService needs. WithTx @@ -41,6 +50,8 @@ var ( type installQueries interface { WithTx(tx pgx.Tx) installQueries UpsertChannelInstallation(ctx context.Context, arg db.UpsertChannelInstallationParams) (db.ChannelInstallation, error) + ReclaimDeadChannelInstallationByAppID(ctx context.Context, arg db.ReclaimDeadChannelInstallationByAppIDParams) (pgtype.UUID, error) + GetChannelInstallationOwnerByAppID(ctx context.Context, arg db.GetChannelInstallationOwnerByAppIDParams) (db.GetChannelInstallationOwnerByAppIDRow, error) ListChannelInstallationsByWorkspace(ctx context.Context, arg db.ListChannelInstallationsByWorkspaceParams) ([]db.ChannelInstallation, error) GetChannelInstallationInWorkspace(ctx context.Context, arg db.GetChannelInstallationInWorkspaceParams) (db.ChannelInstallation, error) SetChannelInstallationStatus(ctx context.Context, arg db.SetChannelInstallationStatusParams) error @@ -115,6 +126,10 @@ type installPersist struct { wsID pgtype.UUID agentID pgtype.UUID installerID pgtype.UUID + // appIDKey is the Slack app id stored at config->>'app_id'; it MUST equal the + // app_id inside configJSON. It keys the dead-owner reclaim and the live-owner + // lookup that drives the accurate conflict message. + appIDKey string // configJSON holds the Slack app id (config->>'app_id') used for inbound // routing; the ROW itself is keyed by (workspace, agent) — one bot per agent. configJSON []byte @@ -142,6 +157,21 @@ func (s *InstallService) persistInstall(ctx context.Context, p installPersist) ( defer func() { _ = tx.Rollback(ctx) }() qtx := s.q.WithTx(tx) + // Free the (slack, app_id) routing slot from any DEAD prior owner — a revoked + // placeholder, or an orphan whose owning workspace/agent was deleted (#4810) — + // before the upsert, so a bot whose old owner is gone can be rebound. A live + // owner (active agent, including an archived one) is left in place and trips + // the unique index below, which we turn into an accurate conflict. + if _, err := qtx.ReclaimDeadChannelInstallationByAppID(ctx, db.ReclaimDeadChannelInstallationByAppIDParams{ + ChannelType: string(TypeSlack), + AppID: p.appIDKey, + WorkspaceID: p.wsID, + AgentID: p.agentID, + }); err != nil && !errors.Is(err, pgx.ErrNoRows) { + // pgx.ErrNoRows just means nothing was dead — a no-op, not a failure. + return db.ChannelInstallation{}, fmt.Errorf("reclaim dead slack installation: %w", err) + } + inst, err := qtx.UpsertChannelInstallation(ctx, db.UpsertChannelInstallationParams{ WorkspaceID: p.wsID, AgentID: p.agentID, @@ -152,7 +182,7 @@ func (s *InstallService) persistInstall(ctx context.Context, p installPersist) ( if err != nil { var pgErr *pgconn.PgError if errors.As(err, &pgErr) && pgErr.Code == pgUniqueViolation { - return db.ChannelInstallation{}, ErrTeamOwnedByAnotherWorkspace + return db.ChannelInstallation{}, s.liveOwnerConflictErr(ctx, p.wsID, p.appIDKey) } return db.ChannelInstallation{}, fmt.Errorf("upsert slack installation: %w", err) } @@ -162,6 +192,31 @@ func (s *InstallService) persistInstall(ctx context.Context, p installPersist) ( return inst, nil } +// liveOwnerConflictErr classifies who holds the (slack, app_id) routing slot +// after the dead-owner reclaim ran, so persistInstall returns a sentinel the +// handler renders as an accurate message rather than the old catch-all that +// always blamed "another workspace" (#4810). Read on the base pool (s.q), since +// the failed upsert has aborted the tx. A now-free slot (concurrent disconnect) +// or lookup error falls back to the generic cross-workspace sentinel — a retry +// then succeeds. +func (s *InstallService) liveOwnerConflictErr(ctx context.Context, requestingWorkspaceID pgtype.UUID, appID string) error { + owner, err := s.q.GetChannelInstallationOwnerByAppID(ctx, db.GetChannelInstallationOwnerByAppIDParams{ + ChannelType: string(TypeSlack), + AppID: appID, + }) + if err != nil { + return ErrTeamOwnedByAnotherWorkspace + } + switch { + case owner.WorkspaceID != requestingWorkspaceID: + return ErrTeamOwnedByAnotherWorkspace + case owner.AgentArchivedAt.Valid: + return ErrTeamOwnedByArchivedAgent + default: + return ErrTeamOwnedBySameWorkspace + } +} + // ListByWorkspace returns every Slack installation in the workspace (active and // revoked), for the management surface. func (s *InstallService) ListByWorkspace(ctx context.Context, wsID pgtype.UUID) ([]db.ChannelInstallation, error) { diff --git a/server/internal/integrations/slack/install_test.go b/server/internal/integrations/slack/install_test.go index 8881f76062..2f23e83734 100644 --- a/server/internal/integrations/slack/install_test.go +++ b/server/internal/integrations/slack/install_test.go @@ -41,16 +41,46 @@ type fakeInstallQueries struct { existing *db.ChannelInstallation // appIDTaken makes UpsertChannelInstallation report a unique-constraint // violation on the (channel_type, app_id) routing index — i.e. the pasted app - // is already connected to another agent / workspace. + // is already connected to a LIVE owner (the reclaim has run by then). appIDTaken bool upsertParams db.UpsertChannelInstallationParams upsertCalled bool rowID pgtype.UUID + + // reclaimedID, when set, is returned by ReclaimDeadChannelInstallationByAppID + // to model a dead prior owner having been cleared; otherwise it reports + // pgx.ErrNoRows (nothing was dead). reclaimCalled records that the install + // path ran the reclaim before upserting. + reclaimedID *pgtype.UUID + reclaimCalled bool + // ownerWorkspaceID / ownerArchived / ownerMissing drive the live-owner lookup + // that classifies an appIDTaken conflict into the right sentinel. + ownerWorkspaceID pgtype.UUID + ownerArchived bool + ownerMissing bool } // WithTx returns the same fake — the fake tx is a no-op token. func (f *fakeInstallQueries) WithTx(_ pgx.Tx) installQueries { return f } +func (f *fakeInstallQueries) ReclaimDeadChannelInstallationByAppID(_ context.Context, _ db.ReclaimDeadChannelInstallationByAppIDParams) (pgtype.UUID, error) { + f.reclaimCalled = true + if f.reclaimedID != nil { + return *f.reclaimedID, nil + } + return pgtype.UUID{}, pgx.ErrNoRows +} + +func (f *fakeInstallQueries) GetChannelInstallationOwnerByAppID(_ context.Context, _ db.GetChannelInstallationOwnerByAppIDParams) (db.GetChannelInstallationOwnerByAppIDRow, error) { + if f.ownerMissing { + return db.GetChannelInstallationOwnerByAppIDRow{}, pgx.ErrNoRows + } + return db.GetChannelInstallationOwnerByAppIDRow{ + WorkspaceID: f.ownerWorkspaceID, + AgentArchivedAt: pgtype.Timestamptz{Valid: f.ownerArchived}, + }, nil +} + func (f *fakeInstallQueries) UpsertChannelInstallation(_ context.Context, arg db.UpsertChannelInstallationParams) (db.ChannelInstallation, error) { f.upsertCalled = true f.upsertParams = arg diff --git a/server/pkg/db/generated/channel.sql.go b/server/pkg/db/generated/channel.sql.go index edd1439bc8..817bfb9a30 100644 --- a/server/pkg/db/generated/channel.sql.go +++ b/server/pkg/db/generated/channel.sql.go @@ -400,6 +400,41 @@ func (q *Queries) DeleteChannelChatSessionBindingsByInstallation(ctx context.Con return err } +const deleteChannelInstallationsByArchivedRuntimeAgents = `-- name: DeleteChannelInstallationsByArchivedRuntimeAgents :exec +WITH doomed AS ( + SELECT id FROM channel_installation + WHERE agent_id IN ( + SELECT id FROM agent WHERE runtime_id = $1 AND archived_at IS NOT NULL + ) +), +cleared_chat_sessions AS ( + DELETE FROM channel_chat_session_binding WHERE installation_id IN (SELECT id FROM doomed) +), +cleared_binding_tokens AS ( + DELETE FROM channel_binding_token WHERE installation_id IN (SELECT id FROM doomed) +), +cleared_user_bindings AS ( + DELETE FROM channel_user_binding WHERE installation_id IN (SELECT id FROM doomed) +), +detached_audit AS ( + UPDATE channel_inbound_audit SET installation_id = NULL WHERE installation_id IN (SELECT id FROM doomed) +) +DELETE FROM channel_installation WHERE id IN (SELECT id FROM doomed) +` + +// Application-layer replacement for the (deliberately absent, MUL-3515 §4) +// workspace/agent ON DELETE CASCADE: on runtime teardown, before the archived +// agents are hard-deleted, remove every channel installation they own — plus all +// of each installation's dependent rows — so no orphaned installation keeps +// occupying its bot's (channel_type, app_id) routing slot after its agent is gone +// (#4810). MUST run in the same tx as, and BEFORE, DeleteArchivedAgentsByRuntime. +// Mirrors the agent hard-delete predicate (runtime_id, archived_at IS NOT NULL) +// exactly. +func (q *Queries) DeleteChannelInstallationsByArchivedRuntimeAgents(ctx context.Context, runtimeID pgtype.UUID) error { + _, err := q.db.Exec(ctx, deleteChannelInstallationsByArchivedRuntimeAgents, runtimeID) + return err +} + const deleteChannelUserBindingsByInstallation = `-- name: DeleteChannelUserBindingsByInstallation :exec DELETE FROM channel_user_binding WHERE installation_id = $1 @@ -436,61 +471,6 @@ func (q *Queries) DeleteChannelUserBindingsByWorkspaceMember(ctx context.Context return err } -const deleteRevokedChannelInstallationByAppID = `-- name: DeleteRevokedChannelInstallationByAppID :one -DELETE FROM channel_installation -WHERE channel_type = $1 - AND config ->> 'app_id' = $2::text - AND workspace_id = $3 - AND agent_id <> $4 - AND status = 'revoked' -RETURNING id -` - -type DeleteRevokedChannelInstallationByAppIDParams struct { - ChannelType string `json:"channel_type"` - AppID string `json:"app_id"` - WorkspaceID pgtype.UUID `json:"workspace_id"` - AgentID pgtype.UUID `json:"agent_id"` -} - -// Hard-delete a REVOKED installation that belongs to a DIFFERENT agent, keyed -// by its platform app identity, and RETURN the deleted id. When a Feishu/Lark -// Bot is disconnected from agent A (status → 'revoked') and the same Bot (same -// app_id) is later bound to a DIFFERENT agent B, agent A's revoked row still -// occupies the (channel_type, config->>'app_id') unique slot and blocks the -// UpsertChannelInstallation INSERT for B. Removing that placeholder first lets -// the upsert create a fresh row for the new agent. -// -// The `agent_id <> arg` fence is load-bearing: re-connecting the SAME agent -// must NOT delete its own revoked row. UpsertChannelInstallation conflicts on -// (workspace_id, agent_id, channel_type), so the same agent's revoked row is -// reactivated in place (status → 'active') with its installation_id — and every -// channel_user_binding / channel_chat_session_binding hanging off it — -// preserved. Deleting it here would force an INSERT with a fresh installation_id -// and orphan all of that agent's member links and chat sessions. -// -// Fenced to one workspace AND status='revoked' so an active installation can -// never be silently deleted through this path. -// -// This DELETE is the atomic gate for the whole rebind cleanup: the caller keys -// its dependent-row cleanup off the RETURNING id, so cleanup runs ONLY for a row -// this statement actually removed. Under READ COMMITTED, a concurrent same-agent -// reconnect that reactivates the row to 'active' first makes the WHERE re-check -// fail (EvalPlanQual), this deletes nothing (pgx.ErrNoRows), and no dependents -// are touched — closing the read-then-delete TOCTOU where a stale "it was -// revoked" read could wipe the bindings of a since-reactivated installation. -func (q *Queries) DeleteRevokedChannelInstallationByAppID(ctx context.Context, arg DeleteRevokedChannelInstallationByAppIDParams) (pgtype.UUID, error) { - row := q.db.QueryRow(ctx, deleteRevokedChannelInstallationByAppID, - arg.ChannelType, - arg.AppID, - arg.WorkspaceID, - arg.AgentID, - ) - var id pgtype.UUID - err := row.Scan(&id) - return id, err -} - const findReusableChannelUserBinding = `-- name: FindReusableChannelUserBinding :one SELECT b.id, b.workspace_id, b.multica_user_id, b.installation_id, b.channel_type, b.channel_user_id, b.config, b.bound_at FROM channel_user_binding b JOIN channel_installation ci ON ci.id = b.installation_id @@ -711,6 +691,42 @@ func (q *Queries) GetChannelInstallationInWorkspace(ctx context.Context, arg Get return i, err } +const getChannelInstallationOwnerByAppID = `-- name: GetChannelInstallationOwnerByAppID :one +SELECT ci.workspace_id, ci.agent_id, a.archived_at AS agent_archived_at +FROM channel_installation ci +JOIN agent a ON a.id = ci.agent_id +WHERE ci.channel_type = $1 + AND ci.config ->> 'app_id' = $2::text +` + +type GetChannelInstallationOwnerByAppIDParams struct { + ChannelType string `json:"channel_type"` + AppID string `json:"app_id"` +} + +type GetChannelInstallationOwnerByAppIDRow struct { + WorkspaceID pgtype.UUID `json:"workspace_id"` + AgentID pgtype.UUID `json:"agent_id"` + AgentArchivedAt pgtype.Timestamptz `json:"agent_archived_at"` +} + +// Identifies the LIVE owner of a (channel_type, config->>'app_id') routing slot +// so the install path can refuse a rebind with an ACCURATE message instead of the +// old catch-all "connected to a different Multica workspace". Meant to be read +// only after ReclaimDeadChannelInstallationByAppID has removed every DEAD owner, +// so a returned row is a live active owner. `agent_archived` distinguishes an +// archived (reversible) owner — its bot stays owned, recovered by unarchiving the +// agent or disconnecting the bot — from a plain active one. The JOIN drops a row +// whose agent no longer exists (an orphan the reclaim gate should already have +// cleared), so a missing row (pgx.ErrNoRows) means "no live owner". The caller +// reads agent_archived_at.Valid to tell an archived (reversible) owner apart. +func (q *Queries) GetChannelInstallationOwnerByAppID(ctx context.Context, arg GetChannelInstallationOwnerByAppIDParams) (GetChannelInstallationOwnerByAppIDRow, error) { + row := q.db.QueryRow(ctx, getChannelInstallationOwnerByAppID, arg.ChannelType, arg.AppID) + var i GetChannelInstallationOwnerByAppIDRow + err := row.Scan(&i.WorkspaceID, &i.AgentID, &i.AgentArchivedAt) + return i, err +} + const getChannelOutboundCardByTask = `-- name: GetChannelOutboundCardByTask :one SELECT id, chat_session_id, task_id, channel_type, channel_chat_id, channel_card_message_id, status, last_patched_at, created_at FROM channel_outbound_card_message WHERE task_id = $1 @@ -1031,6 +1047,86 @@ func (q *Queries) PurgeExpiredChannelBindingTokens(ctx context.Context, expiresA return err } +const reclaimDeadChannelInstallationByAppID = `-- name: ReclaimDeadChannelInstallationByAppID :one +WITH dead AS ( + DELETE FROM channel_installation ci + WHERE ci.channel_type = $1 + AND ci.config ->> 'app_id' = $2::text + AND ( + (ci.status = 'revoked' + AND ci.workspace_id = $3 + AND ci.agent_id <> $4) + OR NOT EXISTS (SELECT 1 FROM workspace w WHERE w.id = ci.workspace_id) + OR NOT EXISTS (SELECT 1 FROM agent a WHERE a.id = ci.agent_id) + ) + RETURNING ci.id +), +cleared_chat_sessions AS ( + DELETE FROM channel_chat_session_binding + WHERE installation_id IN (SELECT id FROM dead) +), +cleared_binding_tokens AS ( + DELETE FROM channel_binding_token + WHERE installation_id IN (SELECT id FROM dead) +), +cleared_user_bindings AS ( + DELETE FROM channel_user_binding + WHERE installation_id IN (SELECT id FROM dead) +), +detached_audit AS ( + UPDATE channel_inbound_audit SET installation_id = NULL + WHERE installation_id IN (SELECT id FROM dead) +) +SELECT id FROM dead +` + +type ReclaimDeadChannelInstallationByAppIDParams struct { + ChannelType string `json:"channel_type"` + AppID string `json:"app_id"` + WorkspaceID pgtype.UUID `json:"workspace_id"` + AgentID pgtype.UUID `json:"agent_id"` +} + +// Rebind cleanup gate. Frees the (channel_type, config->>'app_id') routing slot +// so a valid new agent can (re)bind a bot whose previous owner is DEAD, and, in +// the same statement, clears every application-owned dependent row of the removed +// installation (channel_* has no FK/cascade, MUL-3515 §4). Returns the removed id +// (pgx.ErrNoRows when nothing was dead — a no-op the caller treats as success). +// +// "Dead" is exactly one of: +// 1. a REVOKED placeholder left by a DIFFERENT agent in the SAME workspace +// (Disconnect-then-rebind: agent A revoked, the same bot rebinds to agent B); +// 2. an ORPHAN whose owning workspace OR agent row no longer exists — the +// workspace was deleted, or the agent was hard-deleted on runtime teardown. +// With no FK the installation outlives its owner and keeps occupying the +// app_id slot: the "ghost binding" that made a bot un-rebindable (#4810). +// +// Deliberately NOT dead (the caller refuses these with an accurate conflict): +// - the SAME agent's own revoked row (agent_id = @agent_id): the upsert +// reactivates it in place, preserving its installation_id and every binding; +// - a live ACTIVE owner whose agent still exists — INCLUDING an ARCHIVED agent: +// archive is reversible, so its bot stays owned rather than being silently +// stolen. Only a hard delete frees the slot. +// +// The guard lives in the DELETE predicate (not a prior SELECT) so under READ +// COMMITTED the row is re-checked at execution (EvalPlanQual): a concurrent +// same-agent reconnect that flips the revoked row back to 'active' first makes +// the predicate re-check fail, this deletes nothing, and no dependents are +// touched — closing the read-then-delete TOCTOU. Dependent cleanup keys off the +// actually-deleted id (the `dead` CTE), so it runs ONLY for a row this statement +// removed. The (channel_type, app_id) unique index guarantees at most one match. +func (q *Queries) ReclaimDeadChannelInstallationByAppID(ctx context.Context, arg ReclaimDeadChannelInstallationByAppIDParams) (pgtype.UUID, error) { + row := q.db.QueryRow(ctx, reclaimDeadChannelInstallationByAppID, + arg.ChannelType, + arg.AppID, + arg.WorkspaceID, + arg.AgentID, + ) + var id pgtype.UUID + err := row.Scan(&id) + return id, err +} + const recordChannelInboundDrop = `-- name: RecordChannelInboundDrop :exec INSERT INTO channel_inbound_audit ( diff --git a/server/pkg/db/generated/workspace.sql.go b/server/pkg/db/generated/workspace.sql.go index a89661bdba..f8cfaf26d9 100644 --- a/server/pkg/db/generated/workspace.sql.go +++ b/server/pkg/db/generated/workspace.sql.go @@ -52,12 +52,37 @@ func (q *Queries) CreateWorkspace(ctx context.Context, arg CreateWorkspaceParams } const deleteWorkspace = `-- name: DeleteWorkspace :exec -WITH deleted_pending_check_suites AS ( +WITH ws_installations AS ( + SELECT id FROM channel_installation WHERE workspace_id = $1 +), +cleared_chat_sessions AS ( + DELETE FROM channel_chat_session_binding WHERE installation_id IN (SELECT id FROM ws_installations) +), +detached_audit AS ( + UPDATE channel_inbound_audit SET installation_id = NULL WHERE installation_id IN (SELECT id FROM ws_installations) +), +cleared_user_bindings AS ( + DELETE FROM channel_user_binding WHERE workspace_id = $1 +), +cleared_binding_tokens AS ( + DELETE FROM channel_binding_token WHERE workspace_id = $1 +), +cleared_installations AS ( + DELETE FROM channel_installation WHERE workspace_id = $1 +), +deleted_pending_check_suites AS ( DELETE FROM github_pending_check_suite WHERE workspace_id = $1 ) -DELETE FROM workspace WHERE id = $1 +DELETE FROM workspace WHERE workspace.id = $1 ` +// The channel_* tables carry NO FK to workspace (MUL-3515 §4), so — unlike the +// CASCADE-backed tables the DELETE below sweeps — they are not cleaned up +// implicitly. Remove this workspace's channel installations, and every dependent +// row of each, here so a deleted workspace never leaves an orphaned installation +// occupying its bot's (channel_type, config->>'app_id') routing slot, which would +// make that bot un-rebindable anywhere until an operator hand-deletes the row +// (#4810). All in one statement so it commits atomically with the workspace row. func (q *Queries) DeleteWorkspace(ctx context.Context, id pgtype.UUID) error { _, err := q.db.Exec(ctx, deleteWorkspace, id) return err diff --git a/server/pkg/db/queries/channel.sql b/server/pkg/db/queries/channel.sql index b95bbeb5c7..4d5bee63de 100644 --- a/server/pkg/db/queries/channel.sql +++ b/server/pkg/db/queries/channel.sql @@ -98,40 +98,111 @@ SELECT * FROM channel_installation WHERE channel_type = sqlc.arg('channel_type') AND config ->> 'app_id' = sqlc.arg('app_id')::text; --- name: DeleteRevokedChannelInstallationByAppID :one --- Hard-delete a REVOKED installation that belongs to a DIFFERENT agent, keyed --- by its platform app identity, and RETURN the deleted id. When a Feishu/Lark --- Bot is disconnected from agent A (status → 'revoked') and the same Bot (same --- app_id) is later bound to a DIFFERENT agent B, agent A's revoked row still --- occupies the (channel_type, config->>'app_id') unique slot and blocks the --- UpsertChannelInstallation INSERT for B. Removing that placeholder first lets --- the upsert create a fresh row for the new agent. +-- name: GetChannelInstallationOwnerByAppID :one +-- Identifies the LIVE owner of a (channel_type, config->>'app_id') routing slot +-- so the install path can refuse a rebind with an ACCURATE message instead of the +-- old catch-all "connected to a different Multica workspace". Meant to be read +-- only after ReclaimDeadChannelInstallationByAppID has removed every DEAD owner, +-- so a returned row is a live active owner. `agent_archived` distinguishes an +-- archived (reversible) owner — its bot stays owned, recovered by unarchiving the +-- agent or disconnecting the bot — from a plain active one. The JOIN drops a row +-- whose agent no longer exists (an orphan the reclaim gate should already have +-- cleared), so a missing row (pgx.ErrNoRows) means "no live owner". The caller +-- reads agent_archived_at.Valid to tell an archived (reversible) owner apart. +SELECT ci.workspace_id, ci.agent_id, a.archived_at AS agent_archived_at +FROM channel_installation ci +JOIN agent a ON a.id = ci.agent_id +WHERE ci.channel_type = sqlc.arg('channel_type') + AND ci.config ->> 'app_id' = sqlc.arg('app_id')::text; + +-- name: ReclaimDeadChannelInstallationByAppID :one +-- Rebind cleanup gate. Frees the (channel_type, config->>'app_id') routing slot +-- so a valid new agent can (re)bind a bot whose previous owner is DEAD, and, in +-- the same statement, clears every application-owned dependent row of the removed +-- installation (channel_* has no FK/cascade, MUL-3515 §4). Returns the removed id +-- (pgx.ErrNoRows when nothing was dead — a no-op the caller treats as success). -- --- The `agent_id <> arg` fence is load-bearing: re-connecting the SAME agent --- must NOT delete its own revoked row. UpsertChannelInstallation conflicts on --- (workspace_id, agent_id, channel_type), so the same agent's revoked row is --- reactivated in place (status → 'active') with its installation_id — and every --- channel_user_binding / channel_chat_session_binding hanging off it — --- preserved. Deleting it here would force an INSERT with a fresh installation_id --- and orphan all of that agent's member links and chat sessions. +-- "Dead" is exactly one of: +-- 1. a REVOKED placeholder left by a DIFFERENT agent in the SAME workspace +-- (Disconnect-then-rebind: agent A revoked, the same bot rebinds to agent B); +-- 2. an ORPHAN whose owning workspace OR agent row no longer exists — the +-- workspace was deleted, or the agent was hard-deleted on runtime teardown. +-- With no FK the installation outlives its owner and keeps occupying the +-- app_id slot: the "ghost binding" that made a bot un-rebindable (#4810). -- --- Fenced to one workspace AND status='revoked' so an active installation can --- never be silently deleted through this path. +-- Deliberately NOT dead (the caller refuses these with an accurate conflict): +-- - the SAME agent's own revoked row (agent_id = @agent_id): the upsert +-- reactivates it in place, preserving its installation_id and every binding; +-- - a live ACTIVE owner whose agent still exists — INCLUDING an ARCHIVED agent: +-- archive is reversible, so its bot stays owned rather than being silently +-- stolen. Only a hard delete frees the slot. -- --- This DELETE is the atomic gate for the whole rebind cleanup: the caller keys --- its dependent-row cleanup off the RETURNING id, so cleanup runs ONLY for a row --- this statement actually removed. Under READ COMMITTED, a concurrent same-agent --- reconnect that reactivates the row to 'active' first makes the WHERE re-check --- fail (EvalPlanQual), this deletes nothing (pgx.ErrNoRows), and no dependents --- are touched — closing the read-then-delete TOCTOU where a stale "it was --- revoked" read could wipe the bindings of a since-reactivated installation. -DELETE FROM channel_installation -WHERE channel_type = sqlc.arg('channel_type') - AND config ->> 'app_id' = sqlc.arg('app_id')::text - AND workspace_id = sqlc.arg('workspace_id') - AND agent_id <> sqlc.arg('agent_id') - AND status = 'revoked' -RETURNING id; +-- The guard lives in the DELETE predicate (not a prior SELECT) so under READ +-- COMMITTED the row is re-checked at execution (EvalPlanQual): a concurrent +-- same-agent reconnect that flips the revoked row back to 'active' first makes +-- the predicate re-check fail, this deletes nothing, and no dependents are +-- touched — closing the read-then-delete TOCTOU. Dependent cleanup keys off the +-- actually-deleted id (the `dead` CTE), so it runs ONLY for a row this statement +-- removed. The (channel_type, app_id) unique index guarantees at most one match. +WITH dead AS ( + DELETE FROM channel_installation ci + WHERE ci.channel_type = sqlc.arg('channel_type') + AND ci.config ->> 'app_id' = sqlc.arg('app_id')::text + AND ( + (ci.status = 'revoked' + AND ci.workspace_id = sqlc.arg('workspace_id') + AND ci.agent_id <> sqlc.arg('agent_id')) + OR NOT EXISTS (SELECT 1 FROM workspace w WHERE w.id = ci.workspace_id) + OR NOT EXISTS (SELECT 1 FROM agent a WHERE a.id = ci.agent_id) + ) + RETURNING ci.id +), +cleared_chat_sessions AS ( + DELETE FROM channel_chat_session_binding + WHERE installation_id IN (SELECT id FROM dead) +), +cleared_binding_tokens AS ( + DELETE FROM channel_binding_token + WHERE installation_id IN (SELECT id FROM dead) +), +cleared_user_bindings AS ( + DELETE FROM channel_user_binding + WHERE installation_id IN (SELECT id FROM dead) +), +detached_audit AS ( + UPDATE channel_inbound_audit SET installation_id = NULL + WHERE installation_id IN (SELECT id FROM dead) +) +SELECT id FROM dead; + +-- name: DeleteChannelInstallationsByArchivedRuntimeAgents :exec +-- Application-layer replacement for the (deliberately absent, MUL-3515 §4) +-- workspace/agent ON DELETE CASCADE: on runtime teardown, before the archived +-- agents are hard-deleted, remove every channel installation they own — plus all +-- of each installation's dependent rows — so no orphaned installation keeps +-- occupying its bot's (channel_type, app_id) routing slot after its agent is gone +-- (#4810). MUST run in the same tx as, and BEFORE, DeleteArchivedAgentsByRuntime. +-- Mirrors the agent hard-delete predicate (runtime_id, archived_at IS NOT NULL) +-- exactly. +WITH doomed AS ( + SELECT id FROM channel_installation + WHERE agent_id IN ( + SELECT id FROM agent WHERE runtime_id = sqlc.arg('runtime_id') AND archived_at IS NOT NULL + ) +), +cleared_chat_sessions AS ( + DELETE FROM channel_chat_session_binding WHERE installation_id IN (SELECT id FROM doomed) +), +cleared_binding_tokens AS ( + DELETE FROM channel_binding_token WHERE installation_id IN (SELECT id FROM doomed) +), +cleared_user_bindings AS ( + DELETE FROM channel_user_binding WHERE installation_id IN (SELECT id FROM doomed) +), +detached_audit AS ( + UPDATE channel_inbound_audit SET installation_id = NULL WHERE installation_id IN (SELECT id FROM doomed) +) +DELETE FROM channel_installation WHERE id IN (SELECT id FROM doomed); -- name: ListChannelInstallationsByWorkspace :many -- Scoped by channel_type so a per-channel management surface (e.g. the Lark diff --git a/server/pkg/db/queries/workspace.sql b/server/pkg/db/queries/workspace.sql index 3f9daa0dd5..a567712c28 100644 --- a/server/pkg/db/queries/workspace.sql +++ b/server/pkg/db/queries/workspace.sql @@ -46,7 +46,32 @@ WHERE id = $1 RETURNING issue_counter; -- name: DeleteWorkspace :exec -WITH deleted_pending_check_suites AS ( +-- The channel_* tables carry NO FK to workspace (MUL-3515 §4), so — unlike the +-- CASCADE-backed tables the DELETE below sweeps — they are not cleaned up +-- implicitly. Remove this workspace's channel installations, and every dependent +-- row of each, here so a deleted workspace never leaves an orphaned installation +-- occupying its bot's (channel_type, config->>'app_id') routing slot, which would +-- make that bot un-rebindable anywhere until an operator hand-deletes the row +-- (#4810). All in one statement so it commits atomically with the workspace row. +WITH ws_installations AS ( + SELECT id FROM channel_installation WHERE workspace_id = $1 +), +cleared_chat_sessions AS ( + DELETE FROM channel_chat_session_binding WHERE installation_id IN (SELECT id FROM ws_installations) +), +detached_audit AS ( + UPDATE channel_inbound_audit SET installation_id = NULL WHERE installation_id IN (SELECT id FROM ws_installations) +), +cleared_user_bindings AS ( + DELETE FROM channel_user_binding WHERE workspace_id = $1 +), +cleared_binding_tokens AS ( + DELETE FROM channel_binding_token WHERE workspace_id = $1 +), +cleared_installations AS ( + DELETE FROM channel_installation WHERE workspace_id = $1 +), +deleted_pending_check_suites AS ( DELETE FROM github_pending_check_suite WHERE workspace_id = $1 ) -DELETE FROM workspace WHERE id = $1; +DELETE FROM workspace WHERE workspace.id = $1;