MUL-6582: add durable scheduled Plugin hooks (#7448)

* MUL-6582 feat(plugins): add durable scheduled hooks

Co-authored-by: multica-agent <[email protected]>

* MUL-6582: add schedule-pulse example plugin

Ship a real scheduled plugin that records each unique delivery_id in
workspace storage, so a host retry does not look like a second pulse.
The cmd/server test installs this example manifest and drives two
scheduler replicas through a dropped first attempt.

Co-authored-by: multica-agent <[email protected]>

* MUL-6582 fix(migrations): make schedule indexes retry-safe

Co-authored-by: multica-agent <[email protected]>

---------

Co-authored-by: multica-agent <[email protected]>
This commit is contained in:
Jiayuan Zhang
2026-08-23 17:37:21 +08:00
committed by GitHub
co-authored by multica-agent
parent a4e1ea7140
commit 5bf10d485c
47 changed files with 2917 additions and 81 deletions
+48
View File
@@ -0,0 +1,48 @@
# Schedule Pulse
A real scheduled plugin, not a log line. Multica POSTs the plugin's own HTTPS
endpoint every five minutes. The handler records each unique `delivery_id` in
workspace storage **before** it answers, so a retry of the same planned wake
does not look like a second pulse.
The issue panel reads that storage row. After install you should see the count
go up about once every five minutes, not on every HTTP attempt.
## What it verifies
| Host behavior | What this plugin does |
| --- | --- |
| Durable schedule trigger | Declares `triggers: ["schedule"]` and a five-minute cron |
| At-least-once delivery | Persists `delivery_id` first; a retry with the same id returns `{duplicate: true}` |
| Plugin actor | Writes as the installation through the callback token, into `storage:workspace` |
| Consent | `net:pulse.example.com` is the only outbound host; schedule is not a data scope |
It does not create issues. That is a later Plugin Action API, not this trigger.
## Running the handler
```bash
MULTICA_SIGNING_SECRET=whsec_… node server/handler.mjs
```
The signing secret is shown once, next to the install token, when an admin
rotates the plugin's token in workspace settings.
Point `transport.url` at this process and declare the matching `net:` scope.
Local development also needs:
```bash
export MULTICA_PLUGIN_DEV_ORIGINS=https://localhost:8787
```
Without that opt-in the host refuses a private address.
## Why storage, not a log
A log line cannot prove idempotency. The panel shows `count`, `delivery_id`,
`planned_at`, and `attempt`. If the host retries the same plan, `attempt`
changes and `count` does not.
The callback token is revoked when this HTTP request returns. All storage
writes happen before the 200. Work that must continue after the response
should use the standing `mpi_` install token instead.
@@ -0,0 +1,48 @@
{
"manifest_version": 1,
"key": "ai.multica.schedule-pulse",
"name": "Schedule Pulse",
"description": "A scheduled Hook that Multica wakes every five minutes. Each unique delivery is recorded once in workspace storage, including when the host retries.",
"version": "1.0.0",
"author": {
"name": "multica",
"url": "https://multica.ai"
},
"scopes": [
"storage:workspace",
"net:pulse.example.com"
],
"contributes": {
"surfaces": [
{
"key": "pulse",
"type": "issue_panel",
"name": "Schedule pulse",
"entry": "ui/main.js",
"platforms": [
"web",
"desktop"
]
}
],
"hooks": [
{
"key": "pulse",
"name": "Schedule pulse",
"description": "Records one workspace-storage pulse per planned delivery. Retries of the same delivery_id do not increment the count.",
"triggers": [
"schedule"
],
"schedule": {
"cron": "*/5 * * * *",
"timezone": "UTC"
},
"transport": {
"type": "http",
"url": "https://pulse.example.com/hooks/pulse"
},
"timeout_ms": 8000
}
]
}
}
@@ -0,0 +1,149 @@
#!/usr/bin/env node
/**
* Schedule Pulse handler — the plugin author's side of a durable scheduled Hook.
*
* Run:
* MULTICA_SIGNING_SECRET=whsec_... node handler.mjs
*/
import { createServer as createHTTPServer } from "node:http";
import { createServer as createHTTPSServer } from "node:https";
import { createHmac, timingSafeEqual } from "node:crypto";
import { readFileSync } from "node:fs";
const PORT = Number(process.env.PORT ?? 8787);
const SIGNING_SECRET = process.env.MULTICA_SIGNING_SECRET ?? "";
const TOLERANCE_SECONDS = 5 * 60;
const PULSE_KEY = "last_pulse";
if (!SIGNING_SECRET) {
console.error("MULTICA_SIGNING_SECRET is required. Rotate the plugin token in Multica to obtain it.");
process.exit(1);
}
const seenSignatures = new Map();
function rememberSignature(signature, now) {
for (const [seen, at] of seenSignatures) {
if (now - at > TOLERANCE_SECONDS * 1000) seenSignatures.delete(seen);
}
if (seenSignatures.has(signature)) return false;
seenSignatures.set(signature, now);
return true;
}
function verify(rawBody, headers) {
const timestamp = headers["x-multica-timestamp"];
const presented = String(headers["x-multica-signature"] ?? "").replace(/^v1=/, "");
if (!timestamp || !presented) return "missing signature headers";
const drift = Math.abs(Math.floor(Date.now() / 1000) - Number(timestamp));
if (!Number.isFinite(drift) || drift > TOLERANCE_SECONDS) return "timestamp outside the accepted window";
const secret = Buffer.from(SIGNING_SECRET.replace(/^whsec_/, ""), "hex");
const expected = createHmac("sha256", secret)
.update(timestamp)
.update(".")
.update(rawBody)
.digest("hex");
const expectedBytes = Buffer.from(expected, "utf8");
const presentedBytes = Buffer.from(presented, "utf8");
if (expectedBytes.length !== presentedBytes.length) return "signature does not match";
if (!timingSafeEqual(expectedBytes, presentedBytes)) return "signature does not match";
if (!rememberSignature(presented, Date.now())) return "this request was already delivered";
return null;
}
async function callback(body, method, path, payload) {
const response = await fetch(`${body.callback_url}${path}`, {
method,
headers: {
"Content-Type": "application/json",
Authorization: `Bearer ${body.callback_token}`,
},
body: payload === undefined ? undefined : JSON.stringify(payload),
});
const text = await response.text();
if (response.status === 404) return null;
if (!response.ok) {
throw new Error(`Multica answered ${response.status}: ${text}`);
}
return text ? JSON.parse(text) : null;
}
async function loadPulse(body) {
const stored = await callback(body, "GET", `/storage/workspace/${PULSE_KEY}`);
if (!stored?.value) return { count: 0, delivery_id: "" };
try {
return JSON.parse(stored.value);
} catch {
return { count: 0, delivery_id: "" };
}
}
const tlsCert = process.env.MULTICA_HOOK_TLS_CERT;
const tlsKey = process.env.MULTICA_HOOK_TLS_KEY;
const createServer = tlsCert && tlsKey
? (handler) => createHTTPSServer({ cert: readFileSync(tlsCert), key: readFileSync(tlsKey) }, handler)
: createHTTPServer;
const server = createServer((request, response) => {
if (request.method !== "POST" || request.url !== "/hooks/pulse") {
response.writeHead(404).end();
return;
}
const chunks = [];
request.on("data", (chunk) => chunks.push(chunk));
request.on("end", async () => {
const rawBody = Buffer.concat(chunks);
const failure = verify(rawBody, request.headers);
if (failure) {
console.warn(`rejected: ${failure}`);
response.writeHead(401, { "Content-Type": "application/json" });
response.end(JSON.stringify({ error: failure }));
return;
}
const body = JSON.parse(rawBody.toString("utf8"));
if (body.hook_key !== "pulse" || body.trigger !== "schedule") {
response.writeHead(200, { "Content-Type": "application/json" });
response.end(JSON.stringify({ skipped: "not a scheduled pulse" }));
return;
}
if (!body.delivery_id) {
response.writeHead(400, { "Content-Type": "application/json" });
response.end(JSON.stringify({ error: "scheduled delivery is missing delivery_id" }));
return;
}
try {
const previous = await loadPulse(body);
if (previous.delivery_id === body.delivery_id) {
console.log(`duplicate delivery ${body.delivery_id} attempt ${body.attempt}`);
response.writeHead(200, { "Content-Type": "application/json" });
response.end(JSON.stringify({ duplicate: true, delivery_id: body.delivery_id, count: previous.count }));
return;
}
const pulse = {
delivery_id: body.delivery_id,
count: (previous.count ?? 0) + 1,
last_planned_at: body.schedule?.planned_at ?? "",
last_occurred_at: body.occurred_at ?? "",
last_attempt: body.attempt ?? 1,
};
await callback(body, "PUT", `/storage/workspace/${PULSE_KEY}`, { value: JSON.stringify(pulse) });
console.log(`pulse ${pulse.count} delivery ${pulse.delivery_id} attempt ${pulse.last_attempt}`);
response.writeHead(200, { "Content-Type": "application/json" });
response.end(JSON.stringify({ accepted: true, ...pulse }));
} catch (error) {
console.error(error);
response.writeHead(502, { "Content-Type": "application/json" });
response.end(JSON.stringify({ error: "pulse failed" }));
}
});
});
server.listen(PORT, () => console.log(`schedule pulse listening on ${tlsCert ? "https" : "http"}://127.0.0.1:${PORT}`));
@@ -0,0 +1,67 @@
// Schedule Pulse — reads the workspace-storage row the scheduled Hook writes.
//
// Plain ES module, no build step. The surface has no credential; it talks to
// Multica only through the host bridge.
const pending = new Map();
const port = globalThis.__multicaPluginBridgePortV2;
let sequence = 0;
if (!(port instanceof MessagePort)) throw new Error("Multica surface bridge is unavailable");
delete globalThis.__multicaPluginBridgePortV2;
port.onmessage = (message) => {
const payload = message.data;
if (payload?.kind === "theme") return applyTheme(payload.theme);
const entry = pending.get(payload?.id);
if (!entry) return;
pending.delete(payload.id);
if (payload.ok) entry.resolve(payload.data);
else entry.reject(new Error(payload.error));
};
port.start();
start();
function applyTheme(theme) {
for (const [name, value] of Object.entries(theme ?? {})) {
document.documentElement.style.setProperty(name, value);
}
}
function call(method, path, body) {
const id = `r${++sequence}`;
return new Promise((resolve, reject) => {
pending.set(id, { resolve, reject });
port.postMessage({ id, kind: "action", method, path, body });
});
}
function resize() {
port.postMessage({ id: `resize${Date.now()}`, kind: "ui.resize", height: document.body.scrollHeight + 16 });
}
async function start() {
const root = document.getElementById("root");
root.innerHTML = `
<div style="padding:12px 14px; display:grid; gap:8px">
<div style="font-weight:600">Schedule pulse</div>
<div id="status" style="color:var(--muted-foreground)">Loading…</div>
<dl id="pulse" style="display:none; margin:0; grid-template-columns:auto 1fr; gap:4px 12px"></dl>
</div>`;
const status = root.querySelector("#status");
const list = root.querySelector("#pulse");
try {
const stored = await call("GET", "/storage/workspace/last_pulse");
const pulse = JSON.parse(stored.value);
status.textContent = "Multica has woken this plugin on a durable five-minute schedule.";
list.style.display = "grid";
list.innerHTML = `
<dt style="color:var(--muted-foreground)">Count</dt><dd style="margin:0">${pulse.count ?? 0}</dd>
<dt style="color:var(--muted-foreground)">Delivery</dt><dd style="margin:0; overflow-wrap:anywhere">${pulse.delivery_id ?? "—"}</dd>
<dt style="color:var(--muted-foreground)">Planned</dt><dd style="margin:0">${pulse.last_planned_at ?? "—"}</dd>
<dt style="color:var(--muted-foreground)">Attempt</dt><dd style="margin:0">${pulse.last_attempt ?? "—"}</dd>`;
} catch {
status.textContent = "No pulse yet. After install, Multica calls this plugin every five minutes and writes the first delivery here.";
}
resize();
}
+19 -3
View File
@@ -7,16 +7,18 @@ The three previous examples only ever ran inbound — a sandboxed panel asking t
host for things. This one has a backend, which is what makes every check in the
hook engine necessary.
## The three triggers
## The four triggers
The same hook, reached three ways. What differs is who caused it, and therefore
whose name the resulting comment carries.
The triage hook is reached three ways; the heartbeat adds a fourth, scheduled
trigger. What differs is who caused each call and therefore whose name any
resulting comment carries.
| Trigger | Where it comes from | Blocks? | Comment author |
| --- | --- | --- | --- |
| `ui` | The **Triage this issue** button in the panel | The panel only | The person who clicked |
| `manual` | The **Triage this issue** entry in the issue actions menu | That menu only | The person who picked it |
| `event` | Automatically, when an issue is created | **Never** | The plugin itself |
| `schedule` | Every five minutes in UTC | **Never** | The plugin itself |
`event` never blocking is not a performance tuning choice. The event bus runs its
listeners inline on the goroutine of the request that published, so a hook
@@ -58,6 +60,20 @@ it (recorded with `via_plugin_id`), an `event` call writes as the plugin. A
handler cannot elect to write as somebody else, which is the reason the callback
token exists instead of just handing out the install token.
The callback token is revoked when this HTTP request returns and is held by one
Multica server instance. Use it only for work completed before responding. A
scheduled integration that needs to continue asynchronously or reconcile a
large external backlog should store the `mpi_` install token shown to the admin
at token rotation and use that standing credential instead.
## Scheduled delivery identity
The `scheduled_heartbeat` Hook is intentionally side-effect free. It logs the
stable `delivery_id`, per-attempt `invocation_id` and `attempt`, plus
`schedule.planned_at`. A real scheduled Hook must persist `delivery_id` before
performing side effects: network timeouts and stale-lease recovery can deliver
the same planned occurrence more than once.
## Where a plugin can send data
`net:triage.example.com` in the manifest is the whole answer, and the admin sees
@@ -2,8 +2,8 @@
"manifest_version": 1,
"key": "ai.multica.triage-notify",
"name": "Triage Notify",
"description": "Sends an issue to an external triage service and writes its answer back as a comment. Demonstrates all three host-driven hook triggers: a button in the panel, an entry in the issue menu, and automatic dispatch when an issue is created.",
"version": "1.0.0",
"description": "Sends an issue to an external triage service and writes its answer back as a comment. Also demonstrates a no-side-effect Hook that Multica wakes every five minutes.",
"version": "1.1.0",
"author": {
"name": "multica",
"url": "https://multica.ai"
@@ -62,7 +62,24 @@
"url": "https://triage.example.com/hooks/triage"
},
"timeout_ms": 8000
},
{
"key": "scheduled_heartbeat",
"name": "Scheduled heartbeat",
"description": "Acknowledges a durable scheduled delivery without writing to Multica or another service.",
"triggers": [
"schedule"
],
"schedule": {
"cron": "*/5 * * * *",
"timezone": "UTC"
},
"transport": {
"type": "http",
"url": "https://triage.example.com/hooks/heartbeat"
},
"timeout_ms": 8000
}
]
}
}
}
@@ -120,7 +120,7 @@ const createServer = tlsCert && tlsKey
: createHTTPServer;
const server = createServer((request, response) => {
if (request.method !== "POST" || !request.url?.startsWith("/hooks/triage")) {
if (request.method !== "POST" || !request.url?.startsWith("/hooks/")) {
response.writeHead(404).end();
return;
}
@@ -144,6 +144,18 @@ const server = createServer((request, response) => {
console.log(`hook ${body.hook_key} via ${body.trigger}${body.event_type ? ` (${body.event_type})` : ""} as ${body.actor?.type}`);
try {
// A scheduled delivery has no issue or caller input. delivery_id is
// stable across retries while invocation_id changes per HTTP attempt; a
// real side-effecting handler persists delivery_id before doing work.
if (body.hook_key === "scheduled_heartbeat" && body.trigger === "schedule") {
console.log(
`scheduled delivery ${body.delivery_id} attempt ${body.attempt} planned ${body.schedule?.planned_at}`,
);
response.writeHead(200, { "Content-Type": "application/json" });
response.end(JSON.stringify({ received: body.delivery_id }));
return;
}
// The host tells us which issue this is about, having already resolved and
// permission-checked it. Preferred over anything in `input`: for an event
// trigger there was no client to supply it, and for ui/manual a
+27
View File
@@ -1670,6 +1670,33 @@ describe("Plugin schemas", () => {
expect(parsed.installed_version).toBe("1.0.0");
expect(parsed.added_scopes).toEqual(["comments:write"]);
});
it("preserves automatic schedules on both consent and installed Plugin payloads", () => {
const schedule = { cron: "*/5 * * * *", timezone: "Asia/Shanghai" };
const preview = PluginPreviewSchema.parse({
manifest: {
key: "com.example.digest",
name: "Digest",
version: "1.0.0",
author: { name: "example" },
contributes: {
hooks: [{ key: "digest", name: "Digest", triggers: ["schedule"], schedule }],
},
},
});
expect(preview.manifest.contributes?.hooks?.[0]?.schedule).toEqual(schedule);
const installation = PluginInstallationSchema.parse({
id: "installation-1",
hooks: [{
key: "digest",
name: "Digest",
triggers: ["schedule"],
schedule: { ...schedule, next_run_at: "2026-08-23T10:15:00Z" },
}],
});
expect(installation.hooks[0]?.schedule?.next_run_at).toBe("2026-08-23T10:15:00Z");
});
});
// Issue status catalog (MUL-6243). The catalog drives how every status renders,
+18
View File
@@ -112,6 +112,11 @@ export const PluginHookSchema = z.object({
description: z.string().default(""),
triggers: z.array(z.string()).default([]),
events: z.array(z.string()).default([]),
schedule: z.object({
cron: z.string().default(""),
timezone: z.string().default(""),
next_run_at: z.string().optional(),
}).loose().optional(),
transport: z.string().default(""),
}).loose();
@@ -190,6 +195,8 @@ export const PluginInvocationSchema = z.object({
attempt: z.number().default(1),
latency_ms: z.number().default(0),
error: z.string().optional(),
delivery_id: z.string().optional(),
planned_at: z.string().optional(),
created_at: z.string().default(""),
}).loose();
@@ -234,6 +241,17 @@ export const PluginManifestSummarySchema = z.object({
name: z.string().default(""),
url: z.string().optional(),
}).loose().default({ name: "" }),
contributes: z.object({
hooks: z.array(z.object({
key: z.string().default(""),
name: z.string().default(""),
triggers: z.array(z.string()).default([]),
schedule: z.object({
cron: z.string().default(""),
timezone: z.string().default(""),
}).loose().optional(),
}).loose()).default([]),
}).loose().optional(),
}).loose();
export const PluginPreviewSchema = z.object({
+19 -1
View File
@@ -29,7 +29,14 @@ export interface PluginSurface {
platforms?: string[];
}
export type PluginHookTrigger = "ui" | "manual" | "agent" | "event";
export type PluginHookTrigger = "ui" | "manual" | "agent" | "event" | "schedule";
export interface PluginHookSchedule {
cron: string;
timezone: string;
/** Display-only projection. Runtime correctness never depends on this value. */
next_run_at?: string;
}
export interface PluginHook {
key: string;
@@ -37,6 +44,7 @@ export interface PluginHook {
description: string;
triggers: (PluginHookTrigger | string)[];
events?: string[];
schedule?: PluginHookSchedule;
transport: string;
}
@@ -82,6 +90,14 @@ export interface PluginManifestSummary {
description?: string;
version: string;
author: { name: string; url?: string };
contributes?: {
hooks?: Array<{
key: string;
name: string;
triggers: (PluginHookTrigger | string)[];
schedule?: PluginHookSchedule;
}>;
};
}
/**
@@ -186,6 +202,8 @@ export interface PluginInvocation {
attempt: number;
latency_ms: number;
error?: string;
delivery_id?: string;
planned_at?: string;
created_at: string;
}
+14
View File
@@ -368,6 +368,20 @@
"upgrade_from": " · upgrading from v{{version}}",
"new_scope": "New"
},
"schedule": {
"title": "Automatic schedules",
"consent_title": "This Plugin will run hooks automatically",
"consent_description": "Installing or upgrading enables these schedules. If several times are missed, only the latest due run is delivered.",
"next_run": "Next: {{time}}",
"activity_title": "Recent deliveries",
"activity_last": "{{status}} · planned {{planned}} · sent {{actual}} · attempt {{attempt}}",
"activity_empty": "Not delivered yet",
"frequency_minutes_one": "Every minute",
"frequency_minutes_other": "Every {{count}} minutes",
"frequency_hourly": "Every hour at minute {{minute}}",
"frequency_daily": "Every day at {{hour}}:{{minute}}",
"frequency_custom": "Custom recurring schedule"
},
"config": {
"title": "Configuration",
"save": "Save",
+13
View File
@@ -368,6 +368,19 @@
"upgrade_from": " · v{{version}} からアップグレード",
"new_scope": "新規"
},
"schedule": {
"title": "自動スケジュール",
"consent_title": "この Plugin はフックを自動実行します",
"consent_description": "インストールまたはアップグレードすると、次のスケジュールが有効になります。複数回分を逃した場合は、直近の実行だけを配信します。",
"next_run": "次回:{{time}}",
"activity_title": "最近の配信",
"activity_last": "{{status}} · 予定 {{planned}} · 送信 {{actual}} · 試行 {{attempt}}",
"activity_empty": "まだ配信されていません",
"frequency_minutes_other": "{{count}} 分ごと",
"frequency_hourly": "毎時 {{minute}} 分",
"frequency_daily": "毎日 {{hour}}:{{minute}}",
"frequency_custom": "カスタム繰り返しスケジュール"
},
"config": {
"title": "設定",
"save": "保存",
+13
View File
@@ -368,6 +368,19 @@
"upgrade_from": " · v{{version}}에서 업그레이드",
"new_scope": "신규"
},
"schedule": {
"title": "자동 스케줄",
"consent_title": "이 Plugin은 훅을 자동 실행합니다",
"consent_description": "설치하거나 업그레이드하면 다음 스케줄이 활성화됩니다. 여러 실행 시점을 놓친 경우 가장 최근 실행만 전달합니다.",
"next_run": "다음 실행: {{time}}",
"activity_title": "최근 전달",
"activity_last": "{{status}} · 예정 {{planned}} · 전송 {{actual}} · 시도 {{attempt}}",
"activity_empty": "아직 전달되지 않음",
"frequency_minutes_other": "{{count}}분마다",
"frequency_hourly": "매시 {{minute}}분",
"frequency_daily": "매일 {{hour}}:{{minute}}",
"frequency_custom": "사용자 지정 반복 스케줄"
},
"config": {
"title": "설정",
"save": "저장",
@@ -368,6 +368,19 @@
"upgrade_from": " · 从 v{{version}} 升级",
"new_scope": "新增"
},
"schedule": {
"title": "自动运行计划",
"consent_title": "该 Plugin 将自动运行 hook",
"consent_description": "安装或升级后会启用以下计划。错过多个时间点时,只补跑最近一次。",
"next_run": "下次:{{time}}",
"activity_title": "最近投递",
"activity_last": "{{status}} · 计划 {{planned}} · 发出 {{actual}} · 第 {{attempt}} 次尝试",
"activity_empty": "尚未投递",
"frequency_minutes_other": "每 {{count}} 分钟",
"frequency_hourly": "每小时第 {{minute}} 分钟",
"frequency_daily": "每天 {{hour}}:{{minute}}",
"frequency_custom": "自定义周期"
},
"config": {
"title": "配置",
"save": "保存",
+1 -1
View File
@@ -17,5 +17,5 @@ export {
usePluginModalSurfaces,
} from "./plugin-modal-surface";
export type { PluginModalTarget } from "./plugin-modal-surface";
export { PluginHookActivity, summarizeInvocations } from "./plugin-hook-activity";
export { PluginHookActivity, PluginScheduleActivity, summarizeInvocations } from "./plugin-hook-activity";
export { PluginMCPApproval, mcpHooks, initialSelection } from "./plugin-mcp-approval";
@@ -3,9 +3,9 @@
import { useQuery } from "@tanstack/react-query";
import { AlertTriangle } from "lucide-react";
import { pluginInvocationsOptions } from "@multica/core/plugins";
import type { PluginInvocation } from "@multica/core/types";
import type { PluginHook, PluginInvocation } from "@multica/core/types";
import { cn } from "@multica/ui/lib/utils";
import { useT } from "../i18n";
import { useLocale, useT } from "../i18n";
/**
* Why a plugin's hook is failing, where the admin who installed it will look.
@@ -77,3 +77,60 @@ export function PluginHookActivity({ wsId, installationId }: { wsId: string; ins
</div>
);
}
export function PluginScheduleActivity({
wsId,
installationId,
hooks,
}: {
wsId: string;
installationId: string;
hooks: readonly PluginHook[];
}) {
const { t } = useT("settings");
const locale = useLocale();
const { data } = useQuery(pluginInvocationsOptions(wsId, installationId));
const scheduled = hooks.filter((hook) => hook.schedule !== undefined);
if (scheduled.length === 0) return null;
return (
<div className="space-y-1.5">
<div className="text-caption font-medium">{t(($) => $.plugins.schedule.activity_title)}</div>
<ul className="space-y-1">
{scheduled.map((hook) => {
const last = (data ?? []).find((invocation) => (
invocation.trigger === "schedule" && invocation.hook_key === hook.key
));
return (
<li key={hook.key} className="text-caption text-muted-foreground">
<span className="font-medium text-foreground">{hook.name}</span>
{last ? (
<span>
{" · "}
{t(($) => $.plugins.schedule.activity_last, {
status: last.status,
planned: formatInvocationTime(last.planned_at, locale),
actual: formatInvocationTime(last.created_at, locale),
attempt: last.attempt,
})}
</span>
) : (
<span>{" · "}{t(($) => $.plugins.schedule.activity_empty)}</span>
)}
</li>
);
})}
</ul>
</div>
);
}
function formatInvocationTime(value: string | undefined, locale: string): string {
if (!value) return "—";
const timestamp = Date.parse(value);
if (Number.isNaN(timestamp)) return "—";
return new Intl.DateTimeFormat(locale, {
dateStyle: "medium",
timeStyle: "short",
}).format(new Date(timestamp));
}
@@ -2,7 +2,7 @@
import { useMemo, useRef, useState } from "react";
import { useQuery } from "@tanstack/react-query";
import { AlertCircle, Loader2, Trash2, Upload } from "lucide-react";
import { AlertCircle, CalendarClock, Loader2, Trash2, Upload } from "lucide-react";
import { toast } from "sonner";
import { useCurrentMember } from "@multica/core/permissions";
import {
@@ -37,8 +37,8 @@ import {
import { Skeleton } from "@multica/ui/components/ui/skeleton";
import { Textarea } from "@multica/ui/components/ui/textarea";
import { Switch } from "@multica/ui/components/ui/switch";
import { mcpHooks, PluginHookActivity, PluginMCPApproval } from "../../plugins";
import { useT } from "../../i18n";
import { mcpHooks, PluginHookActivity, PluginMCPApproval, PluginScheduleActivity } from "../../plugins";
import { useLocale, useT } from "../../i18n";
import { SettingsCard, SettingsSection, SettingsTab } from "./settings-layout";
/**
@@ -65,8 +65,67 @@ function ScopeList({ scopes, highlighted }: { scopes: string[]; highlighted?: st
);
}
type ScheduledHook = {
key: string;
name: string;
schedule?: { cron: string; timezone: string; next_run_at?: string };
};
function ScheduleList({ hooks, showNextRun = false }: { hooks: ScheduledHook[]; showNextRun?: boolean }) {
const { t } = useT("settings");
const locale = useLocale();
return (
<ul className="mt-2 space-y-1.5">
{hooks.map((hook) => (
<li key={hook.key} className="flex flex-wrap items-baseline gap-2 text-caption">
<span className="font-medium">{hook.name}</span>
<span>{scheduleFrequency(hook.schedule?.cron ?? "", t)}</span>
<code className="rounded bg-muted px-1.5 py-0.5 font-mono">{hook.schedule?.cron ?? ""}</code>
<span className="text-muted-foreground">{hook.schedule?.timezone ?? ""}</span>
{showNextRun && hook.schedule?.next_run_at ? (
<span className="text-muted-foreground">
{t(($) => $.plugins.schedule.next_run, {
time: formatScheduleTime(hook.schedule?.next_run_at, locale),
})}
</span>
) : null}
</li>
))}
</ul>
);
}
function formatScheduleTime(value: string | undefined, locale: string): string {
if (!value) return "—";
const timestamp = Date.parse(value);
if (Number.isNaN(timestamp)) return "—";
return new Intl.DateTimeFormat(locale, {
dateStyle: "medium",
timeStyle: "short",
}).format(new Date(timestamp));
}
type Translate = ReturnType<typeof useT<"settings">>["t"];
function scheduleFrequency(cron: string, t: Translate): string {
const fields = cron.trim().split(/\s+/);
if (fields.length !== 5) return t(($) => $.plugins.schedule.frequency_custom);
const everyMinutes = /^\*\/(\d+)$/.exec(fields[0] ?? "");
if (everyMinutes && fields.slice(1).every((field) => field === "*")) {
return t(($) => $.plugins.schedule.frequency_minutes, { count: Number(everyMinutes[1]) });
}
if (/^\d+$/.test(fields[0] ?? "") && fields.slice(1).every((field) => field === "*")) {
return t(($) => $.plugins.schedule.frequency_hourly, { minute: fields[0] });
}
if (/^\d+$/.test(fields[0] ?? "") && /^\d+$/.test(fields[1] ?? "") && fields.slice(2).every((field) => field === "*")) {
return t(($) => $.plugins.schedule.frequency_daily, {
hour: String(fields[1]).padStart(2, "0"),
minute: String(fields[0]).padStart(2, "0"),
});
}
return t(($) => $.plugins.schedule.frequency_custom);
}
function scopeDescription(scope: string, t: Translate): string {
if (scope.startsWith("net:")) {
return t(($) => $.plugins.scopes.net, { domain: scope.slice("net:".length) });
@@ -270,6 +329,8 @@ function PublishAndInstall({ wsId, canManage }: { wsId: string; canManage: boole
const [preview, setPreview] = useState<PluginPreview | null>(null);
const packages = useMemo(() => data?.packages ?? [], [data]);
const scheduledHooks = (preview?.manifest.contributes?.hooks ?? [])
.filter((hook) => hook.schedule !== undefined);
const reportError = (error: unknown) => {
toast.error(error instanceof Error ? error.message : t(($) => $.plugins.action_failed));
@@ -386,6 +447,17 @@ function PublishAndInstall({ wsId, canManage }: { wsId: string; canManage: boole
<ScopeList scopes={preview.scopes} highlighted={preview.added_scopes} />
{scheduledHooks.length > 0 ? (
<Alert>
<CalendarClock />
<AlertTitle>{t(($) => $.plugins.schedule.consent_title)}</AlertTitle>
<AlertDescription>
{t(($) => $.plugins.schedule.consent_description)}
<ScheduleList hooks={scheduledHooks} />
</AlertDescription>
</Alert>
) : null}
<div className="flex justify-end gap-2">
<Button variant="ghost" onClick={() => setPreview(null)}>
{t(($) => $.plugins.consent.cancel)}
@@ -495,6 +567,7 @@ function InstalledPlugin({
// An mcp hook is inert until its tools are approved, so the panel appears for
// every one of them rather than only when something is already pinned.
const mcpContributions = mcpHooks(installation.hooks ?? []);
const scheduledHooks = (installation.hooks ?? []).filter((hook) => hook.schedule !== undefined);
const contributions = [
...installation.surfaces.map((surface) => `${surface.name} (${surface.type})`),
@@ -551,6 +624,21 @@ function InstalledPlugin({
<ScopeList scopes={installation.granted_scopes} />
</div>
{scheduledHooks.length > 0 ? (
<div className="space-y-1">
<div className="flex items-center gap-2 text-caption font-medium">
<CalendarClock className="size-4" />
{t(($) => $.plugins.schedule.title)}
</div>
<ScheduleList hooks={scheduledHooks} showNextRun />
<PluginScheduleActivity
wsId={wsId}
installationId={installation.id}
hooks={scheduledHooks}
/>
</div>
) : null}
{mcpContributions.length > 0 ? (
<div className="space-y-2">
<div className="text-caption font-medium">{t(($) => $.plugins.mcp.badge)}</div>
+2
View File
@@ -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
+5
View File
@@ -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)
}()
@@ -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)
}
}
}
@@ -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)
}
+33 -8
View File
@@ -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 {
+19 -9
View File
@@ -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})
+168 -1
View File
@@ -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{
@@ -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,
@@ -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[:])
}
@@ -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"])
}
}
+29 -4
View File
@@ -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}
}
+62 -18
View File
@@ -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}
}
+45 -1
View File
@@ -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) {
+131
View File
@@ -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
}
@@ -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;
@@ -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;
@@ -0,0 +1 @@
DROP INDEX CONCURRENTLY IF EXISTS idx_plugin_hook_schedule_installation_key;
@@ -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);
@@ -0,0 +1 @@
DROP INDEX CONCURRENTLY IF EXISTS idx_plugin_hook_schedule_enabled;
@@ -0,0 +1,3 @@
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_plugin_hook_schedule_enabled
ON plugin_hook_schedule (id)
WHERE enabled;
@@ -0,0 +1,3 @@
-- Validation changes no data or contract shape; migration 399 restores the old
-- constraint during rollback.
SELECT 1;
@@ -0,0 +1 @@
ALTER TABLE plugin_invocation VALIDATE CONSTRAINT plugin_invocation_trigger_check;
+17
View File
@@ -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 {
+315 -13
View File
@@ -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,
@@ -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
+72 -2
View File
@@ -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
@@ -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.
+5 -4
View File
@@ -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
+85 -6
View File
@@ -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.
+56 -1
View File
@@ -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},
}