{t(($) => $.plugins.mcp.badge)}
diff --git a/server/cmd/migrate/main.go b/server/cmd/migrate/main.go
index f9adc3c4a..61f07f495 100644
--- a/server/cmd/migrate/main.go
+++ b/server/cmd/migrate/main.go
@@ -252,6 +252,8 @@ var concurrentIndexCleanups = map[string]string{
"396_plugin_package_file_path_index": "idx_plugin_package_file_path",
"397_plugin_installation_package_version_index": "idx_plugin_installation_package_version",
"398_issue_workspace_status_position_index": "idx_issue_workspace_status_position",
+ "400_plugin_hook_schedule_installation_key_index": "idx_plugin_hook_schedule_installation_key",
+ "401_plugin_hook_schedule_enabled_index": "idx_plugin_hook_schedule_enabled",
}
// concurrentDownIndexCleanups covers every migration whose down direction
diff --git a/server/cmd/server/main.go b/server/cmd/server/main.go
index e23e6c213..f3e67d599 100644
--- a/server/cmd/server/main.go
+++ b/server/cmd/server/main.go
@@ -666,6 +666,11 @@ func main() {
if err := schedulerMgr.Register(scheduler.AutopilotScheduleDispatchJob(pool, queries, autopilotSvc)); err != nil {
slog.Warn("scheduler: failed to register autopilot_schedule_dispatch job", "error", err)
}
+ // Manifest-declared Plugin schedules share the same durable lease and retry
+ // machinery. The job is inert while plugins_v1 is disabled.
+ if err := schedulerMgr.Register(scheduler.PluginHookScheduleDispatchJob(queries, h.PluginService)); err != nil {
+ slog.Warn("scheduler: failed to register plugin_hook_schedule_dispatch job", "error", err)
+ }
go func() {
_ = schedulerMgr.Run(sweepCtx)
}()
diff --git a/server/cmd/server/plugin_hook_schedule_job_test.go b/server/cmd/server/plugin_hook_schedule_job_test.go
new file mode 100644
index 000000000..6fd5b3fed
--- /dev/null
+++ b/server/cmd/server/plugin_hook_schedule_job_test.go
@@ -0,0 +1,317 @@
+package main
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "io"
+ "net"
+ "net/http"
+ "net/http/httptest"
+ "net/url"
+ "strconv"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/google/uuid"
+ "github.com/jackc/pgx/v5/pgtype"
+ "github.com/multica-ai/multica/server/internal/featureflags"
+ "github.com/multica-ai/multica/server/internal/scheduler"
+ "github.com/multica-ai/multica/server/internal/service"
+ "github.com/multica-ai/multica/server/internal/util"
+ db "github.com/multica-ai/multica/server/pkg/db/generated"
+ "github.com/multica-ai/multica/server/pkg/featureflag"
+ "github.com/multica-ai/multica/server/pkg/plugincontract"
+)
+
+type receivedScheduledHook struct {
+ InvocationID string `json:"invocation_id"`
+ DeliveryID string `json:"delivery_id"`
+ Attempt int `json:"attempt"`
+ Trigger string `json:"trigger"`
+ Callback string `json:"callback_token"`
+ Actor struct {
+ Type string `json:"type"`
+ ID string `json:"id"`
+ } `json:"actor"`
+ Schedule struct {
+ PlannedAt time.Time `json:"planned_at"`
+ } `json:"schedule"`
+}
+
+// TestPluginHookScheduleTwoReplicasRetryWithStableDelivery is the end-to-end
+// acceptance path for durable scheduled Hooks. Two managers race the same plan,
+// the endpoint loses the connection after receiving attempt one, and the retry
+// is reclaimed by the same DB execution row with the same delivery_id and a
+// fresh invocation.
+func TestPluginHookScheduleTwoReplicasRetryWithStableDelivery(t *testing.T) {
+ ctx := context.Background()
+ queries := db.New(testPool)
+
+ var (
+ mu sync.Mutex
+ received []receivedScheduledHook
+ verificationErr error
+ signingSecret string
+ callbacks = service.NewCallbackTokens()
+ )
+ endpoint := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ raw, err := io.ReadAll(r.Body)
+ if err != nil {
+ http.Error(w, "read", http.StatusBadRequest)
+ return
+ }
+ mu.Lock()
+ secret := signingSecret
+ mu.Unlock()
+ if err := service.VerifyHookSignature(
+ secret,
+ r.Header.Get("X-Multica-Timestamp"),
+ raw,
+ r.Header.Get("X-Multica-Signature"),
+ time.Now(),
+ ); err != nil {
+ mu.Lock()
+ verificationErr = err
+ mu.Unlock()
+ http.Error(w, "signature", http.StatusUnauthorized)
+ return
+ }
+ var body receivedScheduledHook
+ if err := json.Unmarshal(raw, &body); err != nil {
+ http.Error(w, "json", http.StatusBadRequest)
+ return
+ }
+ grant, err := callbacks.Resolve(body.Callback)
+ if err != nil || grant.Trigger != plugincontract.TriggerSchedule ||
+ grant.Actor.Type != "plugin" || grant.IssueID.Valid {
+ mu.Lock()
+ verificationErr = fmt.Errorf("callback grant mismatch: grant=%+v err=%v", grant, err)
+ mu.Unlock()
+ http.Error(w, "callback grant", http.StatusUnauthorized)
+ return
+ }
+ mu.Lock()
+ received = append(received, body)
+ attemptIndex := len(received)
+ mu.Unlock()
+ w.Header().Set("Content-Type", "application/json")
+ if attemptIndex == 1 {
+ // Model the irreducible uncertain-result case: the Plugin received
+ // and may have committed the delivery, but the Host sees no response.
+ connection, _, hijackErr := w.(http.Hijacker).Hijack()
+ if hijackErr != nil {
+ http.Error(w, "hijack", http.StatusInternalServerError)
+ return
+ }
+ _ = connection.Close()
+ return
+ }
+ _, _ = w.Write([]byte(`{"accepted":true}`))
+ }))
+ t.Cleanup(endpoint.Close)
+
+ parsedEndpoint, err := url.Parse(endpoint.URL)
+ if err != nil {
+ t.Fatalf("parse endpoint: %v", err)
+ }
+ _, port, err := net.SplitHostPort(parsedEndpoint.Host)
+ if err != nil {
+ t.Fatalf("split endpoint: %v", err)
+ }
+ hookOrigin := "https://hooks.example.com:" + port
+ hookURL := hookOrigin + "/hooks/scheduled"
+
+ client := endpoint.Client()
+ transport := client.Transport.(*http.Transport).Clone()
+ transport.DialContext = func(ctx context.Context, network, _ string) (net.Conn, error) {
+ return (&net.Dialer{}).DialContext(ctx, network, endpoint.Listener.Addr().String())
+ }
+ client.Transport = transport
+
+ provider := featureflag.NewStaticProvider()
+ provider.Set(featureflags.PluginsV1, featureflag.Rule{Default: true})
+ plugins := service.NewPluginService(queries, testPool)
+ plugins.FeatureFlags = featureflag.NewService(provider)
+ plugins.DeploymentKey = []byte("0123456789abcdef0123456789abcdef")
+ plugins.Callbacks = callbacks
+ plugins.CallbackBaseURL = "https://plugin-api.example.com/v1"
+ plugins.DevOrigins = []string{hookOrigin}
+ plugins.HookClient = client
+
+ manifestRaw := []byte(fmt.Sprintf(`{
+ "manifest_version":1,
+ "key":"com.example.schedule-test",
+ "name":"Schedule Test",
+ "description":"test",
+ "version":"1.0.0",
+ "author":{"name":"test"},
+ "scopes":["net:hooks.example.com"],
+ "contributes":{"hooks":[{
+ "key":"scheduled",
+ "name":"Scheduled",
+ "description":"Scheduled integration test hook.",
+ "triggers":["schedule"],
+ "schedule":{"cron":"*/5 * * * *","timezone":"UTC"},
+ "transport":{"type":"http","url":%s},
+ "timeout_ms":1000
+ }]}
+ }`, strconv.Quote(hookURL)))
+ manifest, canonical, err := plugincontract.ParseManifest(manifestRaw)
+ if err != nil {
+ t.Fatalf("parse fixture manifest: %v", err)
+ }
+ grantedScopes, _ := json.Marshal(manifest.Scopes)
+ installation, err := queries.CreatePluginInstallation(ctx, db.CreatePluginInstallationParams{
+ WorkspaceID: parseUUID(testWorkspaceID),
+ PluginKey: manifest.Key,
+ PackageVersionID: pgtype.UUID{Bytes: uuid.New(), Valid: true},
+ Version: manifest.Version,
+ Manifest: canonical,
+ GrantedScopes: grantedScopes,
+ InstalledBy: parseUUID(testUserID),
+ })
+ if err != nil {
+ t.Fatalf("create plugin installation: %v", err)
+ }
+ derivedSigningSecret, err := plugins.HookSigningSecret(installation.ID)
+ if err != nil {
+ t.Fatalf("derive signing secret: %v", err)
+ }
+ mu.Lock()
+ signingSecret = derivedSigningSecret
+ mu.Unlock()
+ scheduleRow, err := queries.CreatePluginHookSchedule(ctx, db.CreatePluginHookScheduleParams{
+ InstallationID: installation.ID,
+ WorkspaceID: installation.WorkspaceID,
+ HookKey: "scheduled",
+ CronExpression: "*/5 * * * *",
+ Timezone: "UTC",
+ Enabled: true,
+ })
+ if err != nil {
+ t.Fatalf("create plugin schedule: %v", err)
+ }
+ if _, err := testPool.Exec(ctx,
+ `UPDATE plugin_hook_schedule SET activated_at = now() - interval '6 minutes' WHERE id = $1`,
+ scheduleRow.ID,
+ ); err != nil {
+ t.Fatalf("backdate schedule: %v", err)
+ }
+ scopeID := util.UUIDToString(scheduleRow.ID) + ":" + util.UUIDToString(scheduleRow.Generation)
+ t.Cleanup(func() {
+ cleanupCtx := context.Background()
+ _, _ = testPool.Exec(cleanupCtx,
+ `DELETE FROM sys_cron_executions WHERE job_name = $1 AND scope_id = $2`,
+ scheduler.JobNamePluginHookScheduleDispatch, scopeID,
+ )
+ _ = queries.DeletePluginHookSchedulesByInstallation(cleanupCtx, installation.ID)
+ _ = queries.DeletePluginInvocationsByInstallation(cleanupCtx, installation.ID)
+ _ = queries.DeletePluginInstallation(cleanupCtx, installation.ID)
+ })
+
+ managerA := scheduler.NewManager(testPool, scheduler.Options{RunnerID: "plugin-schedule-a"})
+ managerB := scheduler.NewManager(testPool, scheduler.Options{RunnerID: "plugin-schedule-b"})
+ for _, manager := range []*scheduler.Manager{managerA, managerB} {
+ if err := manager.Register(scheduler.PluginHookScheduleDispatchJob(queries, plugins)); err != nil {
+ t.Fatalf("register plugin schedule job: %v", err)
+ }
+ }
+ runManagersTogether(t, ctx, managerA, managerB)
+
+ var status string
+ var attempt int
+ if err := testPool.QueryRow(ctx, `
+ SELECT status, attempt
+ FROM sys_cron_executions
+ WHERE job_name = $1 AND scope_kind = $2 AND scope_id = $3
+ `, scheduler.JobNamePluginHookScheduleDispatch, scheduler.ScopeKindPluginHookSchedule, scopeID).Scan(&status, &attempt); err != nil {
+ t.Fatalf("load failed execution: %v", err)
+ }
+ if status != "FAILED" || attempt != 1 {
+ t.Fatalf("first execution = %s attempt %d, want FAILED attempt 1", status, attempt)
+ }
+ if _, err := testPool.Exec(ctx, `
+ UPDATE sys_cron_executions
+ SET next_retry_at = now() - interval '1 second'
+ WHERE job_name = $1 AND scope_kind = $2 AND scope_id = $3
+ `, scheduler.JobNamePluginHookScheduleDispatch, scheduler.ScopeKindPluginHookSchedule, scopeID); err != nil {
+ t.Fatalf("make retry due: %v", err)
+ }
+ runManagersTogether(t, ctx, managerA, managerB)
+
+ if err := testPool.QueryRow(ctx, `
+ SELECT status, attempt
+ FROM sys_cron_executions
+ WHERE job_name = $1 AND scope_kind = $2 AND scope_id = $3
+ `, scheduler.JobNamePluginHookScheduleDispatch, scheduler.ScopeKindPluginHookSchedule, scopeID).Scan(&status, &attempt); err != nil {
+ t.Fatalf("load successful retry: %v", err)
+ }
+ if status != "SUCCESS" || attempt != 2 {
+ t.Fatalf("retried execution = %s attempt %d, want SUCCESS attempt 2", status, attempt)
+ }
+
+ mu.Lock()
+ requests := append([]receivedScheduledHook(nil), received...)
+ verifyErr := verificationErr
+ mu.Unlock()
+ if verifyErr != nil {
+ t.Fatalf("endpoint rejected HMAC: %v", verifyErr)
+ }
+ if len(requests) != 2 {
+ t.Fatalf("endpoint calls=%d, want one failed attempt and one retry", len(requests))
+ }
+ if requests[0].DeliveryID == "" || requests[0].DeliveryID != requests[1].DeliveryID {
+ t.Fatalf("delivery ids changed across retry: %q != %q", requests[0].DeliveryID, requests[1].DeliveryID)
+ }
+ if requests[0].InvocationID == requests[1].InvocationID {
+ t.Fatal("invocation_id must be unique per HTTP attempt")
+ }
+ if requests[0].Attempt != 1 || requests[1].Attempt != 2 {
+ t.Fatalf("attempts=%d,%d, want 1,2", requests[0].Attempt, requests[1].Attempt)
+ }
+ if requests[1].Trigger != plugincontract.TriggerSchedule ||
+ requests[1].Actor.Type != "plugin" || requests[1].Actor.ID != util.UUIDToString(installation.ID) {
+ t.Fatalf("scheduled actor/trigger mismatch: %+v", requests[1])
+ }
+ if requests[0].Callback == "" || requests[1].Callback == "" || requests[0].Callback == requests[1].Callback {
+ t.Fatal("each attempt must receive a fresh callback token")
+ }
+ if _, err := plugins.Callbacks.Resolve(requests[1].Callback); err == nil {
+ t.Fatal("callback token must be revoked after the endpoint returns")
+ }
+
+ var invocationCount int
+ var distinctDeliveries int
+ if err := testPool.QueryRow(ctx, `
+ SELECT count(*), count(DISTINCT delivery_id)
+ FROM plugin_invocation
+ WHERE installation_id = $1 AND trigger = 'schedule'
+ `, installation.ID).Scan(&invocationCount, &distinctDeliveries); err != nil {
+ t.Fatalf("load invocation audit: %v", err)
+ }
+ if invocationCount != 2 || distinctDeliveries != 1 {
+ t.Fatalf("invocation audit has %d rows/%d deliveries, want 2/1", invocationCount, distinctDeliveries)
+ }
+}
+
+func runManagersTogether(t *testing.T, ctx context.Context, managers ...*scheduler.Manager) {
+ t.Helper()
+ errs := make(chan error, len(managers))
+ var wg sync.WaitGroup
+ for _, manager := range managers {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ errs <- manager.RunOnce(ctx)
+ }()
+ }
+ wg.Wait()
+ close(errs)
+ for err := range errs {
+ if err != nil {
+ t.Fatalf("scheduler tick: %v", err)
+ }
+ }
+}
diff --git a/server/cmd/server/plugin_schedule_pulse_example_test.go b/server/cmd/server/plugin_schedule_pulse_example_test.go
new file mode 100644
index 000000000..a50d00092
--- /dev/null
+++ b/server/cmd/server/plugin_schedule_pulse_example_test.go
@@ -0,0 +1,380 @@
+package main
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "net"
+ "net/http"
+ "net/http/httptest"
+ "net/url"
+ "os"
+ "path/filepath"
+ "runtime"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/google/uuid"
+ "github.com/jackc/pgx/v5/pgtype"
+ "github.com/multica-ai/multica/server/internal/featureflags"
+ "github.com/multica-ai/multica/server/internal/scheduler"
+ "github.com/multica-ai/multica/server/internal/service"
+ "github.com/multica-ai/multica/server/internal/util"
+ db "github.com/multica-ai/multica/server/pkg/db/generated"
+ "github.com/multica-ai/multica/server/pkg/featureflag"
+ "github.com/multica-ai/multica/server/pkg/plugincontract"
+)
+
+type schedulePulseState struct {
+ DeliveryID string `json:"delivery_id"`
+ Count int `json:"count"`
+ LastPlannedAt string `json:"last_planned_at"`
+ LastOccurredAt string `json:"last_occurred_at"`
+ LastAttempt int `json:"last_attempt"`
+}
+
+type schedulePulseHookBody struct {
+ DeliveryID string `json:"delivery_id"`
+ InvocationID string `json:"invocation_id"`
+ Attempt int `json:"attempt"`
+ Trigger string `json:"trigger"`
+ HookKey string `json:"hook_key"`
+ OccurredAt string `json:"occurred_at"`
+ CallbackURL string `json:"callback_url"`
+ Callback string `json:"callback_token"`
+ Schedule struct {
+ PlannedAt string `json:"planned_at"`
+ } `json:"schedule"`
+}
+
+func schedulePulseExampleDir(t *testing.T) string {
+ t.Helper()
+ _, thisFile, _, ok := runtime.Caller(0)
+ if !ok {
+ t.Fatal("cannot locate this test file")
+ }
+ dir := filepath.Join(filepath.Dir(thisFile), "..", "..", "..", "examples", "plugins", "schedule-pulse")
+ if _, err := os.Stat(filepath.Join(dir, plugincontract.ManifestFilename)); err != nil {
+ t.Fatalf("the schedule-pulse example is missing: %v", err)
+ }
+ return dir
+}
+
+// TestSchedulePulseExampleManifestDrivesDurableRetry is the author-facing
+// proof for MUL-6582: the shipped example plugin, not a fixture, is installed
+// and woken by two scheduler replicas. The first HTTP attempt writes workspace
+// storage and then drops the connection; the retry keeps the same delivery_id
+// and does not increment the pulse count.
+func TestSchedulePulseExampleManifestDrivesDurableRetry(t *testing.T) {
+ ctx := context.Background()
+ queries := db.New(testPool)
+
+ var (
+ mu sync.Mutex
+ hookCalls int
+ signingSecret string
+ callbacks = service.NewCallbackTokens()
+ )
+
+ plugins := service.NewPluginService(queries, testPool)
+ provider := featureflag.NewStaticProvider()
+ provider.Set(featureflags.PluginsV1, featureflag.Rule{Default: true})
+ plugins.FeatureFlags = featureflag.NewService(provider)
+ plugins.DeploymentKey = []byte("0123456789abcdef0123456789abcdef")
+ plugins.Callbacks = callbacks
+
+ action := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ token := strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ")
+ grant, err := callbacks.Resolve(token)
+ if err != nil || grant.Trigger != plugincontract.TriggerSchedule || grant.Actor.Type != "plugin" {
+ http.Error(w, "callback grant", http.StatusUnauthorized)
+ return
+ }
+ scopeID, err := service.ResolveStorageScope(service.PluginStorageWorkspace, grant.WorkspaceID, pgtype.UUID{})
+ if err != nil {
+ http.Error(w, err.Error(), http.StatusBadRequest)
+ return
+ }
+ key := strings.TrimPrefix(r.URL.Path, "/storage/workspace/")
+ switch {
+ case r.Method == http.MethodGet:
+ value, getErr := plugins.GetStorageValue(r.Context(), grant.InstallationID, service.PluginStorageWorkspace, scopeID, key)
+ if getErr != nil {
+ var pluginErr *service.PluginError
+ if errors.As(getErr, &pluginErr) && pluginErr.Kind == service.PluginErrorNotFound {
+ http.Error(w, pluginErr.Message, http.StatusNotFound)
+ return
+ }
+ http.Error(w, getErr.Error(), http.StatusBadGateway)
+ return
+ }
+ w.Header().Set("Content-Type", "application/json")
+ _ = json.NewEncoder(w).Encode(map[string]string{"value": value})
+ case r.Method == http.MethodPut:
+ var payload struct {
+ Value string `json:"value"`
+ }
+ if err := json.NewDecoder(r.Body).Decode(&payload); err != nil {
+ http.Error(w, "json", http.StatusBadRequest)
+ return
+ }
+ if err := plugins.SetStorageValue(r.Context(), grant.InstallationID, service.PluginStorageWorkspace, scopeID, key, payload.Value); err != nil {
+ http.Error(w, err.Error(), http.StatusBadGateway)
+ return
+ }
+ w.WriteHeader(http.StatusNoContent)
+ default:
+ http.Error(w, "method", http.StatusMethodNotAllowed)
+ }
+ }))
+ t.Cleanup(action.Close)
+ plugins.CallbackBaseURL = action.URL
+
+ endpoint := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.Method != http.MethodPost || r.URL.Path != "/hooks/pulse" {
+ http.NotFound(w, r)
+ return
+ }
+ raw, err := io.ReadAll(r.Body)
+ if err != nil {
+ http.Error(w, "read", http.StatusBadRequest)
+ return
+ }
+ mu.Lock()
+ secret := signingSecret
+ mu.Unlock()
+ if err := service.VerifyHookSignature(
+ secret,
+ r.Header.Get("X-Multica-Timestamp"),
+ raw,
+ r.Header.Get("X-Multica-Signature"),
+ time.Now(),
+ ); err != nil {
+ http.Error(w, "signature", http.StatusUnauthorized)
+ return
+ }
+ var body schedulePulseHookBody
+ if err := json.Unmarshal(raw, &body); err != nil {
+ http.Error(w, "json", http.StatusBadRequest)
+ return
+ }
+ if body.HookKey != "pulse" || body.Trigger != plugincontract.TriggerSchedule || body.DeliveryID == "" {
+ http.Error(w, "not a scheduled pulse", http.StatusBadRequest)
+ return
+ }
+ req, err := http.NewRequest(http.MethodGet, body.CallbackURL+"/storage/workspace/last_pulse", nil)
+ if err != nil {
+ http.Error(w, err.Error(), http.StatusInternalServerError)
+ return
+ }
+ req.Header.Set("Authorization", "Bearer "+body.Callback)
+ stored, err := http.DefaultClient.Do(req)
+ if err != nil {
+ http.Error(w, err.Error(), http.StatusBadGateway)
+ return
+ }
+ storedBody, _ := io.ReadAll(stored.Body)
+ _ = stored.Body.Close()
+ previous := schedulePulseState{}
+ if stored.StatusCode == http.StatusOK {
+ var wrapped struct {
+ Value string `json:"value"`
+ }
+ if err := json.Unmarshal(storedBody, &wrapped); err == nil {
+ _ = json.Unmarshal([]byte(wrapped.Value), &previous)
+ }
+ } else if stored.StatusCode != http.StatusNotFound {
+ http.Error(w, "storage get", http.StatusBadGateway)
+ return
+ }
+
+ mu.Lock()
+ hookCalls++
+ attemptIndex := hookCalls
+ mu.Unlock()
+
+ w.Header().Set("Content-Type", "application/json")
+ if previous.DeliveryID == body.DeliveryID {
+ _, _ = w.Write([]byte(fmt.Sprintf(`{"duplicate":true,"delivery_id":%q,"count":%d}`, body.DeliveryID, previous.Count)))
+ return
+ }
+ pulse := schedulePulseState{
+ DeliveryID: body.DeliveryID,
+ Count: previous.Count + 1,
+ LastPlannedAt: body.Schedule.PlannedAt,
+ LastOccurredAt: body.OccurredAt,
+ LastAttempt: body.Attempt,
+ }
+ encoded, _ := json.Marshal(pulse)
+ put, err := http.NewRequest(http.MethodPut, body.CallbackURL+"/storage/workspace/last_pulse", strings.NewReader(fmt.Sprintf(`{"value":%s}`, strconvQuote(string(encoded)))))
+ if err != nil {
+ http.Error(w, err.Error(), http.StatusInternalServerError)
+ return
+ }
+ put.Header.Set("Authorization", "Bearer "+body.Callback)
+ put.Header.Set("Content-Type", "application/json")
+ putResp, err := http.DefaultClient.Do(put)
+ if err != nil {
+ http.Error(w, err.Error(), http.StatusBadGateway)
+ return
+ }
+ _ = putResp.Body.Close()
+ if putResp.StatusCode != http.StatusNoContent {
+ http.Error(w, "storage put", http.StatusBadGateway)
+ return
+ }
+ if attemptIndex == 1 {
+ connection, _, hijackErr := w.(http.Hijacker).Hijack()
+ if hijackErr != nil {
+ http.Error(w, "hijack", http.StatusInternalServerError)
+ return
+ }
+ _ = connection.Close()
+ return
+ }
+ _, _ = w.Write([]byte(fmt.Sprintf(`{"accepted":true,"delivery_id":%q,"count":%d}`, pulse.DeliveryID, pulse.Count)))
+ }))
+ t.Cleanup(endpoint.Close)
+
+ parsedEndpoint, err := url.Parse(endpoint.URL)
+ if err != nil {
+ t.Fatalf("parse endpoint: %v", err)
+ }
+ _, port, err := net.SplitHostPort(parsedEndpoint.Host)
+ if err != nil {
+ t.Fatalf("split endpoint: %v", err)
+ }
+ hookOrigin := "https://pulse.example.com:" + port
+ client := endpoint.Client()
+ transport := client.Transport.(*http.Transport).Clone()
+ transport.DialContext = func(ctx context.Context, network, _ string) (net.Conn, error) {
+ return (&net.Dialer{}).DialContext(ctx, network, endpoint.Listener.Addr().String())
+ }
+ client.Transport = transport
+ plugins.DevOrigins = []string{hookOrigin}
+ plugins.HookClient = client
+
+ raw, err := os.ReadFile(filepath.Join(schedulePulseExampleDir(t), plugincontract.ManifestFilename))
+ if err != nil {
+ t.Fatalf("read example manifest: %v", err)
+ }
+ rewritten := strings.ReplaceAll(string(raw), "https://pulse.example.com", hookOrigin)
+ manifest, canonical, err := plugincontract.ParseManifest([]byte(rewritten))
+ if err != nil {
+ t.Fatalf("example manifest does not parse: %v", err)
+ }
+ if err := manifest.CheckCapabilities(plugincontract.HostCapabilities()); err != nil {
+ t.Fatalf("example cannot install on this host: %v", err)
+ }
+ grantedScopes, _ := json.Marshal(manifest.Scopes)
+ installation, err := queries.CreatePluginInstallation(ctx, db.CreatePluginInstallationParams{
+ WorkspaceID: parseUUID(testWorkspaceID),
+ PluginKey: manifest.Key,
+ PackageVersionID: pgtype.UUID{Bytes: uuid.New(), Valid: true},
+ Version: manifest.Version,
+ Manifest: canonical,
+ GrantedScopes: grantedScopes,
+ InstalledBy: parseUUID(testUserID),
+ })
+ if err != nil {
+ t.Fatalf("create plugin installation: %v", err)
+ }
+ derivedSigningSecret, err := plugins.HookSigningSecret(installation.ID)
+ if err != nil {
+ t.Fatalf("derive signing secret: %v", err)
+ }
+ mu.Lock()
+ signingSecret = derivedSigningSecret
+ mu.Unlock()
+ scheduleRow, err := queries.CreatePluginHookSchedule(ctx, db.CreatePluginHookScheduleParams{
+ InstallationID: installation.ID,
+ WorkspaceID: installation.WorkspaceID,
+ HookKey: "pulse",
+ CronExpression: "*/5 * * * *",
+ Timezone: "UTC",
+ Enabled: true,
+ })
+ if err != nil {
+ t.Fatalf("create plugin schedule: %v", err)
+ }
+ if _, err := testPool.Exec(ctx,
+ `UPDATE plugin_hook_schedule SET activated_at = now() - interval '6 minutes' WHERE id = $1`,
+ scheduleRow.ID,
+ ); err != nil {
+ t.Fatalf("backdate schedule: %v", err)
+ }
+ scopeID := util.UUIDToString(scheduleRow.ID) + ":" + util.UUIDToString(scheduleRow.Generation)
+ t.Cleanup(func() {
+ cleanupCtx := context.Background()
+ _, _ = testPool.Exec(cleanupCtx,
+ `DELETE FROM sys_cron_executions WHERE job_name = $1 AND scope_id = $2`,
+ scheduler.JobNamePluginHookScheduleDispatch, scopeID,
+ )
+ _ = queries.DeletePluginHookSchedulesByInstallation(cleanupCtx, installation.ID)
+ _ = queries.DeletePluginStorageByInstallation(cleanupCtx, installation.ID)
+ _ = queries.DeletePluginInvocationsByInstallation(cleanupCtx, installation.ID)
+ _ = queries.DeletePluginInstallation(cleanupCtx, installation.ID)
+ })
+
+ managerA := scheduler.NewManager(testPool, scheduler.Options{RunnerID: "schedule-pulse-a"})
+ managerB := scheduler.NewManager(testPool, scheduler.Options{RunnerID: "schedule-pulse-b"})
+ for _, manager := range []*scheduler.Manager{managerA, managerB} {
+ if err := manager.Register(scheduler.PluginHookScheduleDispatchJob(queries, plugins)); err != nil {
+ t.Fatalf("register plugin schedule job: %v", err)
+ }
+ }
+ runManagersTogether(t, ctx, managerA, managerB)
+ if _, err := testPool.Exec(ctx, `
+ UPDATE sys_cron_executions
+ SET next_retry_at = now() - interval '1 second'
+ WHERE job_name = $1 AND scope_kind = $2 AND scope_id = $3
+ `, scheduler.JobNamePluginHookScheduleDispatch, scheduler.ScopeKindPluginHookSchedule, scopeID); err != nil {
+ t.Fatalf("make retry due: %v", err)
+ }
+ runManagersTogether(t, ctx, managerA, managerB)
+
+ var status string
+ var attempt int
+ if err := testPool.QueryRow(ctx, `
+ SELECT status, attempt
+ FROM sys_cron_executions
+ WHERE job_name = $1 AND scope_kind = $2 AND scope_id = $3
+ `, scheduler.JobNamePluginHookScheduleDispatch, scheduler.ScopeKindPluginHookSchedule, scopeID).Scan(&status, &attempt); err != nil {
+ t.Fatalf("load execution: %v", err)
+ }
+ if status != "SUCCESS" || attempt != 2 {
+ t.Fatalf("execution = %s attempt %d, want SUCCESS attempt 2", status, attempt)
+ }
+
+ mu.Lock()
+ calls := hookCalls
+ mu.Unlock()
+ if calls != 2 {
+ t.Fatalf("hook calls=%d, want 2", calls)
+ }
+
+ scopeIDUUID, err := service.ResolveStorageScope(service.PluginStorageWorkspace, installation.WorkspaceID, pgtype.UUID{})
+ if err != nil {
+ t.Fatalf("resolve storage scope: %v", err)
+ }
+ value, err := plugins.GetStorageValue(ctx, installation.ID, service.PluginStorageWorkspace, scopeIDUUID, "last_pulse")
+ if err != nil {
+ t.Fatalf("read pulse storage: %v", err)
+ }
+ var pulse schedulePulseState
+ if err := json.Unmarshal([]byte(value), &pulse); err != nil {
+ t.Fatalf("decode pulse storage: %v", err)
+ }
+ if pulse.Count != 1 || pulse.DeliveryID == "" {
+ t.Fatalf("pulse storage = %+v, want count 1 with a delivery_id", pulse)
+ }
+}
+
+func strconvQuote(s string) string {
+ encoded, _ := json.Marshal(s)
+ return string(encoded)
+}
diff --git a/server/internal/handler/plugin.go b/server/internal/handler/plugin.go
index 78a8e153f..b40418891 100644
--- a/server/internal/handler/plugin.go
+++ b/server/internal/handler/plugin.go
@@ -80,12 +80,19 @@ type pluginInstallationResponse struct {
// pluginHookResponse omits input_schema: the settings page lists what a hook is
// and who may call it, and the raw JSON Schema is noise there.
type pluginHookResponse struct {
- Key string `json:"key"`
- Name string `json:"name"`
- Description string `json:"description"`
- Triggers []string `json:"triggers"`
- Events []string `json:"events,omitempty"`
- Transport string `json:"transport"`
+ Key string `json:"key"`
+ Name string `json:"name"`
+ Description string `json:"description"`
+ Triggers []string `json:"triggers"`
+ Events []string `json:"events,omitempty"`
+ Transport string `json:"transport"`
+ Schedule *pluginHookScheduleResponse `json:"schedule,omitempty"`
+}
+
+type pluginHookScheduleResponse struct {
+ Cron string `json:"cron"`
+ Timezone string `json:"timezone"`
+ NextRunAt string `json:"next_run_at,omitempty"`
}
func (h *Handler) pluginInstallationPayload(ctx context.Context, installation db.PluginInstallation) (pluginInstallationResponse, error) {
@@ -109,17 +116,35 @@ func (h *Handler) pluginInstallationPayload(ctx context.Context, installation db
if granted == nil {
granted = []string{}
}
+ scheduleRows, err := h.Queries.ListPluginHookSchedulesByInstallation(ctx, installation.ID)
+ if err != nil {
+ return pluginInstallationResponse{}, err
+ }
+ schedules := make(map[string]db.PluginHookSchedule, len(scheduleRows))
+ for _, schedule := range scheduleRows {
+ schedules[schedule.HookKey] = schedule
+ }
hooks := make([]pluginHookResponse, 0, len(manifest.Contributes.Hooks))
for _, hook := range manifest.Contributes.Hooks {
- hooks = append(hooks, pluginHookResponse{
+ response := pluginHookResponse{
Key: hook.Key,
Name: hook.Name,
Description: hook.Description,
Triggers: hook.Triggers,
Events: hook.Events,
Transport: hook.Transport.Type,
- })
+ }
+ if schedule, ok := schedules[hook.Key]; ok {
+ response.Schedule = &pluginHookScheduleResponse{
+ Cron: schedule.CronExpression,
+ Timezone: schedule.Timezone,
+ }
+ if schedule.NextRunAt.Valid {
+ response.Schedule.NextRunAt = schedule.NextRunAt.Time.UTC().Format(timeFormatRFC3339)
+ }
+ }
+ hooks = append(hooks, response)
}
surfaces := manifest.Contributes.Surfaces
if surfaces == nil {
diff --git a/server/internal/handler/plugin_hook.go b/server/internal/handler/plugin_hook.go
index 0db1fbc19..9869acbdb 100644
--- a/server/internal/handler/plugin_hook.go
+++ b/server/internal/handler/plugin_hook.go
@@ -105,15 +105,17 @@ func rawOrNil(raw json.RawMessage) any {
// --- Installation-scoped hook administration ---
type pluginInvocationResponse struct {
- ID string `json:"id"`
- HookKey string `json:"hook_key"`
- Trigger string `json:"trigger"`
- Status string `json:"status"`
- EventType *string `json:"event_type,omitempty"`
- Attempt int `json:"attempt"`
- LatencyMs int `json:"latency_ms"`
- Error *string `json:"error,omitempty"`
- CreatedAt string `json:"created_at"`
+ ID string `json:"id"`
+ HookKey string `json:"hook_key"`
+ Trigger string `json:"trigger"`
+ Status string `json:"status"`
+ EventType *string `json:"event_type,omitempty"`
+ Attempt int `json:"attempt"`
+ LatencyMs int `json:"latency_ms"`
+ Error *string `json:"error,omitempty"`
+ DeliveryID *string `json:"delivery_id,omitempty"`
+ PlannedAt *string `json:"planned_at,omitempty"`
+ CreatedAt string `json:"created_at"`
}
// ListPluginInvocations — GET /api/workspaces/{id}/plugins/{installationId}/invocations
@@ -152,6 +154,14 @@ func (h *Handler) ListPluginInvocations(w http.ResponseWriter, r *http.Request)
value := row.Error.String
item.Error = &value
}
+ if row.DeliveryID.Valid {
+ value := row.DeliveryID.String
+ item.DeliveryID = &value
+ }
+ if row.PlannedAt.Valid {
+ value := row.PlannedAt.Time.UTC().Format("2006-01-02T15:04:05Z07:00")
+ item.PlannedAt = &value
+ }
items = append(items, item)
}
writeJSON(w, http.StatusOK, map[string]any{"invocations": items})
diff --git a/server/internal/handler/plugin_test.go b/server/internal/handler/plugin_test.go
index 7b206becb..5216857be 100644
--- a/server/internal/handler/plugin_test.go
+++ b/server/internal/handler/plugin_test.go
@@ -59,6 +59,25 @@ const hookOnlyTestManifest = `{
}
}`
+const scheduleHookTestManifest = `{
+ "manifest_version": 1,
+ "key": "com.example.scheduled",
+ "name": "Scheduled",
+ "version": "1.0.0",
+ "author": { "name": "example" },
+ "scopes": ["net:example.com"],
+ "contributes": {
+ "hooks": [{
+ "key": "heartbeat",
+ "name": "Heartbeat",
+ "description": "Send a periodic heartbeat.",
+ "triggers": ["schedule"],
+ "schedule": { "cron": "*/5 * * * *", "timezone": "UTC" },
+ "transport": { "type": "http", "url": "https://example.com/hooks/heartbeat" }
+ }]
+ }
+}`
+
func pluginHandlerRequest(method, path string, body []byte, params map[string]string) *http.Request {
request := httptest.NewRequest(method, path, bytes.NewReader(body))
request.Header.Set("X-User-ID", testUserID)
@@ -137,7 +156,7 @@ func withLocalPluginSourceIn(t *testing.T, root string, manifest string) string
testHandler.PluginService.LocalDir = root
testHandler.PluginService.Host = plugincontract.Capabilities{
SurfaceTypes: map[string]bool{plugincontract.SurfaceIssuePanel: true, plugincontract.SurfaceSidebarPanel: true, plugincontract.SurfaceModal: true},
- HookTriggers: map[string]bool{plugincontract.TriggerUI: true, plugincontract.TriggerManual: true, plugincontract.TriggerAgent: true, plugincontract.TriggerEvent: true},
+ HookTriggers: map[string]bool{plugincontract.TriggerUI: true, plugincontract.TriggerManual: true, plugincontract.TriggerAgent: true, plugincontract.TriggerEvent: true, plugincontract.TriggerSchedule: true},
HookTransport: map[string]bool{plugincontract.TransportHTTP: true, plugincontract.TransportMCP: true},
ResourceTypes: map[string]bool{plugincontract.ResourceSkill: true},
}
@@ -182,6 +201,8 @@ func cleanupPluginInstallations(t *testing.T) {
t.Helper()
remove := func() {
ctx := context.Background()
+ testPool.Exec(ctx, `DELETE FROM plugin_invocation WHERE installation_id IN (SELECT id FROM plugin_installation WHERE workspace_id = $1)`, testWorkspaceID)
+ testPool.Exec(ctx, `DELETE FROM plugin_hook_schedule WHERE workspace_id = $1`, testWorkspaceID)
testPool.Exec(ctx, `DELETE FROM plugin_storage WHERE installation_id IN (SELECT id FROM plugin_installation WHERE workspace_id = $1)`, testWorkspaceID)
testPool.Exec(ctx, `DELETE FROM plugin_secret WHERE installation_id IN (SELECT id FROM plugin_installation WHERE workspace_id = $1)`, testWorkspaceID)
testPool.Exec(ctx, `DELETE FROM plugin_installation WHERE workspace_id = $1`, testWorkspaceID)
@@ -193,6 +214,152 @@ func cleanupPluginInstallations(t *testing.T) {
t.Cleanup(remove)
}
+func TestPluginScheduleLifecycleReconcilesAtomically(t *testing.T) {
+ withPluginsV1Flag(t, testHandler, true)
+ cleanupPluginInstallations(t)
+ root := t.TempDir()
+ versionID := withLocalPluginSourceIn(t, root, scheduleHookTestManifest)
+
+ install, _ := json.Marshal(map[string]any{
+ "version_id": versionID,
+ "granted_scopes": []string{"net:example.com"},
+ })
+ recorder := httptest.NewRecorder()
+ testHandler.InstallPlugin(recorder, pluginHandlerRequest(http.MethodPost, "/plugins", install, map[string]string{"id": testWorkspaceID}))
+ if recorder.Code != http.StatusCreated {
+ t.Fatalf("install status=%d body=%s", recorder.Code, recorder.Body.String())
+ }
+ var installed struct {
+ ID string `json:"id"`
+ }
+ if err := json.Unmarshal(recorder.Body.Bytes(), &installed); err != nil {
+ t.Fatalf("decode installation: %v", err)
+ }
+
+ loadSchedule := func() (enabled bool, generation string, cron string) {
+ t.Helper()
+ if err := testPool.QueryRow(context.Background(), `
+ SELECT enabled, generation::text, cron_expression
+ FROM plugin_hook_schedule
+ WHERE installation_id = $1 AND hook_key = 'heartbeat'`, installed.ID).
+ Scan(&enabled, &generation, &cron); err != nil {
+ t.Fatalf("load schedule: %v", err)
+ }
+ return enabled, generation, cron
+ }
+
+ enabled, generation1, cron := loadSchedule()
+ if !enabled || cron != "*/5 * * * *" {
+ t.Fatalf("installed schedule enabled=%v cron=%q", enabled, cron)
+ }
+ params := map[string]string{"id": testWorkspaceID, "installationId": installed.ID}
+
+ recorder = httptest.NewRecorder()
+ testHandler.DisablePlugin(recorder, pluginHandlerRequest(http.MethodPost, "/plugins/disable", nil, params))
+ if recorder.Code != http.StatusOK {
+ t.Fatalf("disable status=%d body=%s", recorder.Code, recorder.Body.String())
+ }
+ enabled, generationDisabled, _ := loadSchedule()
+ if enabled || generationDisabled != generation1 {
+ t.Fatalf("disabled schedule enabled=%v generation=%q, want disabled generation %q", enabled, generationDisabled, generation1)
+ }
+ var hasNextRun bool
+ if err := testPool.QueryRow(context.Background(), `
+ SELECT next_run_at IS NOT NULL FROM plugin_hook_schedule WHERE installation_id = $1`, installed.ID).
+ Scan(&hasNextRun); err != nil {
+ t.Fatalf("load disabled next run: %v", err)
+ }
+ if hasNextRun {
+ t.Fatal("disabled schedule must not advertise a next run")
+ }
+
+ recorder = httptest.NewRecorder()
+ testHandler.EnablePlugin(recorder, pluginHandlerRequest(http.MethodPost, "/plugins/enable", nil, params))
+ if recorder.Code != http.StatusOK {
+ t.Fatalf("enable status=%d body=%s", recorder.Code, recorder.Body.String())
+ }
+ enabled, generation2, _ := loadSchedule()
+ if !enabled || generation2 == generation1 {
+ t.Fatalf("reactivated schedule enabled=%v generation=%q, previous %q", enabled, generation2, generation1)
+ }
+
+ // A code-only upgrade preserves the activation generation and any in-flight
+ // retry. Changing cron rotates it so the new definition cannot collide with
+ // an execution claimed under the previous definition.
+ unchanged := strings.Replace(scheduleHookTestManifest, `"version": "1.0.0"`, `"version": "2.0.0"`, 1)
+ writeLocalPluginManifest(t, root, unchanged)
+ upgradeVersionID := publishLocalPlugin(t, "hello")
+ upgrade, _ := json.Marshal(map[string]any{"version_id": upgradeVersionID, "granted_scopes": []string{"net:example.com"}})
+ recorder = httptest.NewRecorder()
+ testHandler.InstallPlugin(recorder, pluginHandlerRequest(http.MethodPost, "/plugins", upgrade, map[string]string{"id": testWorkspaceID}))
+ if recorder.Code != http.StatusCreated {
+ t.Fatalf("unchanged upgrade status=%d body=%s", recorder.Code, recorder.Body.String())
+ }
+ _, generationUnchanged, _ := loadSchedule()
+ if generationUnchanged != generation2 {
+ t.Fatalf("code-only upgrade rotated generation: got %q want %q", generationUnchanged, generation2)
+ }
+
+ changed := strings.Replace(unchanged, `"version": "2.0.0"`, `"version": "3.0.0"`, 1)
+ changed = strings.Replace(changed, `"cron": "*/5 * * * *"`, `"cron": "*/10 * * * *"`, 1)
+ writeLocalPluginManifest(t, root, changed)
+ changedVersionID := publishLocalPlugin(t, "hello")
+ upgrade, _ = json.Marshal(map[string]any{"version_id": changedVersionID, "granted_scopes": []string{"net:example.com"}})
+ recorder = httptest.NewRecorder()
+ testHandler.InstallPlugin(recorder, pluginHandlerRequest(http.MethodPost, "/plugins", upgrade, map[string]string{"id": testWorkspaceID}))
+ if recorder.Code != http.StatusCreated {
+ t.Fatalf("changed upgrade status=%d body=%s", recorder.Code, recorder.Body.String())
+ }
+ _, generation3, cron := loadSchedule()
+ if generation3 == generation2 || cron != "*/10 * * * *" {
+ t.Fatalf("changed schedule generation=%q cron=%q, previous generation %q", generation3, cron, generation2)
+ }
+
+ removed := strings.Replace(changed, `"version": "3.0.0"`, `"version": "4.0.0"`, 1)
+ removed = strings.Replace(removed, `"triggers": ["schedule"],
+ "schedule": { "cron": "*/10 * * * *", "timezone": "UTC" },`, `"triggers": ["manual"],`, 1)
+ writeLocalPluginManifest(t, root, removed)
+ removedVersionID := publishLocalPlugin(t, "hello")
+ upgrade, _ = json.Marshal(map[string]any{"version_id": removedVersionID, "granted_scopes": []string{"net:example.com"}})
+ recorder = httptest.NewRecorder()
+ testHandler.InstallPlugin(recorder, pluginHandlerRequest(http.MethodPost, "/plugins", upgrade, map[string]string{"id": testWorkspaceID}))
+ if recorder.Code != http.StatusCreated {
+ t.Fatalf("schedule-removing upgrade status=%d body=%s", recorder.Code, recorder.Body.String())
+ }
+ var remaining int
+ if err := testPool.QueryRow(context.Background(), `SELECT COUNT(*) FROM plugin_hook_schedule WHERE installation_id = $1`, installed.ID).Scan(&remaining); err != nil {
+ t.Fatalf("count removed schedules: %v", err)
+ }
+ if remaining != 0 {
+ t.Fatalf("schedule-removing upgrade left %d rows", remaining)
+ }
+
+ // Restore the schedule so uninstall exercises its own cleanup path instead
+ // of succeeding only because the previous upgrade already removed the row.
+ restored := strings.Replace(changed, `"version": "3.0.0"`, `"version": "5.0.0"`, 1)
+ writeLocalPluginManifest(t, root, restored)
+ restoredVersionID := publishLocalPlugin(t, "hello")
+ upgrade, _ = json.Marshal(map[string]any{"version_id": restoredVersionID, "granted_scopes": []string{"net:example.com"}})
+ recorder = httptest.NewRecorder()
+ testHandler.InstallPlugin(recorder, pluginHandlerRequest(http.MethodPost, "/plugins", upgrade, map[string]string{"id": testWorkspaceID}))
+ if recorder.Code != http.StatusCreated {
+ t.Fatalf("schedule-restoring upgrade status=%d body=%s", recorder.Code, recorder.Body.String())
+ }
+ loadSchedule()
+
+ recorder = httptest.NewRecorder()
+ testHandler.UninstallPlugin(recorder, pluginHandlerRequest(http.MethodDelete, "/plugins", nil, params))
+ if recorder.Code != http.StatusNoContent {
+ t.Fatalf("uninstall status=%d body=%s", recorder.Code, recorder.Body.String())
+ }
+ if err := testPool.QueryRow(context.Background(), `SELECT COUNT(*) FROM plugin_hook_schedule WHERE installation_id = $1`, installed.ID).Scan(&remaining); err != nil {
+ t.Fatalf("count schedules after uninstall: %v", err)
+ }
+ if remaining != 0 {
+ t.Fatalf("uninstall left %d schedule rows", remaining)
+ }
+}
+
func TestPluginManagementRequiresPluginsV1(t *testing.T) {
withPluginsV1Flag(t, testHandler, false)
for name, handler := range map[string]http.HandlerFunc{
diff --git a/server/internal/handler/workspace_delete_manifest_test.go b/server/internal/handler/workspace_delete_manifest_test.go
index a0e02060e..6858e3704 100644
--- a/server/internal/handler/workspace_delete_manifest_test.go
+++ b/server/internal/handler/workspace_delete_manifest_test.go
@@ -92,6 +92,7 @@ var workspaceDeletionManifest = map[string]workspaceDeleteAction{
"personal_access_token": workspaceDeleteKeep,
"pinned_item": workspaceDelete,
"plugin_installation": workspaceDelete,
+ "plugin_hook_schedule": workspaceDelete,
"plugin_invocation": workspaceDelete,
"plugin_storage": workspaceDelete,
"plugin_secret": workspaceDelete,
diff --git a/server/internal/scheduler/jobs_plugin_hook.go b/server/internal/scheduler/jobs_plugin_hook.go
new file mode 100644
index 000000000..8ae307d07
--- /dev/null
+++ b/server/internal/scheduler/jobs_plugin_hook.go
@@ -0,0 +1,353 @@
+package scheduler
+
+import (
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "errors"
+ "fmt"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/jackc/pgx/v5"
+ "github.com/jackc/pgx/v5/pgtype"
+ "github.com/multica-ai/multica/server/internal/featureflags"
+ "github.com/multica-ai/multica/server/internal/service"
+ "github.com/multica-ai/multica/server/internal/util"
+ db "github.com/multica-ai/multica/server/pkg/db/generated"
+ "github.com/multica-ai/multica/server/pkg/plugincontract"
+)
+
+const (
+ // JobNamePluginHookScheduleDispatch is persisted in sys_cron_executions and
+ // must remain stable across releases.
+ JobNamePluginHookScheduleDispatch = "plugin_hook_schedule_dispatch"
+ ScopeKindPluginHookSchedule = "plugin_hook_schedule"
+ pluginHookScheduleAttempts = 3
+ pluginHookSchedulePageSize = 1024
+)
+
+type pluginHookScheduleConfig struct {
+ ID pgtype.UUID
+ InstallationID pgtype.UUID
+ HookKey string
+ CronExpression string
+ Timezone string
+ Generation pgtype.UUID
+ ActivatedAt time.Time
+}
+
+type pluginHookScheduleCache struct {
+ mu sync.RWMutex
+ schedules map[string]pluginHookScheduleConfig
+ coalesced map[string]int
+}
+
+func newPluginHookScheduleCache() *pluginHookScheduleCache {
+ return &pluginHookScheduleCache{
+ schedules: make(map[string]pluginHookScheduleConfig),
+ coalesced: make(map[string]int),
+ }
+}
+
+func (c *pluginHookScheduleCache) replace(schedules map[string]pluginHookScheduleConfig) {
+ c.mu.Lock()
+ c.schedules = schedules
+ // Preserve the count across retry ticks, but discard entries for schedules
+ // no longer returned by the database. setCoalesced further bounds this to
+ // one pending entry per active scope, including on a losing replica.
+ for key := range c.coalesced {
+ scopeID, _, _ := strings.Cut(key, "/")
+ if _, active := schedules[scopeID]; !active {
+ delete(c.coalesced, key)
+ }
+ }
+ c.mu.Unlock()
+}
+
+func (c *pluginHookScheduleCache) get(scopeID string) (pluginHookScheduleConfig, bool) {
+ c.mu.RLock()
+ config, ok := c.schedules[scopeID]
+ c.mu.RUnlock()
+ return config, ok
+}
+
+func (c *pluginHookScheduleCache) setCoalesced(scopeID string, plan time.Time, count int) {
+ c.mu.Lock()
+ for key := range c.coalesced {
+ if strings.HasPrefix(key, scopeID+"/") {
+ delete(c.coalesced, key)
+ }
+ }
+ c.coalesced[scopeID+"/"+plan.UTC().Format(time.RFC3339Nano)] = count
+ c.mu.Unlock()
+}
+
+func (c *pluginHookScheduleCache) takeCoalesced(scopeID string, plan time.Time) int {
+ key := scopeID + "/" + plan.UTC().Format(time.RFC3339Nano)
+ c.mu.Lock()
+ count := c.coalesced[key]
+ delete(c.coalesced, key)
+ c.mu.Unlock()
+ return count
+}
+
+// PluginHookScheduleDispatchJob drives manifest-declared Plugin schedules
+// through the shared DB scheduler. Scope includes a generation UUID, so a
+// changed/re-enabled schedule starts a fresh serial timeline while old in-flight
+// delivery is allowed to finish.
+func PluginHookScheduleDispatchJob(queries *db.Queries, plugins *service.PluginService) JobSpec {
+ cache := newPluginHookScheduleCache()
+ return JobSpec{
+ Name: JobNamePluginHookScheduleDispatch,
+ CatchUpMode: CatchUpLatestOnly,
+ RunTimeout: 45 * time.Second,
+ StaleTimeout: 90 * time.Second,
+ HeartbeatInterval: 15 * time.Second,
+ AllowStaleReentry: true,
+ MaxAttempts: pluginHookScheduleAttempts,
+ RetryBackoff: []time.Duration{30 * time.Second, 2 * time.Minute},
+ MaxPlansPerTick: 1,
+ Scopes: pluginHookScheduleScopes(queries, plugins, cache),
+ PlansForScope: pluginHookSchedulePlans(cache),
+ Handler: pluginHookScheduleHandler(queries, plugins, cache),
+ }
+}
+
+func pluginHookScheduleScopes(
+ queries *db.Queries,
+ plugins *service.PluginService,
+ cache *pluginHookScheduleCache,
+) ScopeProvider {
+ return func(ctx context.Context, _ time.Time) ([]Scope, error) {
+ if plugins == nil || !featureflags.PluginsV1Enabled(ctx, plugins.FeatureFlags) {
+ cache.replace(map[string]pluginHookScheduleConfig{})
+ return nil, nil
+ }
+ rows, err := queries.ListEnabledPluginHookSchedules(ctx)
+ if err != nil {
+ return nil, fmt.Errorf("plugin schedule scope: list enabled schedules: %w", err)
+ }
+ configs := make(map[string]pluginHookScheduleConfig, len(rows))
+ scopes := make([]Scope, 0, len(rows))
+ for _, row := range rows {
+ scopeID := pluginHookScheduleScopeID(row.ID, row.Generation)
+ if scopeID == "" || !row.ActivatedAt.Valid {
+ continue
+ }
+ configs[scopeID] = pluginHookScheduleConfig{
+ ID: row.ID,
+ InstallationID: row.InstallationID,
+ HookKey: row.HookKey,
+ CronExpression: row.CronExpression,
+ Timezone: row.Timezone,
+ Generation: row.Generation,
+ ActivatedAt: row.ActivatedAt.Time.UTC(),
+ }
+ scopes = append(scopes, Scope{Kind: ScopeKindPluginHookSchedule, ID: scopeID})
+ }
+ cache.replace(configs)
+ return scopes, nil
+ }
+}
+
+func pluginHookSchedulePlans(cache *pluginHookScheduleCache) func(
+ context.Context, Scope, time.Time, LatestPlanInfo,
+) ([]time.Time, error) {
+ return func(_ context.Context, scope Scope, now time.Time, latest LatestPlanInfo) ([]time.Time, error) {
+ config, ok := cache.get(scope.ID)
+ if !ok {
+ return nil, nil
+ }
+
+ // A generation is strictly serial. Do not plan a newer occurrence while
+ // its latest row is running, nor while that row still owns retry budget.
+ if latest.Found {
+ switch latest.Status {
+ case "RUNNING":
+ return nil, nil
+ case "FAILED":
+ if latest.Attempt < latest.MaxAttempts {
+ if latest.RetryEligible(now) {
+ return []time.Time{latest.PlanTime}, nil
+ }
+ return nil, nil
+ }
+ }
+ }
+
+ after := config.ActivatedAt
+ if latest.Found {
+ after = latest.PlanTime
+ }
+ plan, count, err := latestPluginHookOccurrence(
+ config.CronExpression, config.Timezone, after, now,
+ )
+ if err != nil {
+ return nil, fmt.Errorf("plugin schedule plans for %s: %w", scope.ID, err)
+ }
+ if count == 0 {
+ return nil, nil
+ }
+ cache.setCoalesced(scope.ID, plan, count-1)
+ return []time.Time{plan}, nil
+ }
+}
+
+// latestPluginHookOccurrence enumerates in bounded pages so a long outage still
+// collapses to the truly latest due occurrence instead of stopping at the first
+// 1024 entries returned by service.NextOccurrencesUTC.
+func latestPluginHookOccurrence(cronExpression, timezone string, after, until time.Time) (time.Time, int, error) {
+ var latest time.Time
+ count := 0
+ cursor := after
+ for {
+ occurrences, err := service.NextOccurrencesUTC(cronExpression, timezone, cursor, until)
+ if err != nil {
+ return time.Time{}, 0, err
+ }
+ if len(occurrences) == 0 {
+ return latest, count, nil
+ }
+ count += len(occurrences)
+ latest = occurrences[len(occurrences)-1]
+ if len(occurrences) < pluginHookSchedulePageSize || !latest.Before(until) {
+ return latest, count, nil
+ }
+ cursor = latest
+ }
+}
+
+func pluginHookScheduleHandler(
+ queries *db.Queries,
+ plugins *service.PluginService,
+ cache *pluginHookScheduleCache,
+) Handler {
+ return func(ctx context.Context, in HandlerInput) (HandlerResult, error) {
+ scheduleID, generation, err := parsePluginHookScheduleScopeID(in.Scope.ID)
+ if err != nil {
+ return HandlerResult{}, fmt.Errorf("parse plugin hook schedule scope: %w", err)
+ }
+ if plugins == nil || !featureflags.PluginsV1Enabled(ctx, plugins.FeatureFlags) {
+ return pluginHookScheduleSkipped("feature_disabled"), nil
+ }
+
+ row, err := queries.GetPluginHookSchedule(ctx, scheduleID)
+ if err != nil {
+ if errors.Is(err, pgx.ErrNoRows) {
+ return pluginHookScheduleSkipped("schedule_not_found"), nil
+ }
+ return HandlerResult{}, fmt.Errorf("load plugin hook schedule: %w", err)
+ }
+ if !row.Enabled || row.Generation != generation {
+ return pluginHookScheduleSkipped("schedule_generation_changed"), nil
+ }
+
+ installation, err := queries.GetPluginInstallation(ctx, row.InstallationID)
+ if err != nil {
+ if errors.Is(err, pgx.ErrNoRows) {
+ return pluginHookScheduleSkipped("installation_not_found"), nil
+ }
+ return HandlerResult{}, fmt.Errorf("load plugin installation: %w", err)
+ }
+ if !installation.Enabled {
+ return pluginHookScheduleSkipped("installation_disabled"), nil
+ }
+ hook, err := service.FindHook(installation, row.HookKey)
+ if err != nil || hook.Schedule == nil ||
+ !service.HookAllowsTrigger(hook, plugincontract.TriggerSchedule) ||
+ hook.Transport.Type != plugincontract.TransportHTTP ||
+ hook.Schedule.Cron != row.CronExpression || hook.Schedule.Timezone != row.Timezone {
+ return pluginHookScheduleSkipped("manifest_changed"), nil
+ }
+
+ if plugins.HookBreakerOpen(ctx, installation.ID, hook.Key) {
+ advancePluginHookNextRun(ctx, queries, row, in.PlanTime)
+ result := pluginHookScheduleSkipped("circuit_open")
+ result.Result["coalesced_occurrences"] = cache.takeCoalesced(in.Scope.ID, in.PlanTime)
+ result.Result["dispatch_lag_ms"] = max(time.Since(in.PlanTime).Milliseconds(), 0)
+ return result, nil
+ }
+
+ deliveryID := pluginHookScheduleDeliveryID(installation.ID, hook.Key, row.Generation, in.PlanTime)
+ _, err = plugins.InvokeHook(ctx, service.HookInvocation{
+ Installation: installation,
+ Hook: hook,
+ Trigger: plugincontract.TriggerSchedule,
+ Actor: service.HookActor{Type: "plugin", ID: installation.ID},
+ DeliveryID: deliveryID,
+ PlannedAt: pgtype.Timestamptz{Time: in.PlanTime.UTC(), Valid: true},
+ }, in.Attempt)
+ if err != nil {
+ if in.Job != nil && in.Attempt >= in.Job.MaxAttempts {
+ advancePluginHookNextRun(ctx, queries, row, in.PlanTime)
+ cache.takeCoalesced(in.Scope.ID, in.PlanTime)
+ }
+ return HandlerResult{}, fmt.Errorf("deliver scheduled plugin hook: %w", err)
+ }
+
+ advancePluginHookNextRun(ctx, queries, row, in.PlanTime)
+ return HandlerResult{RowsAffected: 1, Result: map[string]any{
+ "delivery_id": deliveryID,
+ "coalesced_occurrences": cache.takeCoalesced(in.Scope.ID, in.PlanTime),
+ "dispatch_lag_ms": max(time.Since(in.PlanTime).Milliseconds(), 0),
+ }}, nil
+ }
+}
+
+func pluginHookScheduleSkipped(reason string) HandlerResult {
+ return HandlerResult{RowsAffected: 0, Result: map[string]any{"skipped_reason": reason}}
+}
+
+func advancePluginHookNextRun(ctx context.Context, queries *db.Queries, row db.PluginHookSchedule, plan time.Time) {
+ next, err := service.NextOccurrenceAfterUTC(row.CronExpression, row.Timezone, plan)
+ if err != nil {
+ return
+ }
+ _, _ = queries.UpdatePluginHookScheduleNextRun(ctx, db.UpdatePluginHookScheduleNextRunParams{
+ ID: row.ID,
+ Generation: row.Generation,
+ NextRunAt: pgtype.Timestamptz{Time: next.UTC(), Valid: true},
+ })
+}
+
+func pluginHookScheduleScopeID(id, generation pgtype.UUID) string {
+ idString := util.UUIDToString(id)
+ generationString := util.UUIDToString(generation)
+ if idString == "" || generationString == "" {
+ return ""
+ }
+ return idString + ":" + generationString
+}
+
+func parsePluginHookScheduleScopeID(scopeID string) (pgtype.UUID, pgtype.UUID, error) {
+ parts := strings.Split(scopeID, ":")
+ if len(parts) != 2 {
+ return pgtype.UUID{}, pgtype.UUID{}, errors.New("expected schedule:generation scope")
+ }
+ id, err := util.ParseUUID(parts[0])
+ if err != nil {
+ return pgtype.UUID{}, pgtype.UUID{}, err
+ }
+ generation, err := util.ParseUUID(parts[1])
+ if err != nil {
+ return pgtype.UUID{}, pgtype.UUID{}, err
+ }
+ return id, generation, nil
+}
+
+func pluginHookScheduleDeliveryID(
+ installationID pgtype.UUID,
+ hookKey string,
+ generation pgtype.UUID,
+ plan time.Time,
+) string {
+ digest := sha256.Sum256([]byte(strings.Join([]string{
+ util.UUIDToString(installationID),
+ hookKey,
+ util.UUIDToString(generation),
+ plan.UTC().Format(time.RFC3339Nano),
+ }, "\x00")))
+ return "psd_" + hex.EncodeToString(digest[:])
+}
diff --git a/server/internal/scheduler/jobs_plugin_hook_test.go b/server/internal/scheduler/jobs_plugin_hook_test.go
new file mode 100644
index 000000000..a5dce357e
--- /dev/null
+++ b/server/internal/scheduler/jobs_plugin_hook_test.go
@@ -0,0 +1,127 @@
+package scheduler
+
+import (
+ "context"
+ "testing"
+ "time"
+
+ "github.com/multica-ai/multica/server/internal/featureflags"
+ "github.com/multica-ai/multica/server/internal/service"
+ "github.com/multica-ai/multica/server/internal/util"
+ "github.com/multica-ai/multica/server/pkg/featureflag"
+)
+
+func TestPluginHookSchedulePlansCollapseToLatestAndStaySerial(t *testing.T) {
+ cache := newPluginHookScheduleCache()
+ activated := time.Date(2026, 8, 23, 10, 0, 0, 0, time.UTC)
+ scope := Scope{Kind: ScopeKindPluginHookSchedule, ID: "schedule:generation"}
+ cache.replace(map[string]pluginHookScheduleConfig{
+ scope.ID: {
+ CronExpression: "*/5 * * * *",
+ Timezone: "UTC",
+ ActivatedAt: activated,
+ },
+ })
+ plans := pluginHookSchedulePlans(cache)
+ now := activated.Add(16 * time.Minute)
+
+ got, err := plans(context.Background(), scope, now, LatestPlanInfo{})
+ if err != nil {
+ t.Fatalf("plan initial occurrence: %v", err)
+ }
+ if len(got) != 1 || !got[0].Equal(activated.Add(15*time.Minute)) {
+ t.Fatalf("got %v, want only the latest 10:15 occurrence", got)
+ }
+ if count := cache.takeCoalesced(scope.ID, got[0]); count != 2 {
+ t.Fatalf("coalesced=%d, want 2 older due occurrences", count)
+ }
+
+ latest := LatestPlanInfo{
+ Found: true,
+ PlanTime: activated.Add(5 * time.Minute),
+ Status: "RUNNING",
+ Attempt: 1,
+ MaxAttempts: pluginHookScheduleAttempts,
+ }
+ got, err = plans(context.Background(), scope, now, latest)
+ if err != nil {
+ t.Fatalf("plan while running: %v", err)
+ }
+ if len(got) != 0 {
+ t.Fatalf("a running latest plan must block newer plans, got %v", got)
+ }
+
+ latest.Status = "FAILED"
+ latest.NextRetryAt = now.Add(time.Minute)
+ got, err = plans(context.Background(), scope, now, latest)
+ if err != nil {
+ t.Fatalf("plan before retry: %v", err)
+ }
+ if len(got) != 0 {
+ t.Fatalf("a retryable failure must block newer plans before backoff, got %v", got)
+ }
+
+ latest.NextRetryAt = now.Add(-time.Second)
+ got, err = plans(context.Background(), scope, now, latest)
+ if err != nil {
+ t.Fatalf("plan due retry: %v", err)
+ }
+ if len(got) != 1 || !got[0].Equal(latest.PlanTime) {
+ t.Fatalf("retry must reuse plan_time %s, got %v", latest.PlanTime, got)
+ }
+}
+
+func TestLatestPluginHookOccurrencePagesThroughLongOutage(t *testing.T) {
+ activated := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
+ now := activated.Add(9 * 24 * time.Hour)
+ got, count, err := latestPluginHookOccurrence("*/5 * * * *", "UTC", activated, now)
+ if err != nil {
+ t.Fatalf("enumerate occurrences: %v", err)
+ }
+ if !got.Equal(now) {
+ t.Fatalf("latest=%s, want %s", got, now)
+ }
+ if count != 9*24*12 {
+ t.Fatalf("count=%d, want %d", count, 9*24*12)
+ }
+}
+
+func TestPluginHookScheduleDeliveryIDIsStablePerOccurrence(t *testing.T) {
+ installation := util.MustParseUUID("11111111-1111-4111-8111-111111111111")
+ generation := util.MustParseUUID("22222222-2222-4222-8222-222222222222")
+ plan := time.Date(2026, 8, 23, 10, 15, 0, 0, time.UTC)
+
+ first := pluginHookScheduleDeliveryID(installation, "digest", generation, plan)
+ second := pluginHookScheduleDeliveryID(installation, "digest", generation, plan)
+ if first != second {
+ t.Fatalf("same occurrence changed delivery id: %q != %q", first, second)
+ }
+ if first == pluginHookScheduleDeliveryID(installation, "digest", generation, plan.Add(5*time.Minute)) {
+ t.Fatal("different plan_time must produce a different delivery id")
+ }
+ if first == pluginHookScheduleDeliveryID(
+ installation,
+ "digest",
+ util.MustParseUUID("33333333-3333-4333-8333-333333333333"),
+ plan,
+ ) {
+ t.Fatal("different generation must produce a different delivery id")
+ }
+}
+
+func TestPluginHookScheduleHandlerRechecksFeatureFlagAfterClaim(t *testing.T) {
+ provider := featureflag.NewStaticProvider()
+ provider.Set(featureflags.PluginsV1, featureflag.Rule{Default: false})
+ plugins := &service.PluginService{FeatureFlags: featureflag.NewService(provider)}
+ handler := pluginHookScheduleHandler(nil, plugins, newPluginHookScheduleCache())
+
+ result, err := handler(context.Background(), HandlerInput{
+ Scope: Scope{ID: "11111111-1111-4111-8111-111111111111:22222222-2222-4222-8222-222222222222"},
+ })
+ if err != nil {
+ t.Fatalf("feature-disabled handler: %v", err)
+ }
+ if result.Result["skipped_reason"] != "feature_disabled" {
+ t.Fatalf("skipped_reason=%v, want feature_disabled", result.Result["skipped_reason"])
+ }
+}
diff --git a/server/internal/service/plugin.go b/server/internal/service/plugin.go
index c7280907f..c14fe8d42 100644
--- a/server/internal/service/plugin.go
+++ b/server/internal/service/plugin.go
@@ -402,6 +402,9 @@ func (s *PluginService) InstallPlugin(ctx context.Context, workspaceID, userID p
if skillErr := s.InstallSkillResources(ctx, queries, installation, manifest, userID); skillErr != nil {
return db.PluginInstallation{}, skillErr
}
+ if scheduleErr := s.reconcilePluginHookSchedules(ctx, queries, installation, manifest); scheduleErr != nil {
+ return db.PluginInstallation{}, scheduleErr
+ }
if commitErr := tx.Commit(ctx); commitErr != nil {
return db.PluginInstallation{}, &PluginError{Kind: PluginErrorUnavailable, Message: "commit install", Err: commitErr}
}
@@ -458,6 +461,9 @@ func (s *PluginService) InstallPlugin(ctx context.Context, workspaceID, userID p
if err := s.InstallSkillResources(ctx, queries, updated, manifest, userID); err != nil {
return db.PluginInstallation{}, err
}
+ if err := s.reconcilePluginHookSchedules(ctx, queries, updated, manifest); err != nil {
+ return db.PluginInstallation{}, err
+ }
if err := tx.Commit(ctx); err != nil {
return db.PluginInstallation{}, &PluginError{Kind: PluginErrorUnavailable, Message: "commit upgrade", Err: err}
}
@@ -691,19 +697,35 @@ func (s *PluginService) ConfiguredSecretKeys(ctx context.Context, installationID
// SetEnabled toggles an installation. Disabling hides every contribution
// immediately but keeps storage and secrets, so re-enabling is not a reinstall.
func (s *PluginService) SetEnabled(ctx context.Context, installation db.PluginInstallation, enabled bool) (db.PluginInstallation, error) {
- updated, err := s.Queries.SetPluginInstallationEnabled(ctx, db.SetPluginInstallationEnabledParams{
+ if installation.Enabled == enabled {
+ return installation, nil
+ }
+ tx, err := s.TxStarter.Begin(ctx)
+ if err != nil {
+ return db.PluginInstallation{}, &PluginError{Kind: PluginErrorUnavailable, Message: "begin plugin state update", Err: err}
+ }
+ defer func() { _ = tx.Rollback(ctx) }()
+ queries := s.Queries.WithTx(tx)
+
+ updated, err := queries.SetPluginInstallationEnabled(ctx, db.SetPluginInstallationEnabledParams{
ID: installation.ID,
Enabled: enabled,
})
if err != nil {
return db.PluginInstallation{}, &PluginError{Kind: PluginErrorUnavailable, Message: "update plugin state", Err: err}
}
+ if err := s.setPluginHookSchedulesEnabled(ctx, queries, installation.ID, enabled); err != nil {
+ return db.PluginInstallation{}, err
+ }
+ if err := tx.Commit(ctx); err != nil {
+ return db.PluginInstallation{}, &PluginError{Kind: PluginErrorUnavailable, Message: "commit plugin state update", Err: err}
+ }
return updated, nil
}
-// Uninstall removes the installation with its storage and secrets. There are no
-// foreign keys or cascades by repository policy, so the three deletes share one
-// transaction: a partial uninstall would strand plugin-owned rows nothing can
+// Uninstall removes the installation and all of its application-owned rows.
+// There are no foreign keys or cascades by repository policy, so the deletes
+// share one transaction: a partial uninstall would strand rows nothing can
// reach or clean up later.
func (s *PluginService) Uninstall(ctx context.Context, installation db.PluginInstallation) error {
tx, err := s.TxStarter.Begin(ctx)
@@ -713,6 +735,9 @@ func (s *PluginService) Uninstall(ctx context.Context, installation db.PluginIns
defer func() { _ = tx.Rollback(ctx) }()
queries := s.Queries.WithTx(tx)
+ if err := queries.DeletePluginHookSchedulesByInstallation(ctx, installation.ID); err != nil {
+ return &PluginError{Kind: PluginErrorUnavailable, Message: "delete plugin hook schedules", Err: err}
+ }
if err := queries.DeletePluginStorageByInstallation(ctx, installation.ID); err != nil {
return &PluginError{Kind: PluginErrorUnavailable, Message: "delete plugin storage", Err: err}
}
diff --git a/server/internal/service/plugin_hook.go b/server/internal/service/plugin_hook.go
index 0b7b5ea4b..166ccfda2 100644
--- a/server/internal/service/plugin_hook.go
+++ b/server/internal/service/plugin_hook.go
@@ -16,6 +16,7 @@ import (
"strings"
"time"
+ "github.com/google/uuid"
"github.com/jackc/pgx/v5/pgtype"
db "github.com/multica-ai/multica/server/pkg/db/generated"
"github.com/multica-ai/multica/server/pkg/plugincontract"
@@ -50,9 +51,10 @@ const (
hookEventAttempts = 3
hookEventBackoff = 2 * time.Second
- // Circuit breaker. After this many failures inside the window, event
+ // Circuit breaker. After this many failures inside the window, background
// delivery for the hook stops until the window rolls off — an endpoint that
- // has been down for a minute does not need one request per event to notice.
+ // has been down for a minute does not need one request per event or schedule
+ // occurrence to notice.
hookBreakerThreshold = 5
hookBreakerWindow = 5 * time.Minute
@@ -76,9 +78,9 @@ type HookResult struct {
//
// This is the whole identity model for hooks, and it follows the trigger rather
// than the plugin: a ui/manual hook runs because a person pressed something, so
-// its writes stay that person's (marked via_plugin_id); an event hook has no
-// person behind it and must not borrow the last one who happened to touch the
-// issue.
+// its writes stay that person's (marked via_plugin_id); event/schedule hooks
+// have no person behind them and must not borrow the last one who touched the
+// resource.
type HookActor struct {
Type string
ID pgtype.UUID
@@ -86,6 +88,10 @@ type HookActor struct {
// HookInvocation is one call the engine has been asked to make.
type HookInvocation struct {
+ // ID is allocated before the outbound request. The receiver can log it and
+ // correlate a response with the durable invocation row even when the call
+ // itself fails.
+ ID pgtype.UUID
Installation db.PluginInstallation
Hook plugincontract.Hook
Trigger string
@@ -97,18 +103,27 @@ type HookInvocation struct {
// grant to reach across the workspace.
IssueID pgtype.UUID
Input any
+ // DeliveryID is stable across retries of the same scheduled occurrence.
+ // PlannedAt is the canonical UTC cron occurrence, not delivery wall time.
+ DeliveryID string
+ PlannedAt pgtype.Timestamptz
+ Attempt int
}
// hookRequestBody is the wire format. Stable and small on purpose: a hook
// handler written against v1 should not have to care what else the host learns
// how to send later.
type hookRequestBody struct {
- Version int `json:"version"`
- HookKey string `json:"hook_key"`
- Trigger string `json:"trigger"`
- EventType string `json:"event_type,omitempty"`
- WorkspaceID string `json:"workspace_id"`
- Installation string `json:"installation_id"`
+ Version int `json:"version"`
+ InvocationID string `json:"invocation_id"`
+ DeliveryID string `json:"delivery_id,omitempty"`
+ Attempt int `json:"attempt"`
+ OccurredAt time.Time `json:"occurred_at"`
+ HookKey string `json:"hook_key"`
+ Trigger string `json:"trigger"`
+ EventType string `json:"event_type,omitempty"`
+ WorkspaceID string `json:"workspace_id"`
+ Installation string `json:"installation_id"`
// IssueID is the issue this call is about, as resolved and permission-checked
// by the host. Sent because the alternative is every handler reading it out
// of client-supplied `input` — unvalidated, and absent entirely for the event
@@ -130,8 +145,9 @@ type hookRequestBody struct {
// minutes it is valid. Narrower than the installation's own token and tied
// to this one call, so a handler that leaks it leaks a few minutes of the
// scopes it was already using, not standing access.
- CallbackToken string `json:"callback_token,omitempty"`
- CallbackURL string `json:"callback_url,omitempty"`
+ CallbackToken string `json:"callback_token,omitempty"`
+ CallbackURL string `json:"callback_url,omitempty"`
+ Schedule *hookRequestSchedule `json:"schedule,omitempty"`
}
type hookRequestActor struct {
@@ -139,6 +155,10 @@ type hookRequestActor struct {
ID string `json:"id,omitempty"`
}
+type hookRequestSchedule struct {
+ PlannedAt time.Time `json:"planned_at"`
+}
+
// FindHook returns the named hook from an installation's consented manifest.
//
// Read from the installation row, never from the manifest the source URL serves
@@ -172,10 +192,16 @@ func HookAllowsTrigger(hook plugincontract.Hook, trigger string) bool {
// InvokeHook performs one call and records it.
//
// Callers pick the blocking behaviour, not this function: ui/manual await the
-// result because a person is watching, and the event dispatcher runs it on its
-// own goroutine because the host request that produced the event must not wait
-// for a third party.
+// result because a person is watching, event delivery runs outside the host
+// request, and schedule delivery runs under the durable scheduler lease.
func (s *PluginService) InvokeHook(ctx context.Context, invocation HookInvocation, attempt int) (HookResult, error) {
+ if !invocation.ID.Valid {
+ invocation.ID = pgtype.UUID{Bytes: uuid.New(), Valid: true}
+ }
+ if attempt < 1 {
+ attempt = 1
+ }
+ invocation.Attempt = attempt
hook := invocation.Hook
result := HookResult{HookKey: hook.Key, Trigger: invocation.Trigger, Attempts: attempt}
@@ -355,8 +381,16 @@ func nonSecretConfig(installation db.PluginInstallation) map[string]any {
}
func (s *PluginService) buildHookBody(ctx context.Context, invocation HookInvocation) (hookRequestBody, error) {
+ attempt := invocation.Attempt
+ if attempt < 1 {
+ attempt = 1
+ }
body := hookRequestBody{
Version: 1,
+ InvocationID: uuidString(invocation.ID),
+ DeliveryID: invocation.DeliveryID,
+ Attempt: attempt,
+ OccurredAt: time.Now().UTC(),
HookKey: invocation.Hook.Key,
Trigger: invocation.Trigger,
EventType: invocation.EventType,
@@ -364,6 +398,9 @@ func (s *PluginService) buildHookBody(ctx context.Context, invocation HookInvoca
Installation: uuidString(invocation.Installation.ID),
Actor: hookRequestActor{Type: invocation.Actor.Type, ID: uuidString(invocation.Actor.ID)},
}
+ if invocation.PlannedAt.Valid {
+ body.Schedule = &hookRequestSchedule{PlannedAt: invocation.PlannedAt.Time.UTC()}
+ }
if invocation.IssueID.Valid {
body.IssueID = uuidString(invocation.IssueID)
}
@@ -485,8 +522,8 @@ func (s *PluginService) checkHookRate(ctx context.Context, installationID pgtype
return nil
}
-// HookBreakerOpen reports whether event delivery for this hook is currently
-// suspended after repeated failures.
+// HookBreakerOpen reports whether background delivery for this hook is
+// currently suspended after repeated failures.
func (s *PluginService) HookBreakerOpen(ctx context.Context, installationID pgtype.UUID, hookKey string) bool {
since := pgtype.Timestamptz{Time: time.Now().Add(-hookBreakerWindow), Valid: true}
failures, err := s.Queries.CountRecentPluginFailures(ctx, db.CountRecentPluginFailuresParams{
@@ -502,6 +539,7 @@ func (s *PluginService) HookBreakerOpen(ctx context.Context, installationID pgty
func (s *PluginService) recordInvocation(ctx context.Context, invocation HookInvocation, status string, attempt, latency int, message string) {
params := db.CreatePluginInvocationParams{
+ ID: invocation.ID,
InstallationID: invocation.Installation.ID,
WorkspaceID: invocation.Installation.WorkspaceID,
HookKey: invocation.Hook.Key,
@@ -510,6 +548,12 @@ func (s *PluginService) recordInvocation(ctx context.Context, invocation HookInv
Attempt: int32(attempt),
LatencyMs: int32(latency),
}
+ if invocation.DeliveryID != "" {
+ params.DeliveryID = pgtype.Text{String: invocation.DeliveryID, Valid: true}
+ }
+ if invocation.PlannedAt.Valid {
+ params.PlannedAt = invocation.PlannedAt
+ }
if invocation.EventType != "" {
params.EventType = pgtype.Text{String: invocation.EventType, Valid: true}
}
diff --git a/server/internal/service/plugin_hook_test.go b/server/internal/service/plugin_hook_test.go
index d7360bbfa..0b2d62eda 100644
--- a/server/internal/service/plugin_hook_test.go
+++ b/server/internal/service/plugin_hook_test.go
@@ -1,6 +1,7 @@
package service
import (
+ "context"
"encoding/hex"
"strings"
"testing"
@@ -137,13 +138,56 @@ func TestHookAllowsTriggerOnlyWhenDeclared(t *testing.T) {
if !HookAllowsTrigger(hook, plugincontract.TriggerManual) {
t.Fatal("a declared trigger must be allowed")
}
- for _, undeclared := range []string{plugincontract.TriggerUI, plugincontract.TriggerEvent, plugincontract.TriggerAgent} {
+ for _, undeclared := range []string{plugincontract.TriggerUI, plugincontract.TriggerEvent, plugincontract.TriggerAgent, plugincontract.TriggerSchedule} {
if HookAllowsTrigger(hook, undeclared) {
t.Fatalf("%s was not declared and must not be allowed", undeclared)
}
}
}
+func TestScheduledHookBodyExtendsVersionOneProtocol(t *testing.T) {
+ installationID := testInstallationID(t)
+ invocationID := parseUUIDValueForTest(t, "22222222-2222-4222-8222-222222222222")
+ plannedAt := time.Date(2026, 8, 23, 10, 15, 0, 0, time.UTC)
+ body, err := (&PluginService{}).buildHookBody(context.Background(), HookInvocation{
+ ID: invocationID,
+ Installation: db.PluginInstallation{
+ ID: installationID,
+ WorkspaceID: parseUUIDValueForTest(t, "33333333-3333-4333-8333-333333333333"),
+ },
+ Hook: plugincontract.Hook{Key: "digest"},
+ Trigger: plugincontract.TriggerSchedule,
+ Actor: HookActor{Type: "plugin", ID: installationID},
+ DeliveryID: "psd_stable",
+ PlannedAt: pgtype.Timestamptz{Time: plannedAt, Valid: true},
+ Attempt: 2,
+ })
+ if err != nil {
+ t.Fatalf("build scheduled hook body: %v", err)
+ }
+ if body.Version != 1 {
+ t.Fatalf("version=%d, want existing protocol version 1", body.Version)
+ }
+ if body.InvocationID != uuidString(invocationID) || body.DeliveryID != "psd_stable" || body.Attempt != 2 {
+ t.Fatalf("unexpected identity fields: %+v", body)
+ }
+ if body.OccurredAt.IsZero() {
+ t.Fatal("occurred_at must be populated per request")
+ }
+ if body.Schedule == nil || !body.Schedule.PlannedAt.Equal(plannedAt) {
+ t.Fatalf("schedule.planned_at=%v, want %s", body.Schedule, plannedAt)
+ }
+}
+
+func parseUUIDValueForTest(t *testing.T, value string) pgtype.UUID {
+ t.Helper()
+ parsed, err := parseUUIDValue(value)
+ if err != nil {
+ t.Fatalf("parse uuid: %v", err)
+ }
+ return parsed
+}
+
// FindHook reads the CONSENTED manifest on the installation row. A hook that is
// not in it does not exist, whatever the source URL serves today.
func TestFindHookReadsTheConsentedManifest(t *testing.T) {
diff --git a/server/internal/service/plugin_schedule.go b/server/internal/service/plugin_schedule.go
new file mode 100644
index 000000000..6017a858c
--- /dev/null
+++ b/server/internal/service/plugin_schedule.go
@@ -0,0 +1,131 @@
+package service
+
+import (
+ "context"
+ "fmt"
+ "time"
+
+ "github.com/jackc/pgx/v5/pgtype"
+ db "github.com/multica-ai/multica/server/pkg/db/generated"
+ "github.com/multica-ai/multica/server/pkg/plugincontract"
+)
+
+// reconcilePluginHookSchedules makes the durable schedule projection match the
+// consented manifest inside the caller's installation/upgrade transaction.
+// Unchanged cron/timezone pairs keep their generation: a code-only plugin
+// upgrade must not reset the schedule clock or orphan a retryable execution.
+func (s *PluginService) reconcilePluginHookSchedules(
+ ctx context.Context,
+ queries *db.Queries,
+ installation db.PluginInstallation,
+ manifest plugincontract.Manifest,
+) error {
+ existingRows, err := queries.ListPluginHookSchedulesByInstallation(ctx, installation.ID)
+ if err != nil {
+ return &PluginError{Kind: PluginErrorUnavailable, Message: "list plugin hook schedules", Err: err}
+ }
+ existing := make(map[string]db.PluginHookSchedule, len(existingRows))
+ for _, row := range existingRows {
+ existing[row.HookKey] = row
+ }
+
+ now := time.Now().UTC()
+ for _, hook := range manifest.Contributes.Hooks {
+ if hook.Schedule == nil || !HookAllowsTrigger(hook, plugincontract.TriggerSchedule) {
+ continue
+ }
+ nextRun, err := pluginScheduleNextRun(*hook.Schedule, now)
+ if err != nil {
+ // Published manifests were validated already. Surfacing drift here is
+ // safer than installing a schedule the runtime cannot evaluate.
+ return &PluginError{Kind: PluginErrorInvalid, Message: fmt.Sprintf("schedule for hook %q is invalid", hook.Key), Err: err}
+ }
+ if !installation.Enabled {
+ // Disabled installations have no meaningful "next" run. Re-enable
+ // rotates the generation and computes this projection from that time.
+ nextRun = pgtype.Timestamptz{}
+ }
+
+ row, found := existing[hook.Key]
+ if !found {
+ if _, err := queries.CreatePluginHookSchedule(ctx, db.CreatePluginHookScheduleParams{
+ InstallationID: installation.ID,
+ WorkspaceID: installation.WorkspaceID,
+ HookKey: hook.Key,
+ CronExpression: hook.Schedule.Cron,
+ Timezone: hook.Schedule.Timezone,
+ Enabled: installation.Enabled,
+ NextRunAt: nextRun,
+ }); err != nil {
+ return &PluginError{Kind: PluginErrorUnavailable, Message: "create plugin hook schedule", Err: err}
+ }
+ continue
+ }
+ delete(existing, hook.Key)
+ if row.CronExpression == hook.Schedule.Cron && row.Timezone == hook.Schedule.Timezone {
+ continue
+ }
+ if _, err := queries.UpdatePluginHookScheduleDefinition(ctx, db.UpdatePluginHookScheduleDefinitionParams{
+ ID: row.ID,
+ CronExpression: hook.Schedule.Cron,
+ Timezone: hook.Schedule.Timezone,
+ Enabled: installation.Enabled,
+ NextRunAt: nextRun,
+ }); err != nil {
+ return &PluginError{Kind: PluginErrorUnavailable, Message: "update plugin hook schedule", Err: err}
+ }
+ }
+
+ for _, orphan := range existing {
+ if err := queries.DeletePluginHookSchedule(ctx, orphan.ID); err != nil {
+ return &PluginError{Kind: PluginErrorUnavailable, Message: "delete removed plugin hook schedule", Err: err}
+ }
+ }
+ return nil
+}
+
+func pluginScheduleNextRun(schedule plugincontract.HookSchedule, after time.Time) (pgtype.Timestamptz, error) {
+ next, err := NextOccurrenceAfterUTC(schedule.Cron, schedule.Timezone, after)
+ if err != nil {
+ return pgtype.Timestamptz{}, err
+ }
+ if next.IsZero() {
+ return pgtype.Timestamptz{}, nil
+ }
+ return pgtype.Timestamptz{Time: next.UTC(), Valid: true}, nil
+}
+
+func (s *PluginService) setPluginHookSchedulesEnabled(
+ ctx context.Context,
+ queries *db.Queries,
+ installationID pgtype.UUID,
+ enabled bool,
+) error {
+ if !enabled {
+ if err := queries.DisablePluginHookSchedules(ctx, installationID); err != nil {
+ return &PluginError{Kind: PluginErrorUnavailable, Message: "disable plugin hook schedules", Err: err}
+ }
+ return nil
+ }
+
+ rows, err := queries.ListPluginHookSchedulesByInstallation(ctx, installationID)
+ if err != nil {
+ return &PluginError{Kind: PluginErrorUnavailable, Message: "list plugin hook schedules", Err: err}
+ }
+ now := time.Now().UTC()
+ for _, row := range rows {
+ nextRun, err := pluginScheduleNextRun(plugincontract.HookSchedule{
+ Cron: row.CronExpression, Timezone: row.Timezone,
+ }, now)
+ if err != nil {
+ return &PluginError{Kind: PluginErrorInvalid, Message: fmt.Sprintf("schedule for hook %q is invalid", row.HookKey), Err: err}
+ }
+ if _, err := queries.ReactivatePluginHookSchedule(ctx, db.ReactivatePluginHookScheduleParams{
+ ID: row.ID,
+ NextRunAt: nextRun,
+ }); err != nil {
+ return &PluginError{Kind: PluginErrorUnavailable, Message: "reactivate plugin hook schedule", Err: err}
+ }
+ }
+ return nil
+}
diff --git a/server/migrations/399_plugin_hook_schedule.down.sql b/server/migrations/399_plugin_hook_schedule.down.sql
new file mode 100644
index 000000000..dc8c8fae5
--- /dev/null
+++ b/server/migrations/399_plugin_hook_schedule.down.sql
@@ -0,0 +1,9 @@
+DELETE FROM plugin_invocation WHERE trigger = 'schedule';
+
+ALTER TABLE plugin_invocation DROP CONSTRAINT IF EXISTS plugin_invocation_trigger_check;
+ALTER TABLE plugin_invocation ADD CONSTRAINT plugin_invocation_trigger_check
+ CHECK (trigger IN ('ui', 'manual', 'event', 'agent')) NOT VALID;
+ALTER TABLE plugin_invocation DROP COLUMN IF EXISTS planned_at;
+ALTER TABLE plugin_invocation DROP COLUMN IF EXISTS delivery_id;
+
+DROP TABLE IF EXISTS plugin_hook_schedule;
diff --git a/server/migrations/399_plugin_hook_schedule.up.sql b/server/migrations/399_plugin_hook_schedule.up.sql
new file mode 100644
index 000000000..2323e33eb
--- /dev/null
+++ b/server/migrations/399_plugin_hook_schedule.up.sql
@@ -0,0 +1,35 @@
+-- Durable schedule declarations for Plugin HTTP hooks. The consented manifest
+-- remains the public source of truth; this table is the execution projection
+-- that gives each installed hook an activation epoch and a stable scheduler
+-- scope without reparsing every installation on every tick.
+--
+-- Relationships are application-owned by repository policy: install, upgrade,
+-- enable/disable, uninstall and workspace teardown reconcile these rows in the
+-- same transaction as plugin_installation. No foreign key or cascade is used.
+CREATE TABLE plugin_hook_schedule (
+ id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+ installation_id UUID NOT NULL,
+ workspace_id UUID NOT NULL,
+ hook_key TEXT NOT NULL CHECK (char_length(hook_key) BETWEEN 1 AND 128),
+ cron_expression TEXT NOT NULL CHECK (char_length(cron_expression) BETWEEN 1 AND 255),
+ timezone TEXT NOT NULL CHECK (char_length(timezone) BETWEEN 1 AND 255),
+ generation UUID NOT NULL DEFAULT gen_random_uuid(),
+ activated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+ -- Display-only projection. Dispatch correctness is recovered from cron +
+ -- activated_at + sys_cron_executions even when this value is stale/NULL.
+ next_run_at TIMESTAMPTZ,
+ enabled BOOLEAN NOT NULL DEFAULT TRUE,
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+ updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+-- Schedule attempts use the same operational invocation table as every other
+-- Hook trigger. delivery_id is stable across retries; planned_at is the cron
+-- occurrence, distinct from created_at (the actual attempt time).
+ALTER TABLE plugin_invocation
+ ADD COLUMN delivery_id TEXT CHECK (delivery_id IS NULL OR char_length(delivery_id) BETWEEN 1 AND 128),
+ ADD COLUMN planned_at TIMESTAMPTZ;
+
+ALTER TABLE plugin_invocation DROP CONSTRAINT IF EXISTS plugin_invocation_trigger_check;
+ALTER TABLE plugin_invocation ADD CONSTRAINT plugin_invocation_trigger_check
+ CHECK (trigger IN ('ui', 'manual', 'event', 'agent', 'schedule')) NOT VALID;
diff --git a/server/migrations/400_plugin_hook_schedule_installation_key_index.down.sql b/server/migrations/400_plugin_hook_schedule_installation_key_index.down.sql
new file mode 100644
index 000000000..b1d1f5ae9
--- /dev/null
+++ b/server/migrations/400_plugin_hook_schedule_installation_key_index.down.sql
@@ -0,0 +1 @@
+DROP INDEX CONCURRENTLY IF EXISTS idx_plugin_hook_schedule_installation_key;
diff --git a/server/migrations/400_plugin_hook_schedule_installation_key_index.up.sql b/server/migrations/400_plugin_hook_schedule_installation_key_index.up.sql
new file mode 100644
index 000000000..6445015f2
--- /dev/null
+++ b/server/migrations/400_plugin_hook_schedule_installation_key_index.up.sql
@@ -0,0 +1,2 @@
+CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS idx_plugin_hook_schedule_installation_key
+ ON plugin_hook_schedule (installation_id, hook_key);
diff --git a/server/migrations/401_plugin_hook_schedule_enabled_index.down.sql b/server/migrations/401_plugin_hook_schedule_enabled_index.down.sql
new file mode 100644
index 000000000..d41dad14a
--- /dev/null
+++ b/server/migrations/401_plugin_hook_schedule_enabled_index.down.sql
@@ -0,0 +1 @@
+DROP INDEX CONCURRENTLY IF EXISTS idx_plugin_hook_schedule_enabled;
diff --git a/server/migrations/401_plugin_hook_schedule_enabled_index.up.sql b/server/migrations/401_plugin_hook_schedule_enabled_index.up.sql
new file mode 100644
index 000000000..8110b5024
--- /dev/null
+++ b/server/migrations/401_plugin_hook_schedule_enabled_index.up.sql
@@ -0,0 +1,3 @@
+CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_plugin_hook_schedule_enabled
+ ON plugin_hook_schedule (id)
+ WHERE enabled;
diff --git a/server/migrations/402_plugin_invocation_schedule_trigger_validate.down.sql b/server/migrations/402_plugin_invocation_schedule_trigger_validate.down.sql
new file mode 100644
index 000000000..6997bc9d6
--- /dev/null
+++ b/server/migrations/402_plugin_invocation_schedule_trigger_validate.down.sql
@@ -0,0 +1,3 @@
+-- Validation changes no data or contract shape; migration 399 restores the old
+-- constraint during rollback.
+SELECT 1;
diff --git a/server/migrations/402_plugin_invocation_schedule_trigger_validate.up.sql b/server/migrations/402_plugin_invocation_schedule_trigger_validate.up.sql
new file mode 100644
index 000000000..a9d612f07
--- /dev/null
+++ b/server/migrations/402_plugin_invocation_schedule_trigger_validate.up.sql
@@ -0,0 +1 @@
+ALTER TABLE plugin_invocation VALIDATE CONSTRAINT plugin_invocation_trigger_check;
diff --git a/server/pkg/db/generated/models.go b/server/pkg/db/generated/models.go
index be5748aab..dbde55fdd 100644
--- a/server/pkg/db/generated/models.go
+++ b/server/pkg/db/generated/models.go
@@ -981,6 +981,21 @@ type PinnedItem struct {
CreatedAt pgtype.Timestamptz `json:"created_at"`
}
+type PluginHookSchedule struct {
+ ID pgtype.UUID `json:"id"`
+ InstallationID pgtype.UUID `json:"installation_id"`
+ WorkspaceID pgtype.UUID `json:"workspace_id"`
+ HookKey string `json:"hook_key"`
+ CronExpression string `json:"cron_expression"`
+ Timezone string `json:"timezone"`
+ Generation pgtype.UUID `json:"generation"`
+ ActivatedAt pgtype.Timestamptz `json:"activated_at"`
+ NextRunAt pgtype.Timestamptz `json:"next_run_at"`
+ Enabled bool `json:"enabled"`
+ CreatedAt pgtype.Timestamptz `json:"created_at"`
+ UpdatedAt pgtype.Timestamptz `json:"updated_at"`
+}
+
type PluginInstallation struct {
ID pgtype.UUID `json:"id"`
WorkspaceID pgtype.UUID `json:"workspace_id"`
@@ -1011,6 +1026,8 @@ type PluginInvocation struct {
LatencyMs int32 `json:"latency_ms"`
Error pgtype.Text `json:"error"`
CreatedAt pgtype.Timestamptz `json:"created_at"`
+ DeliveryID pgtype.Text `json:"delivery_id"`
+ PlannedAt pgtype.Timestamptz `json:"planned_at"`
}
type PluginPackage struct {
diff --git a/server/pkg/db/generated/plugin.sql.go b/server/pkg/db/generated/plugin.sql.go
index e4ed4ad36..257671039 100644
--- a/server/pkg/db/generated/plugin.sql.go
+++ b/server/pkg/db/generated/plugin.sql.go
@@ -67,6 +67,52 @@ func (q *Queries) CountRecentPluginInvocations(ctx context.Context, arg CountRec
return count, err
}
+const createPluginHookSchedule = `-- name: CreatePluginHookSchedule :one
+INSERT INTO plugin_hook_schedule (
+ installation_id, workspace_id, hook_key, cron_expression, timezone,
+ next_run_at, enabled
+) VALUES ($1, $2, $3, $4, $5, $7, $6)
+RETURNING id, installation_id, workspace_id, hook_key, cron_expression, timezone, generation, activated_at, next_run_at, enabled, created_at, updated_at
+`
+
+type CreatePluginHookScheduleParams struct {
+ InstallationID pgtype.UUID `json:"installation_id"`
+ WorkspaceID pgtype.UUID `json:"workspace_id"`
+ HookKey string `json:"hook_key"`
+ CronExpression string `json:"cron_expression"`
+ Timezone string `json:"timezone"`
+ Enabled bool `json:"enabled"`
+ NextRunAt pgtype.Timestamptz `json:"next_run_at"`
+}
+
+func (q *Queries) CreatePluginHookSchedule(ctx context.Context, arg CreatePluginHookScheduleParams) (PluginHookSchedule, error) {
+ row := q.db.QueryRow(ctx, createPluginHookSchedule,
+ arg.InstallationID,
+ arg.WorkspaceID,
+ arg.HookKey,
+ arg.CronExpression,
+ arg.Timezone,
+ arg.Enabled,
+ arg.NextRunAt,
+ )
+ var i PluginHookSchedule
+ err := row.Scan(
+ &i.ID,
+ &i.InstallationID,
+ &i.WorkspaceID,
+ &i.HookKey,
+ &i.CronExpression,
+ &i.Timezone,
+ &i.Generation,
+ &i.ActivatedAt,
+ &i.NextRunAt,
+ &i.Enabled,
+ &i.CreatedAt,
+ &i.UpdatedAt,
+ )
+ return i, err
+}
+
const createPluginInstallation = `-- name: CreatePluginInstallation :one
INSERT INTO plugin_installation (
workspace_id, plugin_key, package_version_id, version, manifest, granted_scopes, installed_by
@@ -117,21 +163,29 @@ func (q *Queries) CreatePluginInstallation(ctx context.Context, arg CreatePlugin
const createPluginInvocation = `-- name: CreatePluginInvocation :one
INSERT INTO plugin_invocation (
- installation_id, workspace_id, hook_key, trigger, status, event_type, attempt, latency_ms, error
-) VALUES ($1, $2, $3, $4, $5, $8, $6, $7, $9)
-RETURNING id, installation_id, workspace_id, hook_key, trigger, status, event_type, attempt, latency_ms, error, created_at
+ id, installation_id, workspace_id, hook_key, trigger, status, event_type,
+ delivery_id, planned_at, attempt, latency_ms, error
+) VALUES (
+ COALESCE($8::uuid, gen_random_uuid()),
+ $1, $2, $3, $4, $5, $9, $10,
+ $11, $6, $7, $12
+)
+RETURNING id, installation_id, workspace_id, hook_key, trigger, status, event_type, attempt, latency_ms, error, created_at, delivery_id, planned_at
`
type CreatePluginInvocationParams struct {
- InstallationID pgtype.UUID `json:"installation_id"`
- WorkspaceID pgtype.UUID `json:"workspace_id"`
- HookKey string `json:"hook_key"`
- Trigger string `json:"trigger"`
- Status string `json:"status"`
- Attempt int32 `json:"attempt"`
- LatencyMs int32 `json:"latency_ms"`
- EventType pgtype.Text `json:"event_type"`
- Error pgtype.Text `json:"error"`
+ InstallationID pgtype.UUID `json:"installation_id"`
+ WorkspaceID pgtype.UUID `json:"workspace_id"`
+ HookKey string `json:"hook_key"`
+ Trigger string `json:"trigger"`
+ Status string `json:"status"`
+ Attempt int32 `json:"attempt"`
+ LatencyMs int32 `json:"latency_ms"`
+ ID pgtype.UUID `json:"id"`
+ EventType pgtype.Text `json:"event_type"`
+ DeliveryID pgtype.Text `json:"delivery_id"`
+ PlannedAt pgtype.Timestamptz `json:"planned_at"`
+ Error pgtype.Text `json:"error"`
}
func (q *Queries) CreatePluginInvocation(ctx context.Context, arg CreatePluginInvocationParams) (PluginInvocation, error) {
@@ -143,7 +197,10 @@ func (q *Queries) CreatePluginInvocation(ctx context.Context, arg CreatePluginIn
arg.Status,
arg.Attempt,
arg.LatencyMs,
+ arg.ID,
arg.EventType,
+ arg.DeliveryID,
+ arg.PlannedAt,
arg.Error,
)
var i PluginInvocation
@@ -159,6 +216,8 @@ func (q *Queries) CreatePluginInvocation(ctx context.Context, arg CreatePluginIn
&i.LatencyMs,
&i.Error,
&i.CreatedAt,
+ &i.DeliveryID,
+ &i.PlannedAt,
)
return i, err
}
@@ -278,6 +337,24 @@ func (q *Queries) DeleteExpiredPluginInvocations(ctx context.Context, createdAt
return result.RowsAffected(), nil
}
+const deletePluginHookSchedule = `-- name: DeletePluginHookSchedule :exec
+DELETE FROM plugin_hook_schedule WHERE id = $1
+`
+
+func (q *Queries) DeletePluginHookSchedule(ctx context.Context, id pgtype.UUID) error {
+ _, err := q.db.Exec(ctx, deletePluginHookSchedule, id)
+ return err
+}
+
+const deletePluginHookSchedulesByInstallation = `-- name: DeletePluginHookSchedulesByInstallation :exec
+DELETE FROM plugin_hook_schedule WHERE installation_id = $1
+`
+
+func (q *Queries) DeletePluginHookSchedulesByInstallation(ctx context.Context, installationID pgtype.UUID) error {
+ _, err := q.db.Exec(ctx, deletePluginHookSchedulesByInstallation, installationID)
+ return err
+}
+
const deletePluginInstallation = `-- name: DeletePluginInstallation :exec
DELETE FROM plugin_installation WHERE id = $1
`
@@ -410,6 +487,41 @@ func (q *Queries) DeletePluginStorageValue(ctx context.Context, arg DeletePlugin
return result.RowsAffected(), nil
}
+const disablePluginHookSchedules = `-- name: DisablePluginHookSchedules :exec
+UPDATE plugin_hook_schedule
+SET enabled = FALSE, next_run_at = NULL, updated_at = now()
+WHERE installation_id = $1
+`
+
+func (q *Queries) DisablePluginHookSchedules(ctx context.Context, installationID pgtype.UUID) error {
+ _, err := q.db.Exec(ctx, disablePluginHookSchedules, installationID)
+ return err
+}
+
+const getPluginHookSchedule = `-- name: GetPluginHookSchedule :one
+SELECT id, installation_id, workspace_id, hook_key, cron_expression, timezone, generation, activated_at, next_run_at, enabled, created_at, updated_at FROM plugin_hook_schedule WHERE id = $1
+`
+
+func (q *Queries) GetPluginHookSchedule(ctx context.Context, id pgtype.UUID) (PluginHookSchedule, error) {
+ row := q.db.QueryRow(ctx, getPluginHookSchedule, id)
+ var i PluginHookSchedule
+ err := row.Scan(
+ &i.ID,
+ &i.InstallationID,
+ &i.WorkspaceID,
+ &i.HookKey,
+ &i.CronExpression,
+ &i.Timezone,
+ &i.Generation,
+ &i.ActivatedAt,
+ &i.NextRunAt,
+ &i.Enabled,
+ &i.CreatedAt,
+ &i.UpdatedAt,
+ )
+ return i, err
+}
+
const getPluginInstallation = `-- name: GetPluginInstallation :one
SELECT id, workspace_id, plugin_key, version, manifest, granted_scopes, config, enabled, installed_by, created_at, updated_at, token_hash, token_rotated_at, mcp_approvals, package_version_id FROM plugin_installation WHERE id = $1
`
@@ -725,8 +837,88 @@ func (q *Queries) GetWorkspacePluginPackageVersion(ctx context.Context, arg GetW
return i, err
}
+const listEnabledPluginHookSchedules = `-- name: ListEnabledPluginHookSchedules :many
+SELECT id, installation_id, workspace_id, hook_key, cron_expression, timezone, generation, activated_at, next_run_at, enabled, created_at, updated_at FROM plugin_hook_schedule
+WHERE enabled
+ORDER BY id ASC
+`
+
+// Do not filter by next_run_at: it is display-only, while retries and recovery
+// derive eligibility from cron + activated_at + sys_cron_executions.
+func (q *Queries) ListEnabledPluginHookSchedules(ctx context.Context) ([]PluginHookSchedule, error) {
+ rows, err := q.db.Query(ctx, listEnabledPluginHookSchedules)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+ items := []PluginHookSchedule{}
+ for rows.Next() {
+ var i PluginHookSchedule
+ if err := rows.Scan(
+ &i.ID,
+ &i.InstallationID,
+ &i.WorkspaceID,
+ &i.HookKey,
+ &i.CronExpression,
+ &i.Timezone,
+ &i.Generation,
+ &i.ActivatedAt,
+ &i.NextRunAt,
+ &i.Enabled,
+ &i.CreatedAt,
+ &i.UpdatedAt,
+ ); err != nil {
+ return nil, err
+ }
+ items = append(items, i)
+ }
+ if err := rows.Err(); err != nil {
+ return nil, err
+ }
+ return items, nil
+}
+
+const listPluginHookSchedulesByInstallation = `-- name: ListPluginHookSchedulesByInstallation :many
+SELECT id, installation_id, workspace_id, hook_key, cron_expression, timezone, generation, activated_at, next_run_at, enabled, created_at, updated_at FROM plugin_hook_schedule
+WHERE installation_id = $1
+ORDER BY hook_key ASC
+`
+
+func (q *Queries) ListPluginHookSchedulesByInstallation(ctx context.Context, installationID pgtype.UUID) ([]PluginHookSchedule, error) {
+ rows, err := q.db.Query(ctx, listPluginHookSchedulesByInstallation, installationID)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+ items := []PluginHookSchedule{}
+ for rows.Next() {
+ var i PluginHookSchedule
+ if err := rows.Scan(
+ &i.ID,
+ &i.InstallationID,
+ &i.WorkspaceID,
+ &i.HookKey,
+ &i.CronExpression,
+ &i.Timezone,
+ &i.Generation,
+ &i.ActivatedAt,
+ &i.NextRunAt,
+ &i.Enabled,
+ &i.CreatedAt,
+ &i.UpdatedAt,
+ ); err != nil {
+ return nil, err
+ }
+ items = append(items, i)
+ }
+ if err := rows.Err(); err != nil {
+ return nil, err
+ }
+ return items, nil
+}
+
const listPluginInvocations = `-- name: ListPluginInvocations :many
-SELECT id, installation_id, workspace_id, hook_key, trigger, status, event_type, attempt, latency_ms, error, created_at FROM plugin_invocation
+SELECT id, installation_id, workspace_id, hook_key, trigger, status, event_type, attempt, latency_ms, error, created_at, delivery_id, planned_at FROM plugin_invocation
WHERE installation_id = $1
ORDER BY created_at DESC
LIMIT $2
@@ -758,6 +950,8 @@ func (q *Queries) ListPluginInvocations(ctx context.Context, arg ListPluginInvoc
&i.LatencyMs,
&i.Error,
&i.CreatedAt,
+ &i.DeliveryID,
+ &i.PlannedAt,
); err != nil {
return nil, err
}
@@ -1041,6 +1235,45 @@ func (q *Queries) LockPluginPackageKey(ctx context.Context, dollar_1 string) err
return err
}
+const reactivatePluginHookSchedule = `-- name: ReactivatePluginHookSchedule :one
+UPDATE plugin_hook_schedule
+SET enabled = TRUE,
+ generation = gen_random_uuid(),
+ activated_at = now(),
+ next_run_at = $2,
+ updated_at = now()
+WHERE id = $1
+RETURNING id, installation_id, workspace_id, hook_key, cron_expression, timezone, generation, activated_at, next_run_at, enabled, created_at, updated_at
+`
+
+type ReactivatePluginHookScheduleParams struct {
+ ID pgtype.UUID `json:"id"`
+ NextRunAt pgtype.Timestamptz `json:"next_run_at"`
+}
+
+// Re-enable starts a new epoch so occurrences while the installation was off
+// are never caught up. An already-sent request from the old generation may
+// finish, but no unstarted old plan survives the generation check.
+func (q *Queries) ReactivatePluginHookSchedule(ctx context.Context, arg ReactivatePluginHookScheduleParams) (PluginHookSchedule, error) {
+ row := q.db.QueryRow(ctx, reactivatePluginHookSchedule, arg.ID, arg.NextRunAt)
+ var i PluginHookSchedule
+ err := row.Scan(
+ &i.ID,
+ &i.InstallationID,
+ &i.WorkspaceID,
+ &i.HookKey,
+ &i.CronExpression,
+ &i.Timezone,
+ &i.Generation,
+ &i.ActivatedAt,
+ &i.NextRunAt,
+ &i.Enabled,
+ &i.CreatedAt,
+ &i.UpdatedAt,
+ )
+ return i, err
+}
+
const setPluginInstallationEnabled = `-- name: SetPluginInstallationEnabled :one
UPDATE plugin_installation
SET enabled = $2,
@@ -1128,6 +1361,75 @@ func (q *Queries) SetPluginMCPApprovals(ctx context.Context, arg SetPluginMCPApp
return i, err
}
+const updatePluginHookScheduleDefinition = `-- name: UpdatePluginHookScheduleDefinition :one
+UPDATE plugin_hook_schedule
+SET cron_expression = $2,
+ timezone = $3,
+ generation = gen_random_uuid(),
+ activated_at = now(),
+ next_run_at = $5,
+ enabled = $4,
+ updated_at = now()
+WHERE id = $1
+RETURNING id, installation_id, workspace_id, hook_key, cron_expression, timezone, generation, activated_at, next_run_at, enabled, created_at, updated_at
+`
+
+type UpdatePluginHookScheduleDefinitionParams struct {
+ ID pgtype.UUID `json:"id"`
+ CronExpression string `json:"cron_expression"`
+ Timezone string `json:"timezone"`
+ Enabled bool `json:"enabled"`
+ NextRunAt pgtype.Timestamptz `json:"next_run_at"`
+}
+
+// A cron/timezone change creates a new scheduler generation. The old
+// sys_cron_executions rows remain immutable history under the prior scope id.
+func (q *Queries) UpdatePluginHookScheduleDefinition(ctx context.Context, arg UpdatePluginHookScheduleDefinitionParams) (PluginHookSchedule, error) {
+ row := q.db.QueryRow(ctx, updatePluginHookScheduleDefinition,
+ arg.ID,
+ arg.CronExpression,
+ arg.Timezone,
+ arg.Enabled,
+ arg.NextRunAt,
+ )
+ var i PluginHookSchedule
+ err := row.Scan(
+ &i.ID,
+ &i.InstallationID,
+ &i.WorkspaceID,
+ &i.HookKey,
+ &i.CronExpression,
+ &i.Timezone,
+ &i.Generation,
+ &i.ActivatedAt,
+ &i.NextRunAt,
+ &i.Enabled,
+ &i.CreatedAt,
+ &i.UpdatedAt,
+ )
+ return i, err
+}
+
+const updatePluginHookScheduleNextRun = `-- name: UpdatePluginHookScheduleNextRun :execrows
+UPDATE plugin_hook_schedule
+SET next_run_at = $3, updated_at = now()
+WHERE id = $1 AND generation = $2 AND enabled
+`
+
+type UpdatePluginHookScheduleNextRunParams struct {
+ ID pgtype.UUID `json:"id"`
+ Generation pgtype.UUID `json:"generation"`
+ NextRunAt pgtype.Timestamptz `json:"next_run_at"`
+}
+
+func (q *Queries) UpdatePluginHookScheduleNextRun(ctx context.Context, arg UpdatePluginHookScheduleNextRunParams) (int64, error) {
+ result, err := q.db.Exec(ctx, updatePluginHookScheduleNextRun, arg.ID, arg.Generation, arg.NextRunAt)
+ if err != nil {
+ return 0, err
+ }
+ return result.RowsAffected(), nil
+}
+
const updatePluginInstallationConfig = `-- name: UpdatePluginInstallationConfig :one
UPDATE plugin_installation
SET config = $2,
diff --git a/server/pkg/db/generated/workspace_delete.sql.go b/server/pkg/db/generated/workspace_delete.sql.go
index f75106277..d39617c78 100644
--- a/server/pkg/db/generated/workspace_delete.sql.go
+++ b/server/pkg/db/generated/workspace_delete.sql.go
@@ -503,6 +503,10 @@ deleted_secrets AS (
DELETE FROM plugin_secret
WHERE installation_id IN (SELECT id FROM installations)
),
+deleted_hook_schedules AS (
+ DELETE FROM plugin_hook_schedule
+ WHERE installation_id IN (SELECT id FROM installations)
+),
deleted_invocations AS (
DELETE FROM plugin_invocation
WHERE workspace_id = $1
diff --git a/server/pkg/db/queries/plugin.sql b/server/pkg/db/queries/plugin.sql
index e4653f6bc..19e38595b 100644
--- a/server/pkg/db/queries/plugin.sql
+++ b/server/pkg/db/queries/plugin.sql
@@ -143,6 +143,71 @@ ORDER BY created_at ASC;
-- name: DeletePluginInstallation :exec
DELETE FROM plugin_installation WHERE id = $1;
+-- name: CreatePluginHookSchedule :one
+INSERT INTO plugin_hook_schedule (
+ installation_id, workspace_id, hook_key, cron_expression, timezone,
+ next_run_at, enabled
+) VALUES ($1, $2, $3, $4, $5, sqlc.narg(next_run_at), $6)
+RETURNING *;
+
+-- name: GetPluginHookSchedule :one
+SELECT * FROM plugin_hook_schedule WHERE id = $1;
+
+-- name: ListPluginHookSchedulesByInstallation :many
+SELECT * FROM plugin_hook_schedule
+WHERE installation_id = $1
+ORDER BY hook_key ASC;
+
+-- name: ListEnabledPluginHookSchedules :many
+-- Do not filter by next_run_at: it is display-only, while retries and recovery
+-- derive eligibility from cron + activated_at + sys_cron_executions.
+SELECT * FROM plugin_hook_schedule
+WHERE enabled
+ORDER BY id ASC;
+
+-- name: UpdatePluginHookScheduleDefinition :one
+-- A cron/timezone change creates a new scheduler generation. The old
+-- sys_cron_executions rows remain immutable history under the prior scope id.
+UPDATE plugin_hook_schedule
+SET cron_expression = $2,
+ timezone = $3,
+ generation = gen_random_uuid(),
+ activated_at = now(),
+ next_run_at = sqlc.narg(next_run_at),
+ enabled = $4,
+ updated_at = now()
+WHERE id = $1
+RETURNING *;
+
+-- name: DisablePluginHookSchedules :exec
+UPDATE plugin_hook_schedule
+SET enabled = FALSE, next_run_at = NULL, updated_at = now()
+WHERE installation_id = $1;
+
+-- name: ReactivatePluginHookSchedule :one
+-- Re-enable starts a new epoch so occurrences while the installation was off
+-- are never caught up. An already-sent request from the old generation may
+-- finish, but no unstarted old plan survives the generation check.
+UPDATE plugin_hook_schedule
+SET enabled = TRUE,
+ generation = gen_random_uuid(),
+ activated_at = now(),
+ next_run_at = sqlc.narg(next_run_at),
+ updated_at = now()
+WHERE id = $1
+RETURNING *;
+
+-- name: UpdatePluginHookScheduleNextRun :execrows
+UPDATE plugin_hook_schedule
+SET next_run_at = sqlc.narg(next_run_at), updated_at = now()
+WHERE id = $1 AND generation = $2 AND enabled;
+
+-- name: DeletePluginHookSchedule :exec
+DELETE FROM plugin_hook_schedule WHERE id = $1;
+
+-- name: DeletePluginHookSchedulesByInstallation :exec
+DELETE FROM plugin_hook_schedule WHERE installation_id = $1;
+
-- name: UpsertPluginStorageValue :one
INSERT INTO plugin_storage (installation_id, scope_type, scope_id, key, value)
VALUES ($1, $2, $3, $4, $5)
@@ -211,8 +276,13 @@ SELECT * FROM plugin_installation WHERE token_hash = $1;
-- name: CreatePluginInvocation :one
INSERT INTO plugin_invocation (
- installation_id, workspace_id, hook_key, trigger, status, event_type, attempt, latency_ms, error
-) VALUES ($1, $2, $3, $4, $5, sqlc.narg(event_type), $6, $7, sqlc.narg(error))
+ id, installation_id, workspace_id, hook_key, trigger, status, event_type,
+ delivery_id, planned_at, attempt, latency_ms, error
+) VALUES (
+ COALESCE(sqlc.narg('id')::uuid, gen_random_uuid()),
+ $1, $2, $3, $4, $5, sqlc.narg(event_type), sqlc.narg(delivery_id),
+ sqlc.narg(planned_at), $6, $7, sqlc.narg(error)
+)
RETURNING *;
-- name: ListPluginInvocations :many
diff --git a/server/pkg/db/queries/workspace_delete.sql b/server/pkg/db/queries/workspace_delete.sql
index 42006eacb..f26a837c0 100644
--- a/server/pkg/db/queries/workspace_delete.sql
+++ b/server/pkg/db/queries/workspace_delete.sql
@@ -602,6 +602,10 @@ deleted_secrets AS (
DELETE FROM plugin_secret
WHERE installation_id IN (SELECT id FROM installations)
),
+deleted_hook_schedules AS (
+ DELETE FROM plugin_hook_schedule
+ WHERE installation_id IN (SELECT id FROM installations)
+),
-- Hook call records are workspace-scoped in their own right, so this deletes by
-- workspace rather than through the installation ids: a row whose installation
-- was already uninstalled would otherwise survive the workspace it described.
diff --git a/server/pkg/plugincontract/capabilities.go b/server/pkg/plugincontract/capabilities.go
index f2cea286e..76c4275f1 100644
--- a/server/pkg/plugincontract/capabilities.go
+++ b/server/pkg/plugincontract/capabilities.go
@@ -39,10 +39,11 @@ func HostCapabilities() Capabilities {
// offered to an agent as an MCP tool and the agent decides. The
// daemon-side server that renders it is daemon/plugin_hook_mcp.go.
HookTriggers: map[string]bool{
- TriggerUI: true,
- TriggerManual: true,
- TriggerEvent: true,
- TriggerAgent: true,
+ TriggerUI: true,
+ TriggerManual: true,
+ TriggerEvent: true,
+ TriggerAgent: true,
+ TriggerSchedule: true,
},
// http calls one declared endpoint; mcp adopts an external MCP server's
// tools, which is why it ships with an approval step that pins them by
diff --git a/server/pkg/plugincontract/manifest.go b/server/pkg/plugincontract/manifest.go
index c78007d6f..c1b6ce2c5 100644
--- a/server/pkg/plugincontract/manifest.go
+++ b/server/pkg/plugincontract/manifest.go
@@ -20,6 +20,9 @@ import (
"net/url"
"regexp"
"strings"
+ "time"
+
+ "github.com/robfig/cron/v3"
)
const (
@@ -44,14 +47,24 @@ const (
)
// Hook triggers. Declaring a trigger says who may invoke the hook, never what
-// the hook does. Only `event` is asynchronous, and it never blocks the host.
+// the hook does. `event` and `schedule` are asynchronous, and neither blocks a
+// user-facing host request.
const (
- TriggerUI = "ui"
- TriggerManual = "manual"
- TriggerAgent = "agent"
- TriggerEvent = "event"
+ TriggerUI = "ui"
+ TriggerManual = "manual"
+ TriggerAgent = "agent"
+ TriggerEvent = "event"
+ TriggerSchedule = "schedule"
)
+// Scheduled hooks are a polling primitive, not a high-frequency timer. The
+// floor keeps one installation from turning its manifest into an unbounded
+// source of outbound traffic and leaves enough room for the HTTP timeout plus
+// durable retries before the next plan becomes due.
+const MinimumScheduleInterval = 5 * time.Minute
+
+var scheduleCronParser = cron.NewParser(cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow)
+
// Hook transports.
const (
TransportHTTP = "http"
@@ -203,10 +216,19 @@ type Hook struct {
InputSchema json.RawMessage `json:"input_schema,omitempty"`
Triggers []string `json:"triggers"`
Events []string `json:"events,omitempty"`
+ Schedule *HookSchedule `json:"schedule,omitempty"`
Transport HookTransport `json:"transport"`
TimeoutMs int `json:"timeout_ms,omitempty"`
}
+// HookSchedule declares the one automatic cadence attached to a hook. Multiple
+// cadences are represented by multiple hook keys, which keeps the durable
+// execution identity unambiguous.
+type HookSchedule struct {
+ Cron string `json:"cron"`
+ Timezone string `json:"timezone"`
+}
+
type HookTransport struct {
Type string `json:"type"`
URL string `json:"url"`
@@ -605,7 +627,7 @@ func (m Manifest) validateContributions() error {
triggerSeen := map[string]bool{}
for _, trigger := range hook.Triggers {
switch trigger {
- case TriggerUI, TriggerManual, TriggerAgent, TriggerEvent:
+ case TriggerUI, TriggerManual, TriggerAgent, TriggerEvent, TriggerSchedule:
default:
return fmt.Errorf("%s.triggers contains unsupported trigger %q", field, trigger)
}
@@ -640,6 +662,19 @@ func (m Manifest) validateContributions() error {
} else if len(hook.Events) > 0 {
return fmt.Errorf("%s.events requires the event trigger", field)
}
+ if triggerSeen[TriggerSchedule] {
+ if hook.Schedule == nil {
+ return fmt.Errorf("%s.schedule is required when the schedule trigger is declared", field)
+ }
+ if hook.Transport.Type != TransportHTTP {
+ return fmt.Errorf("%s.schedule only supports the http transport", field)
+ }
+ if err := validateHookSchedule(field+".schedule", *hook.Schedule); err != nil {
+ return err
+ }
+ } else if hook.Schedule != nil {
+ return fmt.Errorf("%s.schedule requires the schedule trigger", field)
+ }
if err := m.validateHookTransport(field, hook.Transport); err != nil {
return err
}
@@ -671,6 +706,50 @@ func (m Manifest) validateContributions() error {
return nil
}
+func validateHookSchedule(field string, schedule HookSchedule) error {
+ expression := strings.TrimSpace(schedule.Cron)
+ if expression == "" {
+ return fmt.Errorf("%s.cron must not be empty", field)
+ }
+ // Timezone belongs in its own field. Besides keeping the public shape
+ // singular, rejecting an inline prefix avoids robfig/cron's malformed-prefix
+ // panic path and prevents two timezone declarations from disagreeing.
+ if strings.HasPrefix(expression, "TZ=") || strings.HasPrefix(expression, "CRON_TZ=") {
+ return fmt.Errorf("%s.cron must not contain an inline timezone", field)
+ }
+ parsed, err := scheduleCronParser.Parse(expression)
+ if err != nil {
+ return fmt.Errorf("%s.cron must be a standard five-field cron expression: %w", field, err)
+ }
+ if strings.TrimSpace(schedule.Timezone) == "" {
+ return fmt.Errorf("%s.timezone must not be empty", field)
+ }
+ location, err := time.LoadLocation(schedule.Timezone)
+ if err != nil {
+ return fmt.Errorf("%s.timezone is invalid: %w", field, err)
+ }
+
+ // A 400-day window covers every month plus a DST cycle. High-frequency
+ // expressions fail after their first two occurrences; compliant five-minute
+ // expressions stay below 116k iterations, bounded work at publish/preview
+ // time rather than on every scheduler tick. A monthly/yearly expression with
+ // zero or one occurrence in the window is valid by definition.
+ start := time.Date(2024, time.January, 1, 0, 0, 0, 0, time.UTC).In(location)
+ end := start.AddDate(0, 0, 400)
+ previous := parsed.Next(start)
+ for !previous.IsZero() && !previous.After(end) {
+ next := parsed.Next(previous)
+ if next.IsZero() || next.After(end) {
+ break
+ }
+ if next.Sub(previous) < MinimumScheduleInterval {
+ return fmt.Errorf("%s.cron must not run more often than every five minutes", field)
+ }
+ previous = next
+ }
+ return nil
+}
+
// validateHookTransport enforces that a hook can only reach a host the manifest
// declared through a `net:` scope. Without this the consent screen would not
// describe where plugin data actually goes.
diff --git a/server/pkg/plugincontract/manifest_test.go b/server/pkg/plugincontract/manifest_test.go
index 295391199..71388c448 100644
--- a/server/pkg/plugincontract/manifest_test.go
+++ b/server/pkg/plugincontract/manifest_test.go
@@ -77,6 +77,61 @@ func TestParseManifestAcceptsReferenceDocument(t *testing.T) {
}
}
+func TestParseManifestAcceptsScheduledHook(t *testing.T) {
+ raw := mutate(t, func(doc map[string]any) {
+ contributes := doc["contributes"].(map[string]any)
+ hook := contributes["hooks"].([]any)[0].(map[string]any)
+ hook["triggers"] = []any{"schedule", "manual"}
+ hook["schedule"] = map[string]any{"cron": "*/5 * * * *", "timezone": "Asia/Shanghai"}
+ })
+ manifest, _, err := ParseManifest(raw)
+ if err != nil {
+ t.Fatalf("ParseManifest: %v", err)
+ }
+ hook := manifest.Contributes.Hooks[0]
+ if hook.Schedule == nil || hook.Schedule.Cron != "*/5 * * * *" || hook.Schedule.Timezone != "Asia/Shanghai" {
+ t.Fatalf("schedule = %+v", hook.Schedule)
+ }
+}
+
+func TestParseManifestValidatesScheduledHookContract(t *testing.T) {
+ tests := []struct {
+ name string
+ triggers []any
+ schedule any
+ transport any
+ want string
+ }{
+ {name: "missing schedule", triggers: []any{"schedule"}, want: "schedule is required"},
+ {name: "schedule without trigger", triggers: []any{"manual"}, schedule: map[string]any{"cron": "*/5 * * * *", "timezone": "UTC"}, want: "requires the schedule trigger"},
+ {name: "seconds field", triggers: []any{"schedule"}, schedule: map[string]any{"cron": "0 */5 * * * *", "timezone": "UTC"}, want: "five-field"},
+ {name: "inline timezone", triggers: []any{"schedule"}, schedule: map[string]any{"cron": "CRON_TZ=UTC */5 * * * *", "timezone": "UTC"}, want: "inline timezone"},
+ {name: "missing timezone", triggers: []any{"schedule"}, schedule: map[string]any{"cron": "*/5 * * * *", "timezone": ""}, want: "timezone must not be empty"},
+ {name: "invalid timezone", triggers: []any{"schedule"}, schedule: map[string]any{"cron": "*/5 * * * *", "timezone": "Mars/Olympus"}, want: "timezone is invalid"},
+ {name: "too frequent", triggers: []any{"schedule"}, schedule: map[string]any{"cron": "*/4 * * * *", "timezone": "UTC"}, want: "every five minutes"},
+ {name: "mcp transport", triggers: []any{"schedule"}, schedule: map[string]any{"cron": "*/5 * * * *", "timezone": "UTC"}, transport: map[string]any{"type": "mcp", "url": "https://example.com/mcp"}, want: "only supports the http transport"},
+ }
+ for _, tc := range tests {
+ t.Run(tc.name, func(t *testing.T) {
+ raw := mutate(t, func(doc map[string]any) {
+ contributes := doc["contributes"].(map[string]any)
+ hook := contributes["hooks"].([]any)[0].(map[string]any)
+ hook["triggers"] = tc.triggers
+ if tc.schedule != nil {
+ hook["schedule"] = tc.schedule
+ }
+ if tc.transport != nil {
+ hook["transport"] = tc.transport
+ }
+ })
+ _, _, err := ParseManifest(raw)
+ if err == nil || !strings.Contains(err.Error(), tc.want) {
+ t.Fatalf("error = %v, want it to mention %q", err, tc.want)
+ }
+ })
+ }
+}
+
func TestConfigSchemaPreservesDeclarationOrder(t *testing.T) {
manifest, canonical, err := ParseManifest([]byte(validManifest))
if err != nil {
@@ -440,7 +495,7 @@ func TestCheckCapabilitiesReportsEveryUnavailableContribution(t *testing.T) {
full := Capabilities{
SurfaceTypes: map[string]bool{SurfaceIssuePanel: true, SurfaceSidebarPanel: true, SurfaceModal: true},
- HookTriggers: map[string]bool{TriggerUI: true, TriggerManual: true, TriggerAgent: true, TriggerEvent: true},
+ HookTriggers: map[string]bool{TriggerUI: true, TriggerManual: true, TriggerAgent: true, TriggerEvent: true, TriggerSchedule: true},
HookTransport: map[string]bool{TransportHTTP: true, TransportMCP: true},
ResourceTypes: map[string]bool{ResourceSkill: true},
}