[codex] Fix autoscan plugin bindings and poll status (#75)

* fix(autoscan): bind sources by plugin id

* fix(autoscan): skip overlapping source polls

cmd/silo/main.go

@@ -1499,6 +1499,9 @@ func main() {
 		taskMgr.Register(tasks.NewReconcileRequestsTask(requestReconcileSvc, 100))
 		if deps.FolderRepo != nil && deps.LibraryScanQueue != nil && pluginService != nil && pluginInstallationStore != nil {
 			autoscanRepo := autoscan.NewRepository(deps.DB)
+			if err := autoscanRepo.MarkInterruptedEvents(appCtx); err != nil {
+				slog.Warn("autoscan: failed to mark interrupted polls", "err", err)
+			}
 			autoscanSvc := api.BuildAutoscanService(
 				autoscanRepo,
 				pluginService,

internal/api/autoscan_wiring.go

@@ -64,8 +64,8 @@ type PluginScanSourceAdapter struct {
 	Svc *plugins.Service
 }
 
-func (a PluginScanSourceAdapter) ScanSourceClient(ctx context.Context, installationID int, capabilityID string) (autoscan.PollChangesClient, error) {
-	return a.Svc.ScanSourceClient(ctx, installationID, capabilityID)
+func (a PluginScanSourceAdapter) ScanSourceClient(ctx context.Context, pluginID, capabilityID string) (autoscan.PollChangesClient, error) {
+	return a.Svc.ScanSourceClientByPluginID(ctx, pluginID, capabilityID)
 }
 
 // scanSourceCapabilityType is the plugin capability type autoscan discovery
@@ -75,15 +75,6 @@ const scanSourceCapabilityType = "scan_source.v1"
 // PluginScanSourceLister adapts the plugin installation store to
 // autoscan.ScanSourceLister: it enumerates every installed scan_source.v1
 // capability across ALL installed plugins, regardless of enabled state.
-//
-// Using List (not ListEnabled) is deliberate for orphan detection: a temporarily
-// DISABLED-but-installed plugin must NOT be treated as orphaned. If we dropped
-// disabled plugins here, PollOnce would prune their sources and skip them with no
-// last_error — they'd vanish silently. Keeping them present means PollOnce still
-// attempts them; the plugin client fails to load (not running) and the source
-// gets a visible RecordError instead. Only a fully UNINSTALLED plugin (gone from
-// List entirely) is a true orphan. The Add-source picker shares this all-installed
-// set, which is fine — operators can bind sources against an installed capability.
 type PluginScanSourceLister struct {
 	Store *plugins.InstallationStore
 }
@@ -104,10 +95,9 @@ func (l PluginScanSourceLister) ListScanSources(ctx context.Context) ([]autoscan
 				continue
 			}
 			out = append(out, autoscan.DiscoveredSource{
-				InstallationID: c.InstallationID,
-				CapabilityID:   c.ID,
-				PluginID:       inst.PluginID,
-				DisplayName:    scanSourceDisplayName(inst.PluginID, c),
+				PluginID:     inst.PluginID,
+				CapabilityID: c.ID,
+				DisplayName:  scanSourceDisplayName(inst.PluginID, c),
 			})
 		}
 	}

internal/api/handlers/autoscan.go

@@ -36,6 +36,7 @@ type autoscanStore interface {
 	CountAutoscanScans(ctx context.Context, filter autoscan.ScanListFilter) (int, error)
 	ListEvents(ctx context.Context, filter autoscan.EventListFilter) ([]autoscan.EventWithRuns, error)
 	CountEvents(ctx context.Context, filter autoscan.EventListFilter) (int, error)
+	ListRunningEvents(ctx context.Context) ([]autoscan.Event, error)
 	GetQueueSummary(ctx context.Context) (autoscan.QueueSummary, error)
 	LatestEventAt(ctx context.Context) (*time.Time, error)
 }
@@ -309,7 +310,7 @@ func (h *AutoscanHandler) HandleDeleteConnection(w http.ResponseWriter, r *http.
 // credentials: the connection link is by id only.
 type autoscanSourceResponse struct {
 	ID                  string                 `json:"id"`
-	InstallationID      int                    `json:"installation_id"`
+	PluginID            string                 `json:"plugin_id"`
 	CapabilityID        string                 `json:"capability_id"`
 	ConnectionID        *string                `json:"connection_id"`
 	Enabled             bool                   `json:"enabled"`
@@ -331,7 +332,7 @@ func sourceResponse(s autoscan.Source) autoscanSourceResponse {
 	config := normalizeSourceConfig(s.SourceConfig)
 	return autoscanSourceResponse{
 		ID:                  s.ID,
-		InstallationID:      s.InstallationID,
+		PluginID:            s.PluginID,
 		CapabilityID:        s.CapabilityID,
 		ConnectionID:        s.ConnectionID,
 		Enabled:             s.Enabled,
@@ -364,10 +365,9 @@ func (h *AutoscanHandler) HandleListSources(w http.ResponseWriter, r *http.Reque
 // --- Available scan-source plugins (Add-source picker) ---
 
 type autoscanScanSourcePluginResponse struct {
-	InstallationID int    `json:"installation_id"`
-	CapabilityID   string `json:"capability_id"`
-	PluginID       string `json:"plugin_id"`
-	DisplayName    string `json:"display_name"`
+	PluginID     string `json:"plugin_id"`
+	CapabilityID string `json:"capability_id"`
+	DisplayName  string `json:"display_name"`
 }
 
 // HandleListAvailableScanSources returns every installed scan_source capability
@@ -381,10 +381,9 @@ func (h *AutoscanHandler) HandleListAvailableScanSources(w http.ResponseWriter,
 	out := make([]autoscanScanSourcePluginResponse, 0, len(available))
 	for _, a := range available {
 		out = append(out, autoscanScanSourcePluginResponse{
-			InstallationID: a.InstallationID,
-			CapabilityID:   a.CapabilityID,
-			PluginID:       a.PluginID,
-			DisplayName:    a.DisplayName,
+			PluginID:     a.PluginID,
+			CapabilityID: a.CapabilityID,
+			DisplayName:  a.DisplayName,
 		})
 	}
 	writeJSON(w, http.StatusOK, struct {
@@ -394,12 +393,12 @@ func (h *AutoscanHandler) HandleListAvailableScanSources(w http.ResponseWriter,
 
 // --- Create source ---
 
-// autoscanCreateSourceInput is the create payload: it names the installed
-// scan_source capability to bind (installation_id + capability_id) plus the
-// initial binding/scheduling fields. Unlike the update payload, identity is
-// supplied here rather than read from an existing row.
+// autoscanCreateSourceInput is the create payload: it names the scan_source
+// capability to bind (plugin_id + capability_id) plus the initial
+// binding/scheduling fields. Unlike the update payload, identity is supplied
+// here rather than read from an existing row.
 type autoscanCreateSourceInput struct {
-	InstallationID      int                    `json:"installation_id"`
+	PluginID            string                 `json:"plugin_id"`
 	CapabilityID        string                 `json:"capability_id"`
 	ConnectionID        *string                `json:"connection_id"`
 	Enabled             bool                   `json:"enabled"`
@@ -420,6 +419,11 @@ func (h *AutoscanHandler) HandleCreateSource(w http.ResponseWriter, r *http.Requ
 		return
 	}
 	capID := strings.TrimSpace(in.CapabilityID)
+	pluginID := strings.TrimSpace(in.PluginID)
+	if pluginID == "" {
+		writeError(w, http.StatusBadRequest, "bad_request", "plugin_id is required")
+		return
+	}
 	if capID == "" {
 		writeError(w, http.StatusBadRequest, "bad_request", "capability_id is required")
 		return
@@ -432,7 +436,7 @@ func (h *AutoscanHandler) HandleCreateSource(w http.ResponseWriter, r *http.Requ
 		writeError(w, http.StatusBadRequest, "bad_request", err.Error())
 		return
 	}
-	// Validate the (installation_id, capability_id) is a currently-installed
+	// Validate the (plugin_id, capability_id) is a currently-installed
 	// scan_source capability — a source may only be created against an installed
 	// plugin capability.
 	available, err := h.svc.ListAvailableScanSources(r.Context())
@@ -440,13 +444,13 @@ func (h *AutoscanHandler) HandleCreateSource(w http.ResponseWriter, r *http.Requ
 		writeAutoscanError(w, err)
 		return
 	}
-	if !scanSourceInstalled(available, in.InstallationID, capID) {
-		writeError(w, http.StatusBadRequest, "bad_request", "installation_id + capability_id is not a currently-installed scan_source capability")
+	if !scanSourceInstalled(available, pluginID, capID) {
+		writeError(w, http.StatusBadRequest, "bad_request", "plugin_id + capability_id is not a currently-installed scan_source capability")
 		return
 	}
 	connArg := normalizeConnectionID(in.ConnectionID)
 	created, err := h.repo.CreateSource(r.Context(), autoscan.Source{
-		InstallationID:      in.InstallationID,
+		PluginID:            pluginID,
 		CapabilityID:        capID,
 		ConnectionID:        connArg,
 		Enabled:             in.Enabled,
@@ -462,18 +466,18 @@ func (h *AutoscanHandler) HandleCreateSource(w http.ResponseWriter, r *http.Requ
 	writeJSON(w, http.StatusCreated, sourceResponse(created))
 }
 
-// scanSourceInstalled reports whether (installationID, capabilityID) is in the
+// scanSourceInstalled reports whether (pluginID, capabilityID) is in the
 // set of currently-installed scan_source capabilities.
-func scanSourceInstalled(available []autoscan.AvailableScanSource, installationID int, capabilityID string) bool {
+func scanSourceInstalled(available []autoscan.AvailableScanSource, pluginID, capabilityID string) bool {
 	for _, a := range available {
-		if a.InstallationID == installationID && a.CapabilityID == capabilityID {
+		if a.PluginID == pluginID && a.CapabilityID == capabilityID {
 			return true
 		}
 	}
 	return false
 }
 
-// autoscanSourceInput is the source write payload. The (installation_id,
+// autoscanSourceInput is the source write payload. The (plugin_id,
 // capability_id) identity is read from the existing source row, so only the
 // schedulable/binding fields are accepted.
 //
@@ -548,7 +552,7 @@ func (h *AutoscanHandler) HandleUpdateSource(w http.ResponseWriter, r *http.Requ
 	// connection_id is a full-state field: nil means unbind, a UUID string
 	// means bind. A whitespace-only string is normalised to nil (unbound).
 	connArg := normalizeConnectionID(in.ConnectionID)
-	// Update by id; identity (installation_id, capability_id) is immutable and
+	// Update by id; identity (plugin_id, capability_id) is immutable and
 	// preserved by the repo. A missing source maps to 404.
 	updated, err := h.repo.UpdateSource(r.Context(), autoscan.Source{
 		ID:                  id,
@@ -581,9 +585,8 @@ func normalizeConnectionID(in *string) *string {
 }
 
 // HandleDeleteSource removes a source row. This is the operator's escape hatch
-// for an orphaned source (one whose scan_source plugin was uninstalled): the
-// poll loop skips such rows quietly, and this endpoint clears them. An unknown
-// id maps to 404 via autoscan.ErrNotFound.
+// for an orphaned source (one whose scan_source plugin was uninstalled or no
+// longer resolves cleanly). An unknown id maps to 404 via autoscan.ErrNotFound.
 func (h *AutoscanHandler) HandleDeleteSource(w http.ResponseWriter, r *http.Request) {
 	id := strings.TrimSpace(chi.URLParam(r, "id"))
 	if err := h.repo.DeleteSource(r.Context(), id); err != nil {
@@ -710,7 +713,7 @@ type autoscanEventScanRunResponse struct {
 type autoscanEventResponse struct {
 	ID              int64                          `json:"id"`
 	SourceID        *string                        `json:"source_id"`
-	InstallationID  int                            `json:"installation_id"`
+	PluginID        string                         `json:"plugin_id"`
 	CapabilityID    string                         `json:"capability_id"`
 	StartedAt       time.Time                      `json:"started_at"`
 	CompletedAt     time.Time                      `json:"completed_at"`
@@ -753,7 +756,7 @@ func eventResponse(event autoscan.EventWithRuns) autoscanEventResponse {
 	return autoscanEventResponse{
 		ID:              e.ID,
 		SourceID:        e.SourceID,
-		InstallationID:  e.InstallationID,
+		PluginID:        e.PluginID,
 		CapabilityID:    e.CapabilityID,
 		StartedAt:       e.StartedAt,
 		CompletedAt:     e.CompletedAt,
@@ -794,9 +797,9 @@ func (h *AutoscanHandler) HandleListEvents(w http.ResponseWriter, r *http.Reques
 		filter.Offset = offset
 	}
 	switch filter.Status {
-	case "", autoscan.EventStatusSuccess, autoscan.EventStatusError, autoscan.EventStatusUnresolved:
+	case "", autoscan.EventStatusRunning, autoscan.EventStatusSuccess, autoscan.EventStatusError, autoscan.EventStatusUnresolved:
 	default:
-		writeError(w, http.StatusBadRequest, "bad_request", "status must be success, error, or unresolved")
+		writeError(w, http.StatusBadRequest, "bad_request", "status must be running, success, error, or unresolved")
 		return
 	}
 
@@ -835,7 +838,7 @@ type autoscanScanResponse struct {
 	CompletedAt      *time.Time `json:"completed_at,omitempty"`
 	AutoscanEventID  *int64     `json:"autoscan_event_id,omitempty"`
 	SourceID         *string    `json:"source_id,omitempty"`
-	InstallationID   *int       `json:"installation_id,omitempty"`
+	PluginID         string     `json:"plugin_id,omitempty"`
 	CapabilityID     string     `json:"capability_id,omitempty"`
 	EventStatus      string     `json:"event_status,omitempty"`
 	EventCompletedAt *time.Time `json:"event_completed_at,omitempty"`
@@ -862,7 +865,7 @@ func autoscanScanRunResponse(scan autoscan.ScanWithEvent) autoscanScanResponse {
 		CompletedAt:      scan.CompletedAt,
 		AutoscanEventID:  scan.AutoscanEventID,
 		SourceID:         scan.SourceID,
-		InstallationID:   scan.InstallationID,
+		PluginID:         scan.PluginID,
 		CapabilityID:     scan.CapabilityID,
 		EventStatus:      string(scan.EventStatus),
 		EventCompletedAt: scan.EventCompletedAt,
@@ -923,24 +926,35 @@ func (h *AutoscanHandler) HandleListScans(w http.ResponseWriter, r *http.Request
 // --- Status ---
 
 type autoscanStatusSource struct {
-	ID             string                 `json:"id"`
-	InstallationID int                    `json:"installation_id"`
-	CapabilityID   string                 `json:"capability_id"`
-	ConnectionID   *string                `json:"connection_id"`
-	Enabled        bool                   `json:"enabled"`
-	Label          string                 `json:"label"`
-	PathRewrites   []autoscan.PathRewrite `json:"path_rewrites"`
-	LastRunAt      *time.Time             `json:"last_run_at,omitempty"`
-	LastError      *string                `json:"last_error,omitempty"`
+	ID           string                 `json:"id"`
+	PluginID     string                 `json:"plugin_id"`
+	CapabilityID string                 `json:"capability_id"`
+	ConnectionID *string                `json:"connection_id"`
+	Enabled      bool                   `json:"enabled"`
+	Label        string                 `json:"label"`
+	PathRewrites []autoscan.PathRewrite `json:"path_rewrites"`
+	LastRunAt    *time.Time             `json:"last_run_at,omitempty"`
+	LastError    *string                `json:"last_error,omitempty"`
+}
+
+type autoscanRunningPollResponse struct {
+	ID           int64     `json:"id"`
+	SourceID     *string   `json:"source_id"`
+	PluginID     string    `json:"plugin_id"`
+	CapabilityID string    `json:"capability_id"`
+	StartedAt    time.Time `json:"started_at"`
+	ElapsedMS    int64     `json:"elapsed_ms"`
+	MarkerBefore *string   `json:"marker_before,omitempty"`
 }
 
 type autoscanStatusResponse struct {
-	Enabled       bool                   `json:"enabled"`
-	Sources       []autoscanStatusSource `json:"sources"`
-	ActiveScans   int                    `json:"active_scans"`
-	AcceptedScans int                    `json:"accepted_scans"`
-	RunningScans  int                    `json:"running_scans"`
-	LatestEventAt *time.Time             `json:"latest_event_at,omitempty"`
+	Enabled       bool                          `json:"enabled"`
+	Sources       []autoscanStatusSource        `json:"sources"`
+	RunningPolls  []autoscanRunningPollResponse `json:"running_polls"`
+	ActiveScans   int                           `json:"active_scans"`
+	AcceptedScans int                           `json:"accepted_scans"`
+	RunningScans  int                           `json:"running_scans"`
+	LatestEventAt *time.Time                    `json:"latest_event_at,omitempty"`
 }
 
 func (h *AutoscanHandler) HandleStatus(w http.ResponseWriter, r *http.Request) {
@@ -960,6 +974,11 @@ func (h *AutoscanHandler) HandleStatus(w http.ResponseWriter, r *http.Request) {
 		writeAutoscanError(w, err)
 		return
 	}
+	runningEvents, err := h.repo.ListRunningEvents(ctx)
+	if err != nil {
+		writeAutoscanError(w, err)
+		return
+	}
 	latestEventAt, err := h.repo.LatestEventAt(ctx)
 	if err != nil {
 		writeAutoscanError(w, err)
@@ -972,20 +991,38 @@ func (h *AutoscanHandler) HandleStatus(w http.ResponseWriter, r *http.Request) {
 			rewrites = []autoscan.PathRewrite{}
 		}
 		trimmed = append(trimmed, autoscanStatusSource{
-			ID:             src.ID,
-			InstallationID: src.InstallationID,
-			CapabilityID:   src.CapabilityID,
-			ConnectionID:   src.ConnectionID,
-			Enabled:        src.Enabled,
-			Label:          src.Label,
-			PathRewrites:   rewrites,
-			LastRunAt:      src.LastRunAt,
-			LastError:      src.LastError,
+			ID:           src.ID,
+			PluginID:     src.PluginID,
+			CapabilityID: src.CapabilityID,
+			ConnectionID: src.ConnectionID,
+			Enabled:      src.Enabled,
+			Label:        src.Label,
+			PathRewrites: rewrites,
+			LastRunAt:    src.LastRunAt,
+			LastError:    src.LastError,
+		})
+	}
+	now := time.Now()
+	runningPolls := make([]autoscanRunningPollResponse, 0, len(runningEvents))
+	for _, event := range runningEvents {
+		elapsed := now.Sub(event.StartedAt)
+		if elapsed < 0 {
+			elapsed = 0
+		}
+		runningPolls = append(runningPolls, autoscanRunningPollResponse{
+			ID:           event.ID,
+			SourceID:     event.SourceID,
+			PluginID:     event.PluginID,
+			CapabilityID: event.CapabilityID,
+			StartedAt:    event.StartedAt,
+			ElapsedMS:    elapsed.Milliseconds(),
+			MarkerBefore: event.MarkerBefore,
 		})
 	}
 	writeJSON(w, http.StatusOK, autoscanStatusResponse{
 		Enabled:       s.Enabled,
 		Sources:       trimmed,
+		RunningPolls:  runningPolls,
 		ActiveScans:   queue.Active,
 		AcceptedScans: queue.Accepted,
 		RunningScans:  queue.Running,

internal/api/handlers/autoscan_test.go

@@ -31,6 +31,7 @@ type fakeAutoscanStore struct {
 	countScansFn       func(autoscan.ScanListFilter) (int, error)
 	listEventsFn       func(autoscan.EventListFilter) ([]autoscan.EventWithRuns, error)
 	countEventsFn      func(autoscan.EventListFilter) (int, error)
+	listRunningFn      func() ([]autoscan.Event, error)
 	queueSummaryFn     func() (autoscan.QueueSummary, error)
 	latestEventAtFn    func() (*time.Time, error)
 }
@@ -140,6 +141,13 @@ func (f *fakeAutoscanStore) CountEvents(_ context.Context, filter autoscan.Event
 	return 0, nil
 }
 
+func (f *fakeAutoscanStore) ListRunningEvents(context.Context) ([]autoscan.Event, error) {
+	if f.listRunningFn != nil {
+		return f.listRunningFn()
+	}
+	return nil, nil
+}
+
 func (f *fakeAutoscanStore) GetQueueSummary(context.Context) (autoscan.QueueSummary, error) {
 	if f.queueSummaryFn != nil {
 		return f.queueSummaryFn()
@@ -273,7 +281,7 @@ func TestAutoscanHandleListSourcesListsOnly(t *testing.T) {
 	store := &fakeAutoscanStore{
 		listSourcesFn: func() ([]autoscan.Source, error) {
 			listed = true
-			return []autoscan.Source{{ID: "src-1", InstallationID: 1, CapabilityID: "arr"}}, nil
+			return []autoscan.Source{{ID: "src-1", PluginID: "silo.autoscan.arr", CapabilityID: "arr"}}, nil
 		},
 	}
 	h := NewAutoscanHandler(store, &fakeAutoscanTriggerer{})
@@ -300,7 +308,7 @@ func TestAutoscanHandleListSourcesListsOnly(t *testing.T) {
 
 func TestAutoscanHandleListAvailableScanSources(t *testing.T) {
 	trig := &fakeAutoscanTriggerer{available: []autoscan.AvailableScanSource{
-		{InstallationID: 1, CapabilityID: "arr", PluginID: "sonarr", DisplayName: "Sonarr"},
+		{PluginID: "sonarr", CapabilityID: "arr", DisplayName: "Sonarr"},
 	}}
 	h := NewAutoscanHandler(&fakeAutoscanStore{}, trig)
 
@@ -331,19 +339,19 @@ func TestAutoscanHandleCreateSourceSucceeds(t *testing.T) {
 		},
 	}
 	trig := &fakeAutoscanTriggerer{available: []autoscan.AvailableScanSource{
-		{InstallationID: 1, CapabilityID: "arr"},
+		{PluginID: "silo.autoscan.arr", CapabilityID: "arr"},
 	}}
 	h := NewAutoscanHandler(store, trig)
 
 	req := newAutoscanRequest("POST", "/api/v1/admin/autoscan/sources",
-		`{"installation_id":1,"capability_id":"arr","connection_id":"conn-1","enabled":true}`, "")
+		`{"plugin_id":"silo.autoscan.arr","capability_id":"arr","connection_id":"conn-1","enabled":true}`, "")
 	rec := httptest.NewRecorder()
 	h.HandleCreateSource(rec, req)
 
 	if rec.Code != http.StatusCreated {
 		t.Fatalf("status = %d, want 201; body=%s", rec.Code, rec.Body.String())
 	}
-	if got.InstallationID != 1 || got.CapabilityID != "arr" || got.ConnectionID == nil || *got.ConnectionID != "conn-1" {
+	if got.PluginID != "silo.autoscan.arr" || got.CapabilityID != "arr" || got.ConnectionID == nil || *got.ConnectionID != "conn-1" {
 		t.Fatalf("unexpected created source: %+v", got)
 	}
 	var body autoscanSourceResponse
@@ -363,14 +371,14 @@ func TestAutoscanHandleCreateSourceRejectsUnknownCapability(t *testing.T) {
 			return s, nil
 		},
 	}
-	// Installed set does NOT include (1, "arr").
+	// Installed set does NOT include ("silo.autoscan.arr", "arr").
 	trig := &fakeAutoscanTriggerer{available: []autoscan.AvailableScanSource{
-		{InstallationID: 2, CapabilityID: "other"},
+		{PluginID: "silo.autoscan.other", CapabilityID: "other"},
 	}}
 	h := NewAutoscanHandler(store, trig)
 
 	req := newAutoscanRequest("POST", "/api/v1/admin/autoscan/sources",
-		`{"installation_id":1,"capability_id":"arr","enabled":false}`, "")
+		`{"plugin_id":"silo.autoscan.arr","capability_id":"arr","enabled":false}`, "")
 	rec := httptest.NewRecorder()
 	h.HandleCreateSource(rec, req)
 
@@ -394,12 +402,12 @@ func TestAutoscanHandleCreateSourceEnableWithoutConnectionSucceeds(t *testing.T)
 		},
 	}
 	trig := &fakeAutoscanTriggerer{available: []autoscan.AvailableScanSource{
-		{InstallationID: 1, CapabilityID: "arr"},
+		{PluginID: "silo.autoscan.arr", CapabilityID: "arr"},
 	}}
 	h := NewAutoscanHandler(store, trig)
 
 	req := newAutoscanRequest("POST", "/api/v1/admin/autoscan/sources",
-		`{"installation_id":1,"capability_id":"arr","enabled":true}`, "")
+		`{"plugin_id":"silo.autoscan.arr","capability_id":"arr","enabled":true}`, "")
 	rec := httptest.NewRecorder()
 	h.HandleCreateSource(rec, req)
 
@@ -615,7 +623,7 @@ func TestAutoscanHandleUpdateSourceEnableWithoutConnectionSucceeds(t *testing.T)
 	var got autoscan.Source
 	store := &fakeAutoscanStore{
 		getSourceFn: func(id string) (autoscan.Source, error) {
-			return autoscan.Source{ID: id, InstallationID: 1, CapabilityID: "arr", ConnectionID: nil}, nil
+			return autoscan.Source{ID: id, PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: nil}, nil
 		},
 		updateSourceFn: func(s autoscan.Source) (autoscan.Source, error) {
 			got = s
@@ -641,7 +649,7 @@ func TestAutoscanHandleUpdateSourceEnableWithConnectionSucceeds(t *testing.T) {
 	var got autoscan.Source
 	store := &fakeAutoscanStore{
 		getSourceFn: func(id string) (autoscan.Source, error) {
-			return autoscan.Source{ID: id, InstallationID: 1, CapabilityID: "arr", ConnectionID: ptr("conn-1")}, nil
+			return autoscan.Source{ID: id, PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: ptr("conn-1")}, nil
 		},
 		updateSourceFn: func(s autoscan.Source) (autoscan.Source, error) {
 			got = s
@@ -669,7 +677,7 @@ func TestAutoscanHandleUpdateSourceBindConnectionSucceeds(t *testing.T) {
 	store := &fakeAutoscanStore{
 		getSourceFn: func(id string) (autoscan.Source, error) {
 			// Source starts with no connection.
-			return autoscan.Source{ID: id, InstallationID: 1, CapabilityID: "arr", ConnectionID: nil}, nil
+			return autoscan.Source{ID: id, PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: nil}, nil
 		},
 		updateSourceFn: func(s autoscan.Source) (autoscan.Source, error) {
 			got = s
@@ -697,7 +705,7 @@ func TestAutoscanHandleUpdateSourceUnbindConnectionSucceeds(t *testing.T) {
 	store := &fakeAutoscanStore{
 		getSourceFn: func(id string) (autoscan.Source, error) {
 			// Source starts with a connection already bound.
-			return autoscan.Source{ID: id, InstallationID: 1, CapabilityID: "arr", ConnectionID: ptr("conn-1")}, nil
+			return autoscan.Source{ID: id, PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: ptr("conn-1")}, nil
 		},
 		updateSourceFn: func(s autoscan.Source) (autoscan.Source, error) {
 			got = s
@@ -727,7 +735,7 @@ func TestAutoscanHandleUpdateSourceUnbindWhileEnabledSucceeds(t *testing.T) {
 	var got autoscan.Source
 	store := &fakeAutoscanStore{
 		getSourceFn: func(id string) (autoscan.Source, error) {
-			return autoscan.Source{ID: id, InstallationID: 1, CapabilityID: "arr", ConnectionID: ptr("conn-1")}, nil
+			return autoscan.Source{ID: id, PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: ptr("conn-1")}, nil
 		},
 		updateSourceFn: func(s autoscan.Source) (autoscan.Source, error) {
 			got = s
@@ -925,7 +933,7 @@ func TestAutoscanHandleUpdateSourceRoundTripsPathRewrites(t *testing.T) {
 	var got autoscan.Source
 	store := &fakeAutoscanStore{
 		getSourceFn: func(id string) (autoscan.Source, error) {
-			return autoscan.Source{ID: id, InstallationID: 1, CapabilityID: "arr", ConnectionID: ptr("conn-1")}, nil
+			return autoscan.Source{ID: id, PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: ptr("conn-1")}, nil
 		},
 		updateSourceFn: func(s autoscan.Source) (autoscan.Source, error) {
 			got = s
@@ -990,7 +998,7 @@ func TestAutoscanHandleUpdateSourceRejectsBlankRewrite(t *testing.T) {
 	upserted := false
 	store := &fakeAutoscanStore{
 		getSourceFn: func(id string) (autoscan.Source, error) {
-			return autoscan.Source{ID: id, InstallationID: 1, CapabilityID: "arr", ConnectionID: ptr("conn-1")}, nil
+			return autoscan.Source{ID: id, PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: ptr("conn-1")}, nil
 		},
 		updateSourceFn: func(s autoscan.Source) (autoscan.Source, error) {
 			upserted = true
@@ -1020,7 +1028,7 @@ func TestAutoscanHandleListScansSerializesFiltersAndEventContext(t *testing.T) {
 	eventCompleted := completed.Add(-1 * time.Second)
 	eventID := int64(42)
 	sourceID := "src-1"
-	installationID := 7
+	pluginID := "silo.autoscan.cephfs"
 	var gotFilter autoscan.ScanListFilter
 	var gotCountFilter autoscan.ScanListFilter
 	store := &fakeAutoscanStore{
@@ -1040,7 +1048,7 @@ func TestAutoscanHandleListScansSerializesFiltersAndEventContext(t *testing.T) {
 				},
 				AutoscanEventID:  &eventID,
 				SourceID:         &sourceID,
-				InstallationID:   &installationID,
+				PluginID:         pluginID,
 				CapabilityID:     "cephfs",
 				EventStatus:      autoscan.EventStatusSuccess,
 				EventCompletedAt: &eventCompleted,
@@ -1084,7 +1092,7 @@ func TestAutoscanHandleListScansSerializesFiltersAndEventContext(t *testing.T) {
 	if scan.AutoscanEventID == nil || *scan.AutoscanEventID != eventID || scan.EventStatus != "success" {
 		t.Fatalf("event context = %+v", scan)
 	}
-	if scan.SourceID == nil || *scan.SourceID != sourceID || scan.InstallationID == nil || *scan.InstallationID != installationID || scan.CapabilityID != "cephfs" {
+	if scan.SourceID == nil || *scan.SourceID != sourceID || scan.PluginID != pluginID || scan.CapabilityID != "cephfs" {
 		t.Fatalf("source context = %+v", scan)
 	}
 }
@@ -1128,7 +1136,7 @@ func TestAutoscanHandleListEventsSerializesFiltersAndRuns(t *testing.T) {
 				Event: autoscan.Event{
 					ID:              42,
 					SourceID:        &sourceID,
-					InstallationID:  7,
+					PluginID:        "silo.autoscan.cephfs",
 					CapabilityID:    "cephfs",
 					StartedAt:       started,
 					CompletedAt:     completed,
@@ -1258,18 +1266,30 @@ func TestAutoscanHandleUpdateSourceNormalizesLabel(t *testing.T) {
 
 func TestAutoscanHandleStatusReturnsTrimmedSources(t *testing.T) {
 	latestEventAt := time.Date(2026, 6, 4, 14, 30, 0, 0, time.UTC)
+	runningStartedAt := time.Now().Add(-90 * time.Second)
+	sourceID := "src-1"
 	store := &fakeAutoscanStore{
 		getSettingsFn: func() (autoscan.Settings, error) {
 			return autoscan.Settings{Enabled: true}, nil
 		},
 		listSourcesFn: func() ([]autoscan.Source, error) {
 			return []autoscan.Source{
-				{ID: "src-1", InstallationID: 7, CapabilityID: "scan_source", ConnectionID: ptr("conn-1"), Enabled: true},
+				{ID: "src-1", PluginID: "silo.autoscan.cephfs", CapabilityID: "scan_source", ConnectionID: ptr("conn-1"), Enabled: true},
 			}, nil
 		},
 		queueSummaryFn: func() (autoscan.QueueSummary, error) {
 			return autoscan.QueueSummary{Active: 3, Accepted: 2, Running: 1}, nil
 		},
+		listRunningFn: func() ([]autoscan.Event, error) {
+			return []autoscan.Event{{
+				ID:           44,
+				SourceID:     &sourceID,
+				PluginID:     "silo.autoscan.cephfs",
+				CapabilityID: "cephfs",
+				StartedAt:    runningStartedAt,
+				Status:       autoscan.EventStatusRunning,
+			}}, nil
+		},
 		latestEventAtFn: func() (*time.Time, error) {
 			return &latestEventAt, nil
 		},
@@ -1296,6 +1316,16 @@ func TestAutoscanHandleStatusReturnsTrimmedSources(t *testing.T) {
 	if body.ActiveScans != 3 || body.AcceptedScans != 2 || body.RunningScans != 1 {
 		t.Errorf("queue summary = %+v", body)
 	}
+	if len(body.RunningPolls) != 1 {
+		t.Fatalf("running_polls = %+v", body.RunningPolls)
+	}
+	poll := body.RunningPolls[0]
+	if poll.ID != 44 || poll.SourceID == nil || *poll.SourceID != sourceID || poll.PluginID != "silo.autoscan.cephfs" || poll.CapabilityID != "cephfs" {
+		t.Fatalf("running poll identity = %+v", poll)
+	}
+	if poll.ElapsedMS <= 0 {
+		t.Fatalf("running poll elapsed_ms = %d, want > 0", poll.ElapsedMS)
+	}
 	if body.LatestEventAt == nil || !body.LatestEventAt.Equal(latestEventAt) {
 		t.Errorf("latest_event_at = %v, want %v", body.LatestEventAt, latestEventAt)
 	}

internal/autoscan/discovery.go

@@ -9,19 +9,17 @@ import (
 // enriched with the metadata the Add-source picker needs (plugin id + a
 // human-friendly display name).
 type DiscoveredSource struct {
-	InstallationID int
-	CapabilityID   string
 	// PluginID is the installation's plugin id (e.g. "sonarr"); empty when the
 	// lister cannot supply it.
-	PluginID string
+	PluginID     string
+	CapabilityID string
 	// DisplayName is a human-friendly label for the capability (from the
 	// capability's manifest display_name, falling back to plugin/capability ids).
 	DisplayName string
 }
 
 // ScanSourceLister enumerates every installed scan_source.v1 capability so the
-// engine can (a) offer them in the Add-source picker and (b) detect orphaned
-// source rows whose plugin has been uninstalled.
+// engine can offer them in the Add-source picker.
 type ScanSourceLister interface {
 	// ListScanSources returns one entry per installed scan_source.v1 capability,
 	// enriched with plugin id + display name.
@@ -31,17 +29,9 @@ type ScanSourceLister interface {
 // AvailableScanSource is one installed scan_source capability an operator can
 // create a source against (the Add-source picker list).
 type AvailableScanSource struct {
-	InstallationID int    `json:"installation_id"`
-	CapabilityID   string `json:"capability_id"`
-	PluginID       string `json:"plugin_id"`
-	DisplayName    string `json:"display_name"`
-}
-
-// installedKey identifies an installed scan_source capability for set
-// membership tests (orphan detection in PollOnce).
-type installedKey struct {
-	InstallationID int
-	CapabilityID   string
+	PluginID     string `json:"plugin_id"`
+	CapabilityID string `json:"capability_id"`
+	DisplayName  string `json:"display_name"`
 }
 
 // ListAvailableScanSources enumerates every installed scan_source capability so
@@ -59,30 +49,10 @@ func (s *Service) ListAvailableScanSources(ctx context.Context) ([]AvailableScan
 	out := make([]AvailableScanSource, 0, len(discovered))
 	for _, d := range discovered {
 		out = append(out, AvailableScanSource{
-			InstallationID: d.InstallationID,
-			CapabilityID:   d.CapabilityID,
-			PluginID:       d.PluginID,
-			DisplayName:    d.DisplayName,
+			PluginID:     d.PluginID,
+			CapabilityID: d.CapabilityID,
+			DisplayName:  d.DisplayName,
 		})
 	}
 	return out, nil
 }
-
-// installedScanSources returns the set of currently-installed scan_source
-// capabilities so PollOnce can skip orphaned source rows (their plugin is
-// gone). A nil set means the set is unavailable (no lister) — callers must treat
-// that as "discovery unavailable" and NOT prune.
-func (s *Service) installedScanSources(ctx context.Context) (map[installedKey]struct{}, error) {
-	if s.lister == nil {
-		return nil, nil
-	}
-	discovered, err := s.lister.ListScanSources(ctx)
-	if err != nil {
-		return nil, fmt.Errorf("list scan sources: %w", err)
-	}
-	present := make(map[installedKey]struct{}, len(discovered))
-	for _, d := range discovered {
-		present[installedKey{InstallationID: d.InstallationID, CapabilityID: d.CapabilityID}] = struct{}{}
-	}
-	return present, nil
-}

internal/autoscan/discovery_test.go

@@ -14,8 +14,8 @@ func (f fakeLister) ListScanSources(context.Context) ([]DiscoveredSource, error)
 
 func TestListAvailableScanSourcesEnumeratesInstalled(t *testing.T) {
 	lister := fakeLister{sources: []DiscoveredSource{
-		{InstallationID: 1, CapabilityID: "arr-a", PluginID: "sonarr", DisplayName: "Sonarr"},
-		{InstallationID: 2, CapabilityID: "arr-b", PluginID: "radarr", DisplayName: "Radarr"},
+		{PluginID: "sonarr", CapabilityID: "arr-a", DisplayName: "Sonarr"},
+		{PluginID: "radarr", CapabilityID: "arr-b", DisplayName: "Radarr"},
 	}}
 	svc := &Service{lister: lister}
 
@@ -26,7 +26,7 @@ func TestListAvailableScanSourcesEnumeratesInstalled(t *testing.T) {
 	if len(available) != 2 {
 		t.Fatalf("expected 2 available, got %d: %+v", len(available), available)
 	}
-	if available[0] != (AvailableScanSource{InstallationID: 1, CapabilityID: "arr-a", PluginID: "sonarr", DisplayName: "Sonarr"}) {
+	if available[0] != (AvailableScanSource{PluginID: "sonarr", CapabilityID: "arr-a", DisplayName: "Sonarr"}) {
 		t.Fatalf("unexpected first available: %+v", available[0])
 	}
 }
@@ -41,32 +41,3 @@ func TestListAvailableScanSourcesNilListerEmpty(t *testing.T) {
 		t.Fatalf("nil lister must return empty, got %+v", available)
 	}
 }
-
-func TestInstalledScanSourcesSetMembership(t *testing.T) {
-	lister := fakeLister{sources: []DiscoveredSource{
-		{InstallationID: 1, CapabilityID: "arr-a"},
-	}}
-	svc := &Service{lister: lister}
-
-	present, err := svc.installedScanSources(context.Background())
-	if err != nil {
-		t.Fatalf("installedScanSources: %v", err)
-	}
-	if _, ok := present[installedKey{1, "arr-a"}]; !ok {
-		t.Fatalf("expected 1/arr-a present: %+v", present)
-	}
-	if _, ok := present[installedKey{2, "arr-b"}]; ok {
-		t.Fatalf("did not expect 2/arr-b present: %+v", present)
-	}
-}
-
-func TestInstalledScanSourcesNilListerNilSet(t *testing.T) {
-	svc := &Service{lister: nil}
-	present, err := svc.installedScanSources(context.Background())
-	if err != nil {
-		t.Fatalf("installedScanSources: %v", err)
-	}
-	if present != nil {
-		t.Fatalf("nil lister must return a nil set, got %+v", present)
-	}
-}

internal/autoscan/errors.go

@@ -5,3 +5,7 @@ import "errors"
 // ErrNotFound is returned when an autoscan operation references a connection or
 // source that does not exist. The admin API surfaces it as a 404.
 var ErrNotFound = errors.New("autoscan: not found")
+
+// ErrPollAlreadyRunning is returned when a source already has an active poll
+// event. Callers should treat it as a no-op skip, not a source failure.
+var ErrPollAlreadyRunning = errors.New("autoscan: source poll already running")

internal/autoscan/probe_test.go

@@ -69,7 +69,7 @@ func TestTestConnectionNoProbeConfigured(t *testing.T) {
 func TestSuggestRewritesMatchesArrRootsToSiloFolders(t *testing.T) {
 	store := &fakeStore{
 		sources: []Source{
-			{ID: "src-1", InstallationID: 1, CapabilityID: "arr", ConnectionID: strptr("c1")},
+			{ID: "src-1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: strptr("c1")},
 		},
 	}
 	arr := &fakeArrClient{roots: []string{"/data/tv"}}
@@ -87,7 +87,7 @@ func TestSuggestRewritesMatchesArrRootsToSiloFolders(t *testing.T) {
 
 func TestSuggestRewritesNoConnection(t *testing.T) {
 	store := &fakeStore{
-		sources: []Source{{ID: "src-1", InstallationID: 1, CapabilityID: "arr", ConnectionID: nil}},
+		sources: []Source{{ID: "src-1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: nil}},
 	}
 	svc := &Service{store: store, connres: passthroughConnRes{}, rootFolders: &fakeArrClient{}, folders: fakeFolderLister{}}
 

internal/autoscan/provider.go

@@ -9,7 +9,7 @@ import (
 // ScanSourceProvider yields changed paths for one source. The engine calls
 // PollChanges; production wraps the plugins.Service scan_source resolver.
 type ScanSourceProvider interface {
-	PollChanges(ctx context.Context, installationID int, capabilityID, marker string, conn ResolvedConnection, sourceConfig map[string]string) (changes []Change, nextMarker string, err error)
+	PollChanges(ctx context.Context, pluginID, capabilityID, marker string, conn ResolvedConnection, sourceConfig map[string]string) (changes []Change, nextMarker string, err error)
 }
 
 // PollChangesClient is the slice of *pluginhost.ScanSourceClient used here. It
@@ -20,10 +20,10 @@ type PollChangesClient interface {
 	PollChanges(ctx context.Context, req *pluginv1.PollChangesRequest) (*pluginv1.PollChangesResponse, error)
 }
 
-// ScanSourceResolver yields a per-(installation, capability) scan-source client.
+// ScanSourceResolver yields a per-(plugin, capability) scan-source client.
 // Exported for the same cross-package adapter reason as PollChangesClient.
 type ScanSourceResolver interface {
-	ScanSourceClient(ctx context.Context, installationID int, capabilityID string) (PollChangesClient, error)
+	ScanSourceClient(ctx context.Context, pluginID, capabilityID string) (PollChangesClient, error)
 }
 
 type pluginProvider struct{ resolver ScanSourceResolver }
@@ -34,8 +34,8 @@ func NewPluginProvider(resolver ScanSourceResolver) ScanSourceProvider {
 	return &pluginProvider{resolver: resolver}
 }
 
-func (p *pluginProvider) PollChanges(ctx context.Context, installationID int, capabilityID, marker string, conn ResolvedConnection, sourceConfig map[string]string) ([]Change, string, error) {
-	client, err := p.resolver.ScanSourceClient(ctx, installationID, capabilityID)
+func (p *pluginProvider) PollChanges(ctx context.Context, pluginID, capabilityID, marker string, conn ResolvedConnection, sourceConfig map[string]string) ([]Change, string, error) {
+	client, err := p.resolver.ScanSourceClient(ctx, pluginID, capabilityID)
 	if err != nil {
 		return nil, "", err
 	}

internal/autoscan/provider_test.go

@@ -17,10 +17,17 @@ func (c *capturingClient) PollChanges(_ context.Context, req *pluginv1.PollChang
 	return &pluginv1.PollChangesResponse{SourcePaths: []string{"/mnt/media/x"}, NextMarker: "m2"}, nil
 }
 
-// capturingResolver yields a fixed PollChangesClient.
-type capturingResolver struct{ client PollChangesClient }
+// capturingResolver yields a fixed PollChangesClient and records the requested
+// stable scan-source identity.
+type capturingResolver struct {
+	client       PollChangesClient
+	lastPluginID string
+	lastCapID    string
+}
 
-func (r capturingResolver) ScanSourceClient(context.Context, int, string) (PollChangesClient, error) {
+func (r *capturingResolver) ScanSourceClient(_ context.Context, pluginID, capabilityID string) (PollChangesClient, error) {
+	r.lastPluginID = pluginID
+	r.lastCapID = capabilityID
 	return r.client, nil
 }
 
@@ -28,11 +35,12 @@ func (r capturingResolver) ScanSourceClient(context.Context, int, string) (PollC
 // delivered to the plugin on the PollChangesRequest.
 func TestPluginProviderPopulatesConnection(t *testing.T) {
 	client := &capturingClient{}
-	prov := NewPluginProvider(capturingResolver{client})
+	resolver := &capturingResolver{client: client}
+	prov := NewPluginProvider(resolver)
 
 	conn := ResolvedConnection{BaseURL: "https://arr.example", APIKey: "secret-key"}
 	sourceConfig := map[string]string{"exclusions": ".downloads"}
-	changes, next, err := prov.PollChanges(context.Background(), 1, "cap", "m1", conn, sourceConfig)
+	changes, next, err := prov.PollChanges(context.Background(), "silo.autoscan.arr", "cap", "m1", conn, sourceConfig)
 	if err != nil {
 		t.Fatalf("PollChanges: %v", err)
 	}
@@ -45,6 +53,9 @@ func TestPluginProviderPopulatesConnection(t *testing.T) {
 	if client.last.GetCapabilityId() != "cap" || client.last.GetMarker() != "m1" {
 		t.Fatalf("unexpected request fields: cap=%q marker=%q", client.last.GetCapabilityId(), client.last.GetMarker())
 	}
+	if resolver.lastPluginID != "silo.autoscan.arr" || resolver.lastCapID != "cap" {
+		t.Fatalf("resolver identity = %q/%q", resolver.lastPluginID, resolver.lastCapID)
+	}
 	rc := client.last.GetConnection()
 	if rc == nil {
 		t.Fatal("expected connection to be populated on the request")
@@ -59,9 +70,9 @@ func TestPluginProviderPopulatesConnection(t *testing.T) {
 
 func TestPluginProviderPrefersStructuredChanges(t *testing.T) {
 	client := &capturingStructuredClient{}
-	prov := NewPluginProvider(capturingResolver{client: client})
+	prov := NewPluginProvider(&capturingResolver{client: client})
 
-	changes, next, err := prov.PollChanges(context.Background(), 1, "cap", "m1", ResolvedConnection{}, nil)
+	changes, next, err := prov.PollChanges(context.Background(), "silo.autoscan.arr", "cap", "m1", ResolvedConnection{}, nil)
 	if err != nil {
 		t.Fatalf("PollChanges: %v", err)
 	}

internal/autoscan/repository.go

@@ -175,14 +175,14 @@ func (r *Repository) GetConnection(ctx context.Context, id string) (Connection,
 
 // --- Sources ---
 
-const sourceColumns = `id, installation_id, capability_id, connection_id, enabled,
+const sourceColumns = `id, plugin_id, capability_id, connection_id, enabled,
 	poll_interval_seconds, path_rewrites, source_config, label, marker, last_run_at, last_error`
 
 func scanSource(row interface{ Scan(...any) error }) (Source, error) {
 	var s Source
 	var pathRewrites []byte
 	var sourceConfig []byte
-	if err := row.Scan(&s.ID, &s.InstallationID, &s.CapabilityID, &s.ConnectionID,
+	if err := row.Scan(&s.ID, &s.PluginID, &s.CapabilityID, &s.ConnectionID,
 		&s.Enabled, &s.PollIntervalSeconds, &pathRewrites, &sourceConfig, &s.Label, &s.Marker, &s.LastRunAt, &s.LastError); err != nil {
 		return Source{}, err
 	}
@@ -283,11 +283,11 @@ func (r *Repository) CreateSource(ctx context.Context, s Source) (Source, error)
 	}
 	row := r.pool.QueryRow(ctx, `
 		INSERT INTO autoscan_sources (
-			installation_id, capability_id, connection_id, enabled, poll_interval_seconds, path_rewrites, source_config, label
+			plugin_id, capability_id, connection_id, enabled, poll_interval_seconds, path_rewrites, source_config, label
 		)
 		VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
 		RETURNING `+sourceColumns,
-		s.InstallationID, s.CapabilityID, connectionIDArg(s.ConnectionID), s.Enabled, s.PollIntervalSeconds, rewrites, sourceConfig, s.Label)
+		s.PluginID, s.CapabilityID, connectionIDArg(s.ConnectionID), s.Enabled, s.PollIntervalSeconds, rewrites, sourceConfig, s.Label)
 	out, err := scanSource(row)
 	if err != nil {
 		if connID, ok := connectionFKViolation(err, s.ConnectionID); ok {
@@ -299,7 +299,7 @@ func (r *Repository) CreateSource(ctx context.Context, s Source) (Source, error)
 }
 
 // UpdateSource updates a source's binding/scheduling fields by id. Identity
-// (installation_id, capability_id) and bookkeeping fields (marker/last_run_at/
+// (plugin_id, capability_id) and bookkeeping fields (marker/last_run_at/
 // last_error) are left untouched. An unknown id maps to ErrNotFound; a
 // non-existent connection trips the FK constraint and also maps to ErrNotFound.
 func (r *Repository) UpdateSource(ctx context.Context, s Source) (Source, error) {
@@ -353,7 +353,7 @@ func connectionFKViolation(err error, connectionID *string) (string, bool) {
 
 func (r *Repository) ListSources(ctx context.Context) ([]Source, error) {
 	rows, err := r.pool.Query(ctx, `SELECT `+sourceColumns+`
-		FROM autoscan_sources ORDER BY installation_id, capability_id`)
+		FROM autoscan_sources ORDER BY plugin_id, capability_id`)
 	if err != nil {
 		return nil, fmt.Errorf("list autoscan sources: %w", err)
 	}
@@ -364,7 +364,7 @@ func (r *Repository) ListSources(ctx context.Context) ([]Source, error) {
 func (r *Repository) ListEnabledSources(ctx context.Context) ([]Source, error) {
 	rows, err := r.pool.Query(ctx, `SELECT `+sourceColumns+`
 		FROM autoscan_sources WHERE enabled = true
-		ORDER BY installation_id, capability_id`)
+		ORDER BY plugin_id, capability_id`)
 	if err != nil {
 		return nil, fmt.Errorf("list enabled autoscan sources: %w", err)
 	}
@@ -462,7 +462,7 @@ func (r *Repository) RecordError(ctx context.Context, sourceID, msg string) erro
 	return nil
 }
 
-const eventColumns = `id, source_id, installation_id, capability_id, started_at, completed_at,
+const eventColumns = `id, source_id, plugin_id, capability_id, started_at, completed_at,
 	duration_ms, status, changes_returned, changes_resolved, targets_claimed, scans_created,
 	scans_reused, scans_suppressed, error_message, marker_before, marker_after`
 
@@ -472,7 +472,7 @@ func scanEvent(row interface{ Scan(...any) error }) (Event, error) {
 	if err := row.Scan(
 		&e.ID,
 		&e.SourceID,
-		&e.InstallationID,
+		&e.PluginID,
 		&e.CapabilityID,
 		&e.StartedAt,
 		&e.CompletedAt,
@@ -499,25 +499,80 @@ func (r *Repository) CreateEvent(ctx context.Context, in EventCreate) (int64, er
 	if started.IsZero() {
 		started = time.Now()
 	}
-	row := r.pool.QueryRow(ctx, `
+	sourceID := strings.TrimSpace(in.SourceID)
+	if sourceID == "" {
+		row := r.pool.QueryRow(ctx, `
+			INSERT INTO autoscan_events (
+				source_id, plugin_id, capability_id, started_at, completed_at,
+				duration_ms, status, error_message, marker_before
+			)
+		VALUES ($1, $2, $3, $4, $4, 0, $5, $6, $7)
+		RETURNING id`,
+			nil,
+			in.PluginID,
+			in.CapabilityID,
+			started,
+			string(EventStatusRunning),
+			"",
+			nullable(in.MarkerBefore),
+		)
+		var id int64
+		if err := row.Scan(&id); err != nil {
+			return 0, fmt.Errorf("create autoscan event: %w", err)
+		}
+		return id, nil
+	}
+
+	tx, err := r.pool.Begin(ctx)
+	if err != nil {
+		return 0, fmt.Errorf("begin autoscan event: %w", err)
+	}
+	defer func() { _ = tx.Rollback(ctx) }()
+
+	if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(900173, hashtext($1))`, sourceID); err != nil {
+		return 0, fmt.Errorf("lock autoscan event: %w", err)
+	}
+
+	var runningID int64
+	err = tx.QueryRow(ctx, `
+		SELECT id
+		FROM autoscan_events
+		WHERE source_id = $1
+		  AND status = $2
+		ORDER BY started_at ASC, id ASC
+		LIMIT 1`,
+		sourceID,
+		string(EventStatusRunning),
+	).Scan(&runningID)
+	if err == nil {
+		return 0, fmt.Errorf("%w: source %s event %d", ErrPollAlreadyRunning, sourceID, runningID)
+	}
+	if !errors.Is(err, pgx.ErrNoRows) {
+		return 0, fmt.Errorf("check running autoscan event: %w", err)
+	}
+
+	row := tx.QueryRow(ctx, `
 		INSERT INTO autoscan_events (
-			source_id, installation_id, capability_id, started_at, completed_at,
+			source_id, plugin_id, capability_id, started_at, completed_at,
 			duration_ms, status, error_message, marker_before
 		)
 		VALUES ($1, $2, $3, $4, $4, 0, $5, $6, $7)
 		RETURNING id`,
-		nullable(in.SourceID),
-		in.InstallationID,
+		sourceID,
+		in.PluginID,
 		in.CapabilityID,
 		started,
-		string(EventStatusError),
-		"poll started but did not finish",
+		string(EventStatusRunning),
+		"",
 		nullable(in.MarkerBefore),
 	)
 	var id int64
 	if err := row.Scan(&id); err != nil {
 		return 0, fmt.Errorf("create autoscan event: %w", err)
 	}
+	if err := tx.Commit(ctx); err != nil {
+		return 0, fmt.Errorf("commit autoscan event: %w", err)
+	}
 	return id, nil
 }
 
@@ -565,6 +620,25 @@ func (r *Repository) FinishEvent(ctx context.Context, in EventFinish) error {
 	return nil
 }
 
+func (r *Repository) MarkInterruptedEvents(ctx context.Context) error {
+	msg := "poll started but did not finish"
+	_, err := r.pool.Exec(ctx, `
+		UPDATE autoscan_events
+		SET completed_at = now(),
+			duration_ms = GREATEST(0, EXTRACT(EPOCH FROM (now() - started_at)) * 1000)::bigint,
+			status = $1,
+			error_message = $2
+		WHERE status = $3`,
+		string(EventStatusError),
+		msg,
+		string(EventStatusRunning),
+	)
+	if err != nil {
+		return fmt.Errorf("mark interrupted autoscan events: %w", err)
+	}
+	return nil
+}
+
 // clampAutoscanLimit bounds a requested page size to a sane window: the
 // default page when unset, capped so a single query can never fan out.
 func clampAutoscanLimit(limit int) int {
@@ -758,6 +832,30 @@ func (r *Repository) ListEvents(ctx context.Context, filter EventListFilter) ([]
 	return events, runRows.Err()
 }
 
+func (r *Repository) ListRunningEvents(ctx context.Context) ([]Event, error) {
+	rows, err := r.pool.Query(ctx, `
+		SELECT `+eventColumns+`
+		FROM autoscan_events
+		WHERE status = $1
+		ORDER BY started_at ASC, id ASC`,
+		string(EventStatusRunning),
+	)
+	if err != nil {
+		return nil, fmt.Errorf("list running autoscan events: %w", err)
+	}
+	defer rows.Close()
+
+	events := make([]Event, 0)
+	for rows.Next() {
+		event, scanErr := scanEvent(rows)
+		if scanErr != nil {
+			return nil, scanErr
+		}
+		events = append(events, event)
+	}
+	return events, rows.Err()
+}
+
 func (r *Repository) ListAutoscanScans(ctx context.Context, filter ScanListFilter) ([]ScanWithEvent, error) {
 	limit := clampAutoscanLimit(filter.Limit)
 	offset := filter.Offset
@@ -780,15 +878,15 @@ func (r *Repository) ListAutoscanScans(ctx context.Context, filter ScanListFilte
 			sr.trigger,
 			sr.status,
 			COALESCE(sr.error_message, ''),
-			sr.requested_at,
-			sr.started_at,
-			sr.completed_at,
-			sr.autoscan_event_id,
-			e.source_id,
-			e.installation_id,
-			COALESCE(e.capability_id, ''),
-			COALESCE(e.status, ''),
-			e.completed_at
+				sr.requested_at,
+				sr.started_at,
+				sr.completed_at,
+				sr.autoscan_event_id,
+				e.source_id,
+				COALESCE(e.plugin_id, ''),
+				COALESCE(e.capability_id, ''),
+				COALESCE(e.status, ''),
+				e.completed_at
 		FROM scan_runs sr
 		LEFT JOIN autoscan_events e ON e.id = sr.autoscan_event_id
 		WHERE `+strings.Join(clauses, " AND ")+`
@@ -818,7 +916,7 @@ func (r *Repository) ListAutoscanScans(ctx context.Context, filter ScanListFilte
 			&scan.CompletedAt,
 			&scan.AutoscanEventID,
 			&scan.SourceID,
-			&scan.InstallationID,
+			&scan.PluginID,
 			&scan.CapabilityID,
 			&eventStatus,
 			&scan.EventCompletedAt,
@@ -853,7 +951,7 @@ func (r *Repository) GetQueueSummary(ctx context.Context) (QueueSummary, error)
 
 func (r *Repository) LatestEventAt(ctx context.Context) (*time.Time, error) {
 	var latest *time.Time
-	err := r.pool.QueryRow(ctx, `SELECT max(completed_at) FROM autoscan_events`).Scan(&latest)
+	err := r.pool.QueryRow(ctx, `SELECT max(completed_at) FROM autoscan_events WHERE status <> $1`, string(EventStatusRunning)).Scan(&latest)
 	if err != nil {
 		return nil, fmt.Errorf("get latest autoscan event time: %w", err)
 	}

internal/autoscan/service.go

@@ -91,8 +91,7 @@ func (s *Service) SetSuggesterDeps(rootFolders RootFolderClient, folders FolderL
 // from scan_source plugins; connres resolves a source's connection to concrete
 // credentials; resolver/queue/suppress drive the resolve→suppress→enqueue loop.
 // lister enumerates installed scan_source capabilities so the Add-source picker
-// can offer them and PollOnce can skip orphaned source rows; it may be nil
-// (the picker then returns an empty list and orphan-pruning is disabled).
+// can offer them; it may be nil (the picker then returns an empty list).
 func NewService(
 	store Store,
 	provider ScanSourceProvider,
@@ -113,24 +112,15 @@ func NewService(
 	}
 }
 
-// PollOnce runs one autoscan cycle. Per-source failures are logged and skipped;
-// only settings/listing errors propagate. The opaque next marker returned by the
+// PollOnce runs one autoscan cycle. Per-source failures are logged, recorded on
+// the source/event, and the loop continues; only settings/listing errors
+// propagate. The opaque next marker returned by the
 // provider is stored verbatim, but only when the cycle's work is genuinely
 // consumed: the provider returned no paths, or it returned paths and at least one
 // resolved+enqueued. When paths come back but NONE resolve to a library folder
 // (e.g. a freshly-enabled source with unconfigured rewrites) the marker is held
 // and an error recorded, so those imports aren't skipped forever.
 func (s *Service) PollOnce(ctx context.Context) error {
-	// Fetch the set of currently-installed scan_source capabilities so we can skip
-	// orphaned source rows (their plugin was uninstalled). Listing failures are
-	// non-fatal: a nil set means the installed set is unavailable, in which case
-	// we do NOT prune orphans — every enabled source is assumed present.
-	present, derr := s.installedScanSources(ctx)
-	if derr != nil {
-		slog.WarnContext(ctx, "autoscan: list installed scan sources failed", "err", derr)
-		present = nil
-	}
-
 	settings, err := s.store.GetSettings(ctx)
 	if err != nil {
 		return err
@@ -146,16 +136,6 @@ func (s *Service) PollOnce(ctx context.Context) error {
 	now := time.Now()
 
 	for _, src := range sources {
-		// Skip orphaned sources: an enabled source whose scan_source plugin has
-		// been uninstalled/disabled is no longer in the discovered set, so polling
-		// it would error every cycle. Skip it quietly (no RecordError) so the spam
-		// stops; the operator can delete it via the source delete endpoint. A nil
-		// `present` set means discovery is unavailable — don't prune in that case.
-		if present != nil {
-			if _, ok := present[installedKey{InstallationID: src.InstallationID, CapabilityID: src.CapabilityID}]; !ok {
-				continue
-			}
-		}
 		// Honor the per-source poll interval as a "poll at most every N seconds"
 		// floor: the global task fires at the default cadence, so a source with a
 		// longer interval is skipped until enough time has elapsed.
@@ -170,7 +150,10 @@ func (s *Service) PollOnce(ctx context.Context) error {
 		if src.Marker != nil {
 			marker = *src.Marker
 		}
-		eventID := s.createEvent(ctx, src, marker, time.Now())
+		eventID, started := s.createEvent(ctx, src, marker, time.Now())
+		if !started {
+			continue
+		}
 		// A connection is OPTIONAL. Server-based providers (Sonarr/Radarr) bind a
 		// connection and the resolved {base_url, api_key} is handed to the plugin.
 		// Other providers (e.g. a filesystem/CephFS watcher) need none and get an
@@ -195,7 +178,7 @@ func (s *Service) PollOnce(ctx context.Context) error {
 			}
 			conn = resolved
 		}
-		changes, next, perr := s.provider.PollChanges(ctx, src.InstallationID, src.CapabilityID, marker, conn, src.SourceConfig)
+		changes, next, perr := s.provider.PollChanges(ctx, src.PluginID, src.CapabilityID, marker, conn, src.SourceConfig)
 		if perr != nil {
 			slog.WarnContext(ctx, "autoscan: poll changes failed", "source_id", src.ID, "err", perr)
 			if rerr := s.store.RecordError(ctx, src.ID, perr.Error()); rerr != nil {
@@ -295,22 +278,26 @@ func (s *Service) PollOnce(ctx context.Context) error {
 	return nil
 }
 
-func (s *Service) createEvent(ctx context.Context, src Source, marker string, startedAt time.Time) int64 {
+func (s *Service) createEvent(ctx context.Context, src Source, marker string, startedAt time.Time) (int64, bool) {
 	if s == nil || s.store == nil {
-		return 0
+		return 0, true
 	}
 	id, err := s.store.CreateEvent(ctx, EventCreate{
-		SourceID:       src.ID,
-		InstallationID: src.InstallationID,
-		CapabilityID:   src.CapabilityID,
-		StartedAt:      startedAt,
-		MarkerBefore:   marker,
+		SourceID:     src.ID,
+		PluginID:     src.PluginID,
+		CapabilityID: src.CapabilityID,
+		StartedAt:    startedAt,
+		MarkerBefore: marker,
 	})
 	if err != nil {
+		if errors.Is(err, ErrPollAlreadyRunning) {
+			slog.DebugContext(ctx, "autoscan: source poll already running", "source_id", src.ID)
+			return 0, false
+		}
 		slog.WarnContext(ctx, "autoscan: create event failed", "source_id", src.ID, "err", err)
-		return 0
+		return 0, true
 	}
-	return id
+	return id, true
 }
 
 func (s *Service) finishEvent(ctx context.Context, eventID int64, finish EventFinish) {

internal/autoscan/service_test.go

@@ -2,6 +2,7 @@ package autoscan
 
 import (
 	"context"
+	"errors"
 	"path/filepath"
 	"strings"
 	"testing"
@@ -12,13 +13,14 @@ import (
 )
 
 type fakeStore struct {
-	settings      Settings
-	sources       []Source
-	connection    Connection
-	advanced      map[string]string // source ID -> marker
-	recorded      map[string]string // source ID -> error message
-	createdEvents []EventCreate
-	events        []EventFinish
+	settings       Settings
+	sources        []Source
+	connection     Connection
+	advanced       map[string]string // source ID -> marker
+	recorded       map[string]string // source ID -> error message
+	createdEvents  []EventCreate
+	events         []EventFinish
+	createEventErr error
 }
 
 func (f *fakeStore) GetSettings(context.Context) (Settings, error) { return f.settings, nil }
@@ -51,6 +53,9 @@ func (f *fakeStore) RecordError(_ context.Context, sourceID, msg string) error {
 	return nil
 }
 func (f *fakeStore) CreateEvent(_ context.Context, event EventCreate) (int64, error) {
+	if f.createEventErr != nil {
+		return 0, f.createEventErr
+	}
 	f.createdEvents = append(f.createdEvents, event)
 	return int64(len(f.createdEvents)), nil
 }
@@ -65,14 +70,20 @@ type fakeProvider struct {
 	changes    map[string][]Change // key: capabilityID, structured changes
 	nextMarker string
 	err        error
+	errByCap   map[string]error
 	lastConfig map[string]string
+	calls      int
 }
 
-func (f *fakeProvider) PollChanges(_ context.Context, _ int, capabilityID, _ string, _ ResolvedConnection, sourceConfig map[string]string) ([]Change, string, error) {
+func (f *fakeProvider) PollChanges(_ context.Context, _ string, capabilityID, _ string, _ ResolvedConnection, sourceConfig map[string]string) ([]Change, string, error) {
+	f.calls++
 	f.lastConfig = sourceConfig
 	if f.err != nil {
 		return nil, "", f.err
 	}
+	if err := f.errByCap[capabilityID]; err != nil {
+		return nil, "", err
+	}
 	if changes, ok := f.changes[capabilityID]; ok {
 		return changes, f.nextMarker, nil
 	}
@@ -87,7 +98,7 @@ func TestPollOncePassesSourceConfigToProvider(t *testing.T) {
 	store := &fakeStore{
 		settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60},
 		sources: []Source{{
-			ID: "s1", InstallationID: 1, CapabilityID: "cephfs", Enabled: true,
+			ID: "s1", PluginID: "silo.autoscan.cephfs", CapabilityID: "cephfs", Enabled: true,
 			SourceConfig: map[string]string{"exclusions": ".downloads\n.recyclebin"},
 		}},
 	}
@@ -101,6 +112,38 @@ func TestPollOncePassesSourceConfigToProvider(t *testing.T) {
 	}
 }
 
+func TestPollOnceSkipsSourceWhenPollAlreadyRunning(t *testing.T) {
+	store := &fakeStore{
+		settings:       Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60},
+		createEventErr: ErrPollAlreadyRunning,
+		sources: []Source{{
+			ID: "s1", PluginID: "silo.autoscan.cephfs", CapabilityID: "cephfs", Enabled: true,
+		}},
+	}
+	prov := &fakeProvider{paths: map[string][]string{"cephfs": {"/mnt/media/Movie/movie.mkv"}}, nextMarker: "m1"}
+	q := &recordingQueuer{}
+	svc := newService(store, prov, q, allowSuppressor{})
+
+	if err := svc.PollOnce(context.Background()); err != nil {
+		t.Fatalf("PollOnce: %v", err)
+	}
+	if prov.calls != 0 {
+		t.Fatalf("provider calls = %d, want 0", prov.calls)
+	}
+	if len(q.enqueued) != 0 {
+		t.Fatalf("enqueued scans = %+v, want none", q.enqueued)
+	}
+	if len(store.events) != 0 {
+		t.Fatalf("finished events = %+v, want none", store.events)
+	}
+	if len(store.recorded) != 0 {
+		t.Fatalf("recorded source errors = %+v, want none", store.recorded)
+	}
+	if len(store.advanced) != 0 {
+		t.Fatalf("advanced markers = %+v, want none", store.advanced)
+	}
+}
+
 // passthroughConnRes resolves to empty credentials; the engine doesn't inspect
 // them in tests (the fake provider ignores conn).
 type passthroughConnRes struct{}
@@ -210,7 +253,7 @@ func TestPollOnceEnqueuesDedupedFolders(t *testing.T) {
 	store := &fakeStore{
 		settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60},
 		sources: []Source{{
-			ID: "s1", InstallationID: 1, CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true,
+			ID: "s1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true,
 		}},
 	}
 	prov := &fakeProvider{paths: map[string][]string{
@@ -241,7 +284,7 @@ func TestPollOnceRecordsSuccessfulEvent(t *testing.T) {
 	store := &fakeStore{
 		settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60},
 		sources: []Source{{
-			ID: "s1", InstallationID: 1, CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true, Marker: &marker,
+			ID: "s1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true, Marker: &marker,
 		}},
 	}
 	prov := &fakeProvider{paths: map[string][]string{
@@ -283,7 +326,7 @@ func TestPollOnceRecordsReusedScanCounts(t *testing.T) {
 	store := &fakeStore{
 		settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60},
 		sources: []Source{{
-			ID: "s1", InstallationID: 1, CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true,
+			ID: "s1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true,
 		}},
 	}
 	prov := &fakeProvider{paths: map[string][]string{
@@ -318,7 +361,7 @@ func TestPollOnceAppliesSourceRewritesBeforeEnqueue(t *testing.T) {
 	store := &fakeStore{
 		settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60},
 		sources: []Source{{
-			ID: "s1", InstallationID: 1, CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true,
+			ID: "s1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true,
 			PathRewrites: []PathRewrite{{From: "/data/tv", To: "/mnt/media/tv"}},
 		}},
 	}
@@ -347,7 +390,7 @@ func TestPollOnceStructuredFileChangeEnqueuesExactFile(t *testing.T) {
 	store := &fakeStore{
 		settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60},
 		sources: []Source{{
-			ID: "s1", InstallationID: 1, CapabilityID: "cephfs", Enabled: true,
+			ID: "s1", PluginID: "silo.autoscan.cephfs", CapabilityID: "cephfs", Enabled: true,
 			PathRewrites: []PathRewrite{{From: "/ceph/tv", To: "/mnt/media/tv"}},
 		}},
 	}
@@ -380,7 +423,7 @@ func TestPollOnceStructuredSubtreeChangeEnqueuesExactSubtree(t *testing.T) {
 	store := &fakeStore{
 		settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60},
 		sources: []Source{{
-			ID: "s1", InstallationID: 1, CapabilityID: "cephfs", Enabled: true,
+			ID: "s1", PluginID: "silo.autoscan.cephfs", CapabilityID: "cephfs", Enabled: true,
 			PathRewrites: []PathRewrite{{From: "/ceph/movies", To: "/mnt/media/movies"}},
 		}},
 	}
@@ -416,7 +459,7 @@ func TestPollOnceScansDistinctPathsUnderSameFolder(t *testing.T) {
 	store := &fakeStore{
 		settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60},
 		sources: []Source{{
-			ID: "s1", InstallationID: 1, CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true,
+			ID: "s1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true,
 		}},
 	}
 	prov := &fakeProvider{paths: map[string][]string{
@@ -446,7 +489,7 @@ func TestPollOnceStoresOpaqueMarkerVerbatim(t *testing.T) {
 	store := &fakeStore{
 		settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60},
 		sources: []Source{{
-			ID: "s1", InstallationID: 1, CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true,
+			ID: "s1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true,
 		}},
 	}
 	const opaque = "eyJjdXJzb3IiOiJhYmMxMjMifQ==|2026-06-02T14:10:00Z"
@@ -470,7 +513,7 @@ func TestPollOnceHoldsMarkerWhenPathsReturnedButNoneResolve(t *testing.T) {
 	store := &fakeStore{
 		settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60},
 		sources: []Source{{
-			ID: "s1", InstallationID: 1, CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true,
+			ID: "s1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true,
 		}},
 	}
 	prov := &fakeProvider{paths: map[string][]string{
@@ -523,7 +566,7 @@ func TestPollOnceAdvancesMarkerWhenResolvedButAllSuppressed(t *testing.T) {
 	store := &fakeStore{
 		settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60},
 		sources: []Source{{
-			ID: "s1", InstallationID: 1, CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true,
+			ID: "s1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true,
 			// rewrite /data/tv -> /mnt/media/tv so fakeResolver resolves the paths.
 			PathRewrites: []PathRewrite{{From: "/data/tv", To: "/mnt/media/tv"}},
 		}},
@@ -556,7 +599,7 @@ func TestPollOnceAdvancesMarkerWhenZeroPathsReturned(t *testing.T) {
 	store := &fakeStore{
 		settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60},
 		sources: []Source{{
-			ID: "s1", InstallationID: 1, CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true,
+			ID: "s1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true,
 		}},
 	}
 	prov := &fakeProvider{paths: map[string][]string{"arr": {}}, nextMarker: "m1"}
@@ -588,7 +631,7 @@ func TestPollOnceDisabledNoop(t *testing.T) {
 func TestPollOnceProviderErrorKeepsMarker(t *testing.T) {
 	store := &fakeStore{
 		settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60},
-		sources:  []Source{{ID: "s1", InstallationID: 1, CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true}},
+		sources:  []Source{{ID: "s1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true}},
 	}
 	prov := &fakeProvider{err: context.DeadlineExceeded}
 	q := &recordingQueuer{}
@@ -607,7 +650,7 @@ func TestPollOnceProviderErrorKeepsMarker(t *testing.T) {
 func TestPollOnceReleasesClaimOnEnqueueFailure(t *testing.T) {
 	store := &fakeStore{
 		settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60},
-		sources:  []Source{{ID: "s1", InstallationID: 1, CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true}},
+		sources:  []Source{{ID: "s1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true}},
 	}
 	prov := &fakeProvider{paths: map[string][]string{"arr": {"/mnt/media/Show/S01/E01.mkv"}}, nextMarker: "m1"}
 	sup := &recordingSuppressor{}
@@ -640,7 +683,7 @@ func TestPollOncePollsConnectionlessSource(t *testing.T) {
 	store := &fakeStore{
 		settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60},
 		sources: []Source{{
-			ID: "s1", InstallationID: 1, CapabilityID: "arr", ConnectionID: nil, Enabled: true,
+			ID: "s1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: nil, Enabled: true,
 		}},
 	}
 	prov := &fakeProvider{paths: map[string][]string{"arr": {"/mnt/media/Show/S01/E01.mkv"}}, nextMarker: "m1"}
@@ -666,7 +709,7 @@ func TestPollOnceSkipsSourcePolledTooRecently(t *testing.T) {
 	store := &fakeStore{
 		settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60},
 		sources: []Source{{
-			ID: "s1", InstallationID: 1, CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true,
+			ID: "s1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true,
 			PollIntervalSeconds: &interval, LastRunAt: &recent,
 		}},
 	}
@@ -690,7 +733,7 @@ func TestPollOnceRunsSourcePastItsInterval(t *testing.T) {
 	store := &fakeStore{
 		settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60},
 		sources: []Source{{
-			ID: "s1", InstallationID: 1, CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true,
+			ID: "s1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true,
 			PollIntervalSeconds: &interval, LastRunAt: &old,
 		}},
 	}
@@ -708,25 +751,23 @@ func TestPollOnceRunsSourcePastItsInterval(t *testing.T) {
 	}
 }
 
-func TestPollOnceSkipsOrphanedSourceQuietly(t *testing.T) {
-	// Two enabled sources: only "arr-live" is still discovered; "arr-gone"'s
-	// plugin has been uninstalled. The orphan must be skipped quietly — no poll,
-	// no RecordError spam — while the live source still polls normally.
+func TestPollOnceRecordsScanSourceResolutionFailure(t *testing.T) {
+	// Two enabled sources: one polls normally, while the other simulates the
+	// production resolver failing because the plugin cannot be resolved.
 	store := &fakeStore{
 		settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60},
 		sources: []Source{
-			{ID: "live", InstallationID: 1, CapabilityID: "arr-live", ConnectionID: strptr("c1"), Enabled: true},
-			{ID: "gone", InstallationID: 2, CapabilityID: "arr-gone", ConnectionID: strptr("c1"), Enabled: true},
+			{ID: "live", PluginID: "silo.autoscan.live", CapabilityID: "arr-live", ConnectionID: strptr("c1"), Enabled: true},
+			{ID: "gone", PluginID: "silo.autoscan.gone", CapabilityID: "arr-gone", ConnectionID: strptr("c1"), Enabled: true},
 		},
 	}
 	prov := &fakeProvider{paths: map[string][]string{
 		"arr-live": {"/mnt/media/Show/S01/E01.mkv"},
-		"arr-gone": {"/mnt/media/Other/S01/E01.mkv"},
+	}, errByCap: map[string]error{
+		"arr-gone": errors.New("scan source plugin \"silo.autoscan.gone\" is not installed"),
 	}, nextMarker: "m1"}
 	q := &recordingQueuer{}
-	// Lister reports only the live capability; the orphan is absent.
-	lister := fakeLister{sources: []DiscoveredSource{{InstallationID: 1, CapabilityID: "arr-live"}}}
-	svc := NewService(store, prov, passthroughConnRes{}, fakeResolver{}, q, allowSuppressor{}, lister)
+	svc := newService(store, prov, q, allowSuppressor{})
 
 	if err := svc.PollOnce(context.Background()); err != nil {
 		t.Fatalf("PollOnce: %v", err)
@@ -738,9 +779,15 @@ func TestPollOnceSkipsOrphanedSourceQuietly(t *testing.T) {
 		t.Fatalf("expected live source marker advanced")
 	}
 	if _, ok := store.advanced["gone"]; ok {
-		t.Fatalf("orphaned source must NOT poll/advance")
+		t.Fatalf("failed source must NOT advance")
 	}
-	if _, ok := store.recorded["gone"]; ok {
-		t.Fatalf("orphaned source must be skipped quietly, but an error was recorded: %q", store.recorded["gone"])
+	if got := store.recorded["gone"]; !strings.Contains(got, "not installed") {
+		t.Fatalf("failed source error = %q", got)
+	}
+	if len(store.events) != 2 {
+		t.Fatalf("expected one event per source, got %+v", store.events)
+	}
+	if store.events[1].Status != EventStatusError || !strings.Contains(store.events[1].ErrorMessage, "not installed") {
+		t.Fatalf("failed source event = %+v", store.events[1])
 	}
 }

internal/autoscan/types.go

@@ -45,11 +45,11 @@ type Change struct {
 	Scope      ChangeScope
 }
 
-// Source ties a scan_source plugin capability instance to a connection plus the
+// Source ties a scan_source plugin capability to a connection plus the
 // host-owned scheduling/bookkeeping state.
 type Source struct {
 	ID                  string
-	InstallationID      int
+	PluginID            string
 	CapabilityID        string
 	ConnectionID        *string // nil until an operator binds a connection
 	Enabled             bool
@@ -65,6 +65,7 @@ type Source struct {
 type EventStatus string
 
 const (
+	EventStatusRunning    EventStatus = "running"
 	EventStatusSuccess    EventStatus = "success"
 	EventStatusError      EventStatus = "error"
 	EventStatusUnresolved EventStatus = "unresolved"
@@ -73,7 +74,7 @@ const (
 type Event struct {
 	ID              int64
 	SourceID        *string
-	InstallationID  int
+	PluginID        string
 	CapabilityID    string
 	StartedAt       time.Time
 	CompletedAt     time.Time
@@ -91,11 +92,11 @@ type Event struct {
 }
 
 type EventCreate struct {
-	SourceID       string
-	InstallationID int
-	CapabilityID   string
-	StartedAt      time.Time
-	MarkerBefore   string
+	SourceID     string
+	PluginID     string
+	CapabilityID string
+	StartedAt    time.Time
+	MarkerBefore string
 }
 
 type EventFinish struct {
@@ -154,7 +155,7 @@ type ScanWithEvent struct {
 	ScanRunSummary
 	AutoscanEventID  *int64
 	SourceID         *string
-	InstallationID   *int
+	PluginID         string
 	CapabilityID     string
 	EventStatus      EventStatus
 	EventCompletedAt *time.Time

internal/plugins/service.go

@@ -482,6 +482,53 @@ func (s *Service) ScanSourceClient(
 	return client.ScanSource(capabilityID)
 }
 
+func (s *Service) ScanSourceClientByPluginID(
+	ctx context.Context,
+	pluginID string,
+	capabilityID string,
+) (*pluginhost.ScanSourceClient, error) {
+	if s == nil || s.installations == nil {
+		return nil, fmt.Errorf("scan source plugin resolver is not configured")
+	}
+	installations, err := s.installations.ListByPluginID(ctx, pluginID)
+	if err != nil {
+		return nil, err
+	}
+	if len(installations) == 0 {
+		return nil, fmt.Errorf("scan source plugin %q is not installed", pluginID)
+	}
+	if len(installations) > 1 {
+		return nil, fmt.Errorf("scan source plugin %q is ambiguous across %d installations", pluginID, len(installations))
+	}
+
+	var matches []*Installation
+	for _, installation := range installations {
+		if installation == nil {
+			continue
+		}
+		capabilities, err := s.installations.ListCapabilities(ctx, installation.ID)
+		if err != nil {
+			return nil, err
+		}
+		for _, capability := range capabilities {
+			if capability == nil {
+				continue
+			}
+			if capability.Type == "scan_source.v1" && capability.ID == capabilityID {
+				matches = append(matches, installation)
+				break
+			}
+		}
+	}
+	if len(matches) == 0 {
+		return nil, fmt.Errorf("scan source capability %q is not installed for plugin %q", capabilityID, pluginID)
+	}
+	if !matches[0].Enabled {
+		return nil, fmt.Errorf("scan source plugin %q is disabled", pluginID)
+	}
+	return s.ScanSourceClient(ctx, matches[0].ID, capabilityID)
+}
+
 func (s *Service) EventConsumerClient(
 	ctx context.Context,
 	installationID int,

internal/plugins/service_scan_source_test.go

@@ -0,0 +1,78 @@
+package plugins
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	pluginv1 "github.com/Silo-Server/silo-plugin-sdk/pkg/pluginproto/silo/plugin/v1"
+	"github.com/Silo-Server/silo-server/internal/pluginhost"
+)
+
+func TestScanSourceClientByPluginIDResolvesCurrentInstallation(t *testing.T) {
+	manifest := testPluginManifest(t, "silo.autoscan.cephfs", "0.2.2")
+	manifest.Capabilities = []*pluginv1.CapabilityDescriptor{
+		{Type: "scan_source.v1", Id: "cephfs", DisplayName: "CephFS"},
+	}
+	installPath := writeInstalledPluginManifest(t, manifest)
+	host := &fakeServiceHost{
+		clientErr:   pluginhost.ErrClientNotFound,
+		startResult: &fakePluginClient{manifest: manifest},
+	}
+	service := &Service{
+		installations: &fakeServiceInstallationStore{
+			byID: map[int]*Installation{
+				8: {
+					ID:          8,
+					PluginID:    "silo.autoscan.cephfs",
+					Version:     "0.2.2",
+					InstallPath: installPath,
+					Enabled:     true,
+				},
+			},
+			byPluginID: map[string][]*Installation{
+				"silo.autoscan.cephfs": {
+					{ID: 8, PluginID: "silo.autoscan.cephfs", Version: "0.2.2", InstallPath: installPath, Enabled: true},
+				},
+			},
+			listCapabilities: []*Capability{{InstallationID: 8, Type: "scan_source.v1", ID: "cephfs"}},
+		},
+		host: host,
+	}
+
+	if _, err := service.ScanSourceClientByPluginID(context.Background(), "silo.autoscan.cephfs", "cephfs"); err != nil {
+		t.Fatalf("ScanSourceClientByPluginID: %v", err)
+	}
+	if len(host.started) != 1 || host.started[0].InstallationID != 8 {
+		t.Fatalf("started installations = %+v, want installation 8", host.started)
+	}
+}
+
+func TestScanSourceClientByPluginIDRejectsAmbiguousInstallations(t *testing.T) {
+	manifest := testPluginManifest(t, "silo.autoscan.cephfs", "0.2.2")
+	manifest.Capabilities = []*pluginv1.CapabilityDescriptor{
+		{Type: "scan_source.v1", Id: "cephfs", DisplayName: "CephFS"},
+	}
+	installPath := writeInstalledPluginManifest(t, manifest)
+	service := &Service{
+		installations: &fakeServiceInstallationStore{
+			byID: map[int]*Installation{
+				8: {ID: 8, PluginID: "silo.autoscan.cephfs", Version: "0.2.2", InstallPath: installPath, Enabled: true},
+				9: {ID: 9, PluginID: "silo.autoscan.cephfs", Version: "0.2.3", InstallPath: installPath, Enabled: true},
+			},
+			byPluginID: map[string][]*Installation{
+				"silo.autoscan.cephfs": {
+					{ID: 8, PluginID: "silo.autoscan.cephfs", Version: "0.2.2", InstallPath: installPath, Enabled: true},
+					{ID: 9, PluginID: "silo.autoscan.cephfs", Version: "0.2.3", InstallPath: installPath, Enabled: true},
+				},
+			},
+			listCapabilities: []*Capability{{Type: "scan_source.v1", ID: "cephfs"}},
+		},
+		host: &fakeServiceHost{},
+	}
+
+	_, err := service.ScanSourceClientByPluginID(context.Background(), "silo.autoscan.cephfs", "cephfs")
+	if err == nil || !strings.Contains(err.Error(), "ambiguous") {
+		t.Fatalf("ScanSourceClientByPluginID error = %v, want ambiguous", err)
+	}
+}

migrations/sql/20260607201204_autoscan_plugin_id_binding.sql

@@ -0,0 +1,145 @@
+-- +goose Up
+-- +goose StatementBegin
+ALTER TABLE public.autoscan_sources
+    ADD COLUMN plugin_id text NOT NULL DEFAULT '';
+
+ALTER TABLE public.autoscan_events
+    ADD COLUMN plugin_id text NOT NULL DEFAULT '';
+
+UPDATE public.autoscan_sources s
+SET plugin_id = pi.plugin_id
+FROM public.plugin_installations pi
+WHERE s.installation_id = pi.id
+  AND s.plugin_id = '';
+
+WITH inferred AS (
+    SELECT s.id, min(pi.plugin_id) AS plugin_id, count(*) AS matches
+    FROM public.autoscan_sources s
+    JOIN public.plugin_capabilities pc
+      ON pc.capability_type = 'scan_source.v1'
+     AND pc.capability_id = s.capability_id
+    JOIN public.plugin_installations pi
+      ON pi.id = pc.plugin_installation_id
+    WHERE s.plugin_id = ''
+    GROUP BY s.id
+)
+UPDATE public.autoscan_sources s
+SET plugin_id = inferred.plugin_id
+FROM inferred
+WHERE s.id = inferred.id
+  AND inferred.matches = 1;
+
+DO $$
+BEGIN
+    IF EXISTS (
+        SELECT 1
+        FROM public.autoscan_sources
+        WHERE enabled = true
+          AND btrim(plugin_id) = ''
+    ) THEN
+        RAISE EXCEPTION 'enabled autoscan source could not be migrated to plugin_id';
+    END IF;
+END $$;
+
+UPDATE public.autoscan_events e
+SET plugin_id = s.plugin_id
+FROM public.autoscan_sources s
+WHERE e.source_id = s.id
+  AND e.plugin_id = ''
+  AND btrim(s.plugin_id) <> '';
+
+UPDATE public.autoscan_events e
+SET plugin_id = pi.plugin_id
+FROM public.plugin_installations pi
+WHERE e.installation_id = pi.id
+  AND e.plugin_id = '';
+
+WITH inferred AS (
+    SELECT e.id, min(pi.plugin_id) AS plugin_id, count(*) AS matches
+    FROM public.autoscan_events e
+    JOIN public.plugin_capabilities pc
+      ON pc.capability_type = 'scan_source.v1'
+     AND pc.capability_id = e.capability_id
+    JOIN public.plugin_installations pi
+      ON pi.id = pc.plugin_installation_id
+    WHERE e.plugin_id = ''
+    GROUP BY e.id
+)
+UPDATE public.autoscan_events e
+SET plugin_id = inferred.plugin_id
+FROM inferred
+WHERE e.id = inferred.id
+  AND inferred.matches = 1;
+
+CREATE INDEX idx_autoscan_sources_plugin_capability
+    ON public.autoscan_sources (plugin_id, capability_id);
+
+CREATE INDEX idx_autoscan_events_plugin_capability_completed
+    ON public.autoscan_events (plugin_id, capability_id, completed_at DESC);
+
+ALTER TABLE public.autoscan_sources
+    ALTER COLUMN plugin_id DROP DEFAULT;
+
+ALTER TABLE public.autoscan_events
+    ALTER COLUMN plugin_id DROP DEFAULT;
+
+ALTER TABLE public.autoscan_sources
+    DROP COLUMN installation_id;
+
+ALTER TABLE public.autoscan_events
+    DROP COLUMN installation_id;
+-- +goose StatementEnd
+
+-- +goose Down
+-- +goose StatementBegin
+ALTER TABLE public.autoscan_sources
+    ADD COLUMN installation_id integer NOT NULL DEFAULT 0;
+
+ALTER TABLE public.autoscan_events
+    ADD COLUMN installation_id integer NOT NULL DEFAULT 0;
+
+WITH resolved AS (
+    SELECT s.id, min(pc.plugin_installation_id)::integer AS installation_id, count(*) AS matches
+    FROM public.autoscan_sources s
+    JOIN public.plugin_installations pi
+      ON pi.plugin_id = s.plugin_id
+    JOIN public.plugin_capabilities pc
+      ON pc.plugin_installation_id = pi.id
+     AND pc.capability_type = 'scan_source.v1'
+     AND pc.capability_id = s.capability_id
+    WHERE btrim(s.plugin_id) <> ''
+    GROUP BY s.id
+)
+UPDATE public.autoscan_sources s
+SET installation_id = resolved.installation_id
+FROM resolved
+WHERE s.id = resolved.id
+  AND resolved.matches = 1;
+
+WITH resolved AS (
+    SELECT e.id, min(pc.plugin_installation_id)::integer AS installation_id, count(*) AS matches
+    FROM public.autoscan_events e
+    JOIN public.plugin_installations pi
+      ON pi.plugin_id = e.plugin_id
+    JOIN public.plugin_capabilities pc
+      ON pc.plugin_installation_id = pi.id
+     AND pc.capability_type = 'scan_source.v1'
+     AND pc.capability_id = e.capability_id
+    WHERE btrim(e.plugin_id) <> ''
+    GROUP BY e.id
+)
+UPDATE public.autoscan_events e
+SET installation_id = resolved.installation_id
+FROM resolved
+WHERE e.id = resolved.id
+  AND resolved.matches = 1;
+
+DROP INDEX IF EXISTS public.idx_autoscan_events_plugin_capability_completed;
+DROP INDEX IF EXISTS public.idx_autoscan_sources_plugin_capability;
+
+ALTER TABLE public.autoscan_events
+    DROP COLUMN plugin_id;
+
+ALTER TABLE public.autoscan_sources
+    DROP COLUMN plugin_id;
+-- +goose StatementEnd

migrations/sql/20260607204951_autoscan_running_polls.sql

@@ -0,0 +1,36 @@
+-- +goose Up
+-- +goose StatementBegin
+ALTER TABLE public.autoscan_events
+    DROP CONSTRAINT IF EXISTS autoscan_events_status_check;
+
+ALTER TABLE public.autoscan_events
+    ADD CONSTRAINT autoscan_events_status_check
+    CHECK (status = ANY (ARRAY[
+        'running'::text,
+        'success'::text,
+        'error'::text,
+        'unresolved'::text
+    ]));
+-- +goose StatementEnd
+
+-- +goose Down
+-- +goose StatementBegin
+UPDATE public.autoscan_events
+SET status = 'error',
+    error_message = CASE
+        WHEN error_message = '' THEN 'poll started but did not finish'
+        ELSE error_message
+    END
+WHERE status = 'running';
+
+ALTER TABLE public.autoscan_events
+    DROP CONSTRAINT IF EXISTS autoscan_events_status_check;
+
+ALTER TABLE public.autoscan_events
+    ADD CONSTRAINT autoscan_events_status_check
+    CHECK (status = ANY (ARRAY[
+        'success'::text,
+        'error'::text,
+        'unresolved'::text
+    ]));
+-- +goose StatementEnd

web/src/api/types.ts

@@ -1808,7 +1808,7 @@ export interface AutoscanPathRewrite {
 
 export interface AutoscanSource {
   id: string;
-  installation_id: number;
+  plugin_id: string;
   capability_id: string;
   connection_id: string | null;
   enabled: boolean;
@@ -1834,9 +1834,8 @@ export interface AutoscanSourcesResponse {
 }
 
 export interface AutoscanAvailableSource {
-  installation_id: number;
-  capability_id: string;
   plugin_id: string;
+  capability_id: string;
   display_name: string;
 }
 
@@ -1845,7 +1844,7 @@ export interface AutoscanAvailableSourcesResponse {
 }
 
 export interface AutoscanSourceCreateInput {
-  installation_id: number;
+  plugin_id: string;
   capability_id: string;
   connection_id?: string | null;
   enabled: boolean;
@@ -1887,7 +1886,7 @@ export interface AutoscanRewriteSuggestions {
 
 export interface AutoscanStatusSource {
   id: string;
-  installation_id: number;
+  plugin_id: string;
   capability_id: string;
   connection_id: string | null;
   enabled: boolean;
@@ -1896,16 +1895,27 @@ export interface AutoscanStatusSource {
   last_error: string | null;
 }
 
+export interface AutoscanRunningPoll {
+  id: number;
+  source_id: string | null;
+  plugin_id: string;
+  capability_id: string;
+  started_at: string;
+  elapsed_ms: number;
+  marker_before?: string;
+}
+
 export interface AutoscanStatus {
   enabled: boolean;
   sources: AutoscanStatusSource[];
+  running_polls: AutoscanRunningPoll[];
   active_scans: number;
   accepted_scans: number;
   running_scans: number;
   latest_event_at?: string;
 }
 
-export type AutoscanEventStatus = "success" | "error" | "unresolved";
+export type AutoscanEventStatus = "running" | "success" | "error" | "unresolved";
 
 export interface AutoscanEventScanRun {
   id: string;
@@ -1923,7 +1933,7 @@ export interface AutoscanEventScanRun {
 export interface AutoscanEvent {
   id: number;
   source_id: string | null;
-  installation_id: number;
+  plugin_id: string;
   capability_id: string;
   started_at: string;
   completed_at: string;
@@ -1961,7 +1971,7 @@ export interface AutoscanScan {
   completed_at?: string;
   autoscan_event_id?: number;
   source_id?: string;
-  installation_id?: number;
+  plugin_id?: string;
   capability_id?: string;
   event_status?: AutoscanEventStatus;
   event_completed_at?: string;

web/src/hooks/queries/useAutoscan.ts

@@ -312,6 +312,7 @@ export function useAutoscanScans(params?: {
 // --- Trigger ---
 
 export function useTriggerAutoscan() {
+  const queryClient = useQueryClient();
   return useMutation({
     mutationFn: () =>
       api<{ status: string }>("/admin/autoscan/trigger", {
@@ -319,6 +320,8 @@ export function useTriggerAutoscan() {
       }),
     onSuccess: () => {
       toast.success("Autoscan triggered");
+      queryClient.invalidateQueries({ queryKey: adminKeys.autoscanStatus() });
+      queryClient.invalidateQueries({ queryKey: ["admin", "autoscan", "events"] });
     },
     onError: (err) => {
       toast.error(err instanceof Error ? err.message : "Failed to trigger autoscan");

web/src/lib/autoscanLabels.test.ts

@@ -8,38 +8,38 @@ import {
 } from "./autoscanLabels";
 
 describe("composeSourceLabel", () => {
-  const base = { capabilityId: "arr", installationId: 4 };
+  const base = { capabilityId: "arr", pluginId: "silo.autoscan.arr" };
 
   it("uses the operator label first, demoting connection to detail", () => {
     expect(
       composeSourceLabel({ ...base, operatorLabel: "4K Movies", connectionName: "Radarr4k" }),
-    ).toEqual({ name: "4K Movies", detail: "Radarr4k · plugin #4" });
+    ).toEqual({ name: "4K Movies", detail: "Radarr4k · silo.autoscan.arr" });
   });
 
   it("uses the connection name when no operator label", () => {
     expect(
       composeSourceLabel({ ...base, connectionName: "Radarr4k", displayName: "Arr Watcher" }),
-    ).toEqual({ name: "Radarr4k", detail: "Arr Watcher · plugin #4" });
+    ).toEqual({ name: "Radarr4k", detail: "Arr Watcher · silo.autoscan.arr" });
   });
 
   it("uses the manifest display name when no connection", () => {
     expect(
       composeSourceLabel({
         capabilityId: "cephfs",
-        installationId: 5,
+        pluginId: "silo.autoscan.cephfs",
         displayName: "CephFS Watcher",
       }),
-    ).toEqual({ name: "CephFS Watcher", detail: "plugin #5" });
+    ).toEqual({ name: "CephFS Watcher", detail: "silo.autoscan.cephfs" });
   });
 
   it("falls back to capability id when nothing else is set", () => {
-    expect(composeSourceLabel(base)).toEqual({ name: "arr", detail: "plugin #4" });
+    expect(composeSourceLabel(base)).toEqual({ name: "arr", detail: "silo.autoscan.arr" });
   });
 
   it("ignores whitespace-only rungs", () => {
     expect(composeSourceLabel({ ...base, operatorLabel: "   ", connectionName: "  " })).toEqual({
       name: "arr",
-      detail: "plugin #4",
+      detail: "silo.autoscan.arr",
     });
   });
 });
@@ -47,7 +47,7 @@ describe("composeSourceLabel", () => {
 describe("resolveEventSourceName", () => {
   const source: AutoscanSource = {
     id: "src-1",
-    installation_id: 4,
+    plugin_id: "silo.autoscan.arr",
     capability_id: "arr",
     connection_id: "conn-1",
     enabled: true,
@@ -61,13 +61,13 @@ describe("resolveEventSourceName", () => {
   const lookups: SourceLabelLookups = {
     sourceByID: new Map([["src-1", source]]),
     connectionByID: new Map([["conn-1", "Radarr4k"]]),
-    displayNames: new Map([["4:arr", "Arr Watcher"]]),
+    displayNames: new Map([["silo.autoscan.arr:arr", "Arr Watcher"]]),
   };
 
   it("resolves the connection name via the source reference", () => {
     expect(
       resolveEventSourceName(
-        { source_id: "src-1", capability_id: "arr", installation_id: 4 },
+        { source_id: "src-1", capability_id: "arr", plugin_id: "silo.autoscan.arr" },
         lookups,
       ),
     ).toBe("Radarr4k");
@@ -80,7 +80,7 @@ describe("resolveEventSourceName", () => {
     };
     expect(
       resolveEventSourceName(
-        { source_id: "src-1", capability_id: "arr", installation_id: 4 },
+        { source_id: "src-1", capability_id: "arr", plugin_id: "silo.autoscan.arr" },
         withLabel,
       ),
     ).toBe("4K Movies");
@@ -89,7 +89,7 @@ describe("resolveEventSourceName", () => {
   it("falls back to display name when the source was deleted (null source_id)", () => {
     expect(
       resolveEventSourceName(
-        { source_id: null, capability_id: "arr", installation_id: 4 },
+        { source_id: null, capability_id: "arr", plugin_id: "silo.autoscan.arr" },
         lookups,
       ),
     ).toBe("Arr Watcher");
@@ -106,7 +106,7 @@ describe("resolveEventSourceName", () => {
     };
     expect(
       resolveEventSourceName(
-        { source_id: "src-1", capability_id: "arr", installation_id: 4 },
+        { source_id: "src-1", capability_id: "arr", plugin_id: "silo.autoscan.arr" },
         orphaned,
       ),
     ).toBe("Arr Watcher");

web/src/lib/autoscanLabels.ts

@@ -5,7 +5,7 @@ export interface SourceLabelParts {
   connectionName?: string | null;
   displayName?: string | null;
   capabilityId: string;
-  installationId: number;
+  pluginId: string;
 }
 
 export interface SourceLabel {
@@ -22,31 +22,33 @@ export function composeSourceLabel(parts: SourceLabelParts): SourceLabel {
   const operator = parts.operatorLabel?.trim() ?? "";
   const connection = parts.connectionName?.trim() ?? "";
   const display = parts.displayName?.trim() ?? "";
-  const pluginSuffix = `plugin #${parts.installationId}`;
-  const pluginIdentity = display || parts.capabilityId;
+  const pluginIdentity = display || parts.pluginId || parts.capabilityId;
+  const pluginDetail = parts.pluginId || parts.capabilityId;
+  const detailWithPlugin = (detail: string) =>
+    detail === pluginDetail ? detail : `${detail} · ${pluginDetail}`;
 
   if (operator) {
-    return { name: operator, detail: `${connection || pluginIdentity} · ${pluginSuffix}` };
+    return { name: operator, detail: detailWithPlugin(connection || pluginIdentity) };
   }
   if (connection) {
-    return { name: connection, detail: `${pluginIdentity} · ${pluginSuffix}` };
+    return { name: connection, detail: detailWithPlugin(pluginIdentity) };
   }
   if (display) {
-    return { name: display, detail: pluginSuffix };
+    return { name: display, detail: pluginDetail };
   }
-  return { name: parts.capabilityId, detail: pluginSuffix };
+  return { name: parts.capabilityId, detail: pluginDetail };
 }
 
-/** Stable key for the (installation, capability) -> manifest display_name map. */
-export function pluginDisplayNameKey(installationId: number, capabilityId: string): string {
-  return `${installationId}:${capabilityId}`;
+/** Stable key for the (plugin, capability) -> manifest display_name map. */
+export function pluginDisplayNameKey(pluginId: string, capabilityId: string): string {
+  return `${pluginId}:${capabilityId}`;
 }
 
-/** Build the (installation, capability) -> display_name lookup from the picker list. */
+/** Build the (plugin, capability) -> display_name lookup from the picker list. */
 export function buildPluginDisplayNames(available: AutoscanAvailableSource[]): Map<string, string> {
   const map = new Map<string, string>();
   for (const a of available) {
-    map.set(pluginDisplayNameKey(a.installation_id, a.capability_id), a.display_name);
+    map.set(pluginDisplayNameKey(a.plugin_id, a.capability_id), a.display_name);
   }
   return map;
 }
@@ -64,22 +66,22 @@ export interface SourceLabelLookups {
  * own fallback, e.g. "Autoscan").
  */
 export function resolveEventSourceName(
-  ref: { source_id?: string | null; capability_id?: string; installation_id?: number | null },
+  ref: { source_id?: string | null; capability_id?: string; plugin_id?: string | null },
   lookups: SourceLabelLookups,
 ): string {
-  if (!ref.capability_id || ref.installation_id == null) return "";
+  if (!ref.capability_id || !ref.plugin_id) return "";
   const source = ref.source_id ? lookups.sourceByID.get(ref.source_id) : undefined;
   const connectionName = source?.connection_id
     ? lookups.connectionByID.get(source.connection_id)
     : undefined;
   const displayName = lookups.displayNames.get(
-    pluginDisplayNameKey(ref.installation_id, ref.capability_id),
+    pluginDisplayNameKey(ref.plugin_id, ref.capability_id),
   );
   return composeSourceLabel({
     operatorLabel: source?.label,
     connectionName,
     displayName,
     capabilityId: ref.capability_id,
-    installationId: ref.installation_id,
+    pluginId: ref.plugin_id,
   }).name;
 }

web/src/pages/admin/autoscan/ActivityPanel.tsx

@@ -15,6 +15,7 @@ import type {
   AutoscanEvent,
   AutoscanEventScanRun,
   AutoscanEventStatus,
+  AutoscanRunningPoll,
   AutoscanScan,
   AutoscanScanStatus,
   Library,
@@ -95,6 +96,18 @@ function formatDuration(ms: number): string {
   return `${minutes}m ${seconds}s`;
 }
 
+function pollEventDuration(event: AutoscanEvent): string {
+  if (event.status !== "running") {
+    return formatDuration(event.duration_ms);
+  }
+  const elapsed = Math.max(0, Date.now() - new Date(event.started_at).getTime());
+  return formatDuration(elapsed);
+}
+
+function pollEventTimestamp(event: AutoscanEvent): string {
+  return formatTimestamp(event.status === "running" ? event.started_at : event.completed_at);
+}
+
 function eventStatusTone(status: AutoscanEventStatus): {
   label: string;
   icon: typeof CheckCircle2;
@@ -107,6 +120,12 @@ function eventStatusTone(status: AutoscanEventStatus): {
         icon: CheckCircle2,
         className: "border-emerald-500/30 bg-emerald-500/10 text-emerald-500",
       };
+    case "running":
+      return {
+        label: "Running",
+        icon: RefreshCw,
+        className: "border-primary/30 bg-primary/10 text-primary",
+      };
     case "unresolved":
       return {
         label: "Unresolved",
@@ -144,9 +163,15 @@ function scanStatusLabel(status: AutoscanScanStatus | ScanRun["status"]) {
 // arr-plugin sources fan out one-per-connection under a single generic
 // capability, so resolve every Activity row through the shared label chain:
 // operator label -> connection name -> manifest display_name -> capability_id.
-function pollSourceName(event: AutoscanEvent, lookups: SourceLabelLookups): string {
+type PollSourceRef = {
+  source_id?: string | null;
+  plugin_id?: string | null;
+  capability_id?: string;
+};
+
+function pollSourceName(event: PollSourceRef, lookups: SourceLabelLookups): string {
   return (
-    resolveEventSourceName(event, lookups) || `${event.capability_id} #${event.installation_id}`
+    resolveEventSourceName(event, lookups) || event.plugin_id || event.capability_id || "Autoscan"
   );
 }
 
@@ -163,7 +188,7 @@ function PollStatusBadge({ status }: { status: AutoscanEventStatus }) {
   const Icon = tone.icon;
   return (
     <Badge variant="outline" className={tone.className}>
-      <Icon className="h-3.5 w-3.5" />
+      <Icon className={cn("h-3.5 w-3.5", status === "running" && "animate-spin")} />
       {tone.label}
     </Badge>
   );
@@ -415,6 +440,55 @@ function AutoscanQueue({
   );
 }
 
+function RunningPolls({
+  polls,
+  lookups,
+}: {
+  polls: AutoscanRunningPoll[];
+  lookups: SourceLabelLookups;
+}) {
+  if (polls.length === 0) return null;
+
+  return (
+    <section className="space-y-3" role="status" aria-live="polite">
+      <div className="flex flex-wrap items-center justify-between gap-2">
+        <div>
+          <div className="flex items-center gap-2 text-sm font-semibold">
+            <RefreshCw className="text-primary h-4 w-4 animate-spin" />
+            Polling now
+          </div>
+          <p className="text-muted-foreground mt-1 text-xs">
+            Scan-source plugin calls currently in progress.
+          </p>
+        </div>
+        <Badge variant="secondary" className="tabular-nums">
+          {polls.length} active
+        </Badge>
+      </div>
+
+      <div className="grid gap-3 lg:grid-cols-2">
+        {polls.map((poll) => (
+          <div key={poll.id} className="border-border rounded-lg border p-4">
+            <div className="flex flex-wrap items-start justify-between gap-3">
+              <div className="min-w-0">
+                <div className="font-medium">{pollSourceName(poll, lookups)}</div>
+                <div className="text-muted-foreground mt-1 text-xs [overflow-wrap:anywhere]">
+                  {poll.plugin_id} · {poll.capability_id}
+                </div>
+              </div>
+              <PollStatusBadge status="running" />
+            </div>
+            <div className="text-muted-foreground mt-3 flex flex-wrap gap-x-4 gap-y-1 text-xs tabular-nums">
+              <span>Started {formatTimestamp(poll.started_at)}</span>
+              <span>{formatDuration(poll.elapsed_ms)} elapsed</span>
+            </div>
+          </div>
+        ))}
+      </div>
+    </section>
+  );
+}
+
 function ScanHistoryCard({
   scan,
   librariesByID,
@@ -536,8 +610,8 @@ function PollEventCard({ event, lookups }: { event: AutoscanEvent; lookups: Sour
           <PollMetricStrip event={event} />
         </div>
         <div className="text-muted-foreground text-right text-xs tabular-nums">
-          <div>{formatTimestamp(event.completed_at)}</div>
-          <div>{formatDuration(event.duration_ms)}</div>
+          <div>{pollEventTimestamp(event)}</div>
+          <div>{pollEventDuration(event)}</div>
         </div>
       </div>
       {event.error_message ? (
@@ -579,7 +653,7 @@ function PollEventTable({
             <TableHead>Counts</TableHead>
             <TableHead>Scans</TableHead>
             <TableHead>Duration</TableHead>
-            <TableHead>Completed</TableHead>
+            <TableHead>Time</TableHead>
           </>
         }
       >
@@ -608,10 +682,10 @@ function PollEventTable({
               </details>
             </TableCell>
             <TableCell className="whitespace-nowrap tabular-nums">
-              {formatDuration(event.duration_ms)}
+              {pollEventDuration(event)}
             </TableCell>
             <TableCell className="text-muted-foreground whitespace-nowrap tabular-nums">
-              {formatTimestamp(event.completed_at)}
+              {pollEventTimestamp(event)}
             </TableCell>
           </TableRow>
         ))}
@@ -740,10 +814,11 @@ export default function ActivityPanel() {
 
   return (
     <div className="space-y-6">
-      <div className="border-border grid gap-4 rounded-lg border p-4 sm:grid-cols-4">
+      <div className="border-border grid gap-4 rounded-lg border p-4 sm:grid-cols-5">
         <StatTile label="Active" value={queue?.active_scans ?? 0} icon={ScanLine} />
         <StatTile label="Queued" value={queue?.accepted_scans ?? 0} />
-        <StatTile label="Running" value={queue?.running_scans ?? 0} />
+        <StatTile label="Running scans" value={queue?.running_scans ?? 0} />
+        <StatTile label="Polling" value={queue?.running_polls?.length ?? 0} icon={RefreshCw} />
         <div className="flex items-start justify-between gap-3 sm:block">
           <StatTile label="Latest poll" value={formatTime(queue?.latest_event_at)} icon={Clock} />
           <Button
@@ -759,6 +834,8 @@ export default function ActivityPanel() {
         </div>
       </div>
 
+      <RunningPolls polls={queue?.running_polls ?? []} lookups={labelLookups} />
+
       <AutoscanQueue
         scans={autoscanQueue}
         statusActiveCount={queue?.active_scans ?? 0}
@@ -860,6 +937,7 @@ export default function ActivityPanel() {
                 </SelectTrigger>
                 <SelectContent>
                   <SelectItem value="all">All statuses</SelectItem>
+                  <SelectItem value="running">Running</SelectItem>
                   <SelectItem value="success">Success</SelectItem>
                   <SelectItem value="unresolved">Unresolved</SelectItem>
                   <SelectItem value="error">Error</SelectItem>

web/src/pages/admin/autoscan/SourcesPanel.tsx

@@ -93,10 +93,10 @@ function resolveSourceName(
     operatorLabel: source.label,
     connectionName: connectionOptions.find((c) => c.id === (source.connection_id ?? ""))?.name,
     displayName: pluginDisplayNames.get(
-      pluginDisplayNameKey(source.installation_id, source.capability_id),
+      pluginDisplayNameKey(source.plugin_id, source.capability_id),
     ),
     capabilityId: source.capability_id,
-    installationId: source.installation_id,
+    pluginId: source.plugin_id,
   }).name;
 }
 
@@ -152,7 +152,7 @@ const CEPHFS_LEGACY_MOVIE_NESTED_KEY = "movie_nested_paths";
 const CEPHFS_LEGACY_TV_NESTED_KEY = "tv_nested_paths";
 
 function isCephFSSource(source: AutoscanSource): boolean {
-  return source.capability_id === CEPHFS_CAPABILITY_ID;
+  return source.plugin_id === CEPHFS_PLUGIN_ID || source.capability_id === CEPHFS_CAPABILITY_ID;
 }
 
 function isCephFSPlugin(plugin: { plugin_id: string; capability_id: string } | undefined): boolean {
@@ -837,10 +837,10 @@ function SourceRow({
       (c) => c.id === (edit.connectionId || source.connection_id || ""),
     )?.name,
     displayName: pluginDisplayNames.get(
-      pluginDisplayNameKey(source.installation_id, source.capability_id),
+      pluginDisplayNameKey(source.plugin_id, source.capability_id),
     ),
     capabilityId: source.capability_id,
-    installationId: source.installation_id,
+    pluginId: source.plugin_id,
   });
   const sourceIdentity = (
     <div className="min-w-0 space-y-1">
@@ -1061,7 +1061,7 @@ function SourceRow({
 // ---------------------------------------------------------------------------
 
 interface AddSourceForm {
-  /** "installation_id:capability_id" composite key of the chosen plugin. */
+  /** "plugin_id:capability_id" composite key of the chosen plugin. */
   pluginKey: string;
   connectionId: string; // "" / "__none__" means no connection
   intervalStr: string;
@@ -1075,8 +1075,8 @@ const BLANK_ADD_SOURCE: AddSourceForm = {
   sourceConfig: {},
 };
 
-function pluginKey(installationId: number, capabilityId: string): string {
-  return `${installationId}:${capabilityId}`;
+function pluginKey(pluginId: string, capabilityId: string): string {
+  return `${pluginId}:${capabilityId}`;
 }
 
 function AddSourceDialog({
@@ -1094,7 +1094,7 @@ function AddSourceDialog({
 
   const plugins = available.data ?? [];
   const selectedPlugin = plugins.find(
-    (p) => pluginKey(p.installation_id, p.capability_id) === form.pluginKey,
+    (p) => pluginKey(p.plugin_id, p.capability_id) === form.pluginKey,
   );
   const selectedIsCephFS = isCephFSPlugin(selectedPlugin);
 
@@ -1111,7 +1111,7 @@ function AddSourceDialog({
     const pollInterval = raw === "" ? null : Number(raw);
     createSource.mutate(
       {
-        installation_id: selectedPlugin.installation_id,
+        plugin_id: selectedPlugin.plugin_id,
         capability_id: selectedPlugin.capability_id,
         connection_id: connectionId,
         enabled: false,
@@ -1157,9 +1157,7 @@ function AddSourceDialog({
               <Select
                 value={form.pluginKey}
                 onValueChange={(v) => {
-                  const plugin = plugins.find(
-                    (p) => pluginKey(p.installation_id, p.capability_id) === v,
-                  );
+                  const plugin = plugins.find((p) => pluginKey(p.plugin_id, p.capability_id) === v);
                   setForm((f) => ({
                     ...f,
                     pluginKey: v,
@@ -1176,8 +1174,8 @@ function AddSourceDialog({
                 <SelectContent>
                   {plugins.map((p) => (
                     <SelectItem
-                      key={pluginKey(p.installation_id, p.capability_id)}
-                      value={pluginKey(p.installation_id, p.capability_id)}
+                      key={pluginKey(p.plugin_id, p.capability_id)}
+                      value={pluginKey(p.plugin_id, p.capability_id)}
                     >
                       {p.display_name}
                     </SelectItem>