mirror of
https://github.com/multica-ai/multica.git
synced 2026-07-13 13:18:56 +02:00
* feat(autopilots): webhook delivery layer + idempotency / signature / replay (MUL-2334)
Splits "inbound webhook receipt" from "autopilot run creation" so we can
record duplicate attempts, signature outcomes, and ignored/skipped
deliveries — and replay a delivery on demand. v1 ingress wrote straight
into autopilot_run.trigger_payload, which collapsed the two concerns and
left run_only autopilots vulnerable to provider retry storms.
Backend only (PR1). UI Deliveries tab follows in PR2.
Schema (migration 093):
- autopilot_trigger.provider: 'generic' | 'github' (default 'generic').
- autopilot_trigger.signing_secret: nullable plaintext (HMAC needs it
cleartext; mirrors how webhook_token is stored).
- webhook_delivery: one row per inbound POST. Carries raw_body,
selected_headers, dedupe_key/source, signature_status,
autopilot_run_id, replayed_from_delivery_id, response_status / body.
- Partial unique index on (trigger_id, dedupe_key) excludes NULL and
'rejected' rows, so a wrong-secret 401 does NOT permanently block a
future retry with the same X-GitHub-Delivery once the operator fixes
the secret.
Ingress flow (autopilot_webhook.go), persist-first + sync dispatch:
1. IP rate limit -> 2. token lookup -> 3. token rate limit ->
4. read raw body -> 5. autopilot/workspace cross-check ->
6. normalize JSON (400 without persistence on parse failure) ->
7. compute dedupe key + signature status ->
8. INSERT delivery (status=queued). On (trigger_id, dedupe_key)
unique-violation: bump attempt_count on existing row and return
the original delivery_id + autopilot_run_id with 200 ->
9. invalid/missing signature: UPDATE -> rejected, return 401 with
delivery_id (no dispatch, not replayable) ->
10. trigger disabled / autopilot paused/archived: UPDATE -> ignored,
return 200 ->
11. DispatchAutopilot synchronously, UPDATE -> dispatched/skipped/failed
with autopilot_run_id and the response body we returned ->
12. TouchAutopilotTriggerFiredAt and return 200.
No new long-running worker. A stale 'queued' row only happens if the
process dies between INSERT and UPDATE; that's a follow-up sweeper, not
this PR.
Authenticated API:
- GET /api/autopilots/{id}/deliveries (slim list)
- GET /api/autopilots/{id}/deliveries/{deliveryId} (with raw_body)
- POST /api/autopilots/{id}/deliveries/{deliveryId}/replay -> creates
a new delivery row (replayed_from_delivery_id set), dispatches a
new run, never collapses onto the original via dedupe.
- PUT /api/autopilots/{id}/triggers/{triggerId}/signing-secret
Write-only; trigger response surfaces has_signing_secret +
signing_secret_hint (last 4 chars), never the secret itself.
Signature verification reuses the GitHub-compatible
X-Hub-Signature-256: sha256=<hex(hmac(body, secret))> scheme; the
HMAC helper is constant-time. Invalid/missing signatures still count
against per-IP and per-token rate limits.
autopilot_run.trigger_payload is intentionally preserved — delivery
records the HTTP receipt; run records the normalized envelope handed
to the agent. They are two different views.
Tests (Postgres-backed):
- delivery persistence on accept
- dedupe via Idempotency-Key and X-GitHub-Delivery; run_only retry
storm pin (3 retries -> 1 run)
- invalid signature: 401 + rejected row + no run linkage
- missing signature when secret configured: 401 + 'missing' state
- valid signature dispatches
- signing secret never echoed in trigger responses; hint shows last 4
- min-length and clear-by-empty for signing secret PUT
- replay creates a NEW delivery + new run; rejected deliveries cannot
be replayed
- list omits raw_body; detail includes it; cross-autopilot ID returns
404 (workspace isolation defense in depth)
- provider validation: unknown -> 400, github -> 201 round-trips
- bad-signature stream still counts against per-token rate limit
Co-authored-by: multica-agent <github@multica.ai>
* fix(autopilots): address PR review on webhook delivery layer (MUL-2334)
- Exclude `failed` from the (trigger_id, dedupe_key) partial unique index
alongside `rejected`, so a transient ingress failure does not strand the
provider's stable X-GitHub-Delivery / Idempotency-Key retry. Update the
dedupe lookup to prefer non-terminal rows under the same predicate.
- Tighten delivery status enum: drop `skipped` from the CHECK constraint
and from the handler. A run that was admission-skipped (e.g. runtime
offline) is now recorded as delivery=`dispatched` linked to the
skipped run, with the response payload carrying status=`skipped`.
Source of truth for skipped-ness is autopilot_run.status, not the
delivery row — keeps the Deliveries UI enum unambiguous.
- On dispatch error, link the (possibly non-nil) autopilot_run returned
by DispatchAutopilot to the failed delivery so Deliveries UI can
navigate to the run row for debugging.
- Slim list projection: ListWebhookDeliveriesByAutopilot no longer pulls
raw_body / selected_headers / response_body — a 100-row page × 256 KiB
would otherwise round-trip ~25 MiB from Postgres per Deliveries reload.
Detail endpoint continues to return the full row.
- Fix backend CI: TestGetDelivery_ReturnsFullPayload now decodes the
response and asserts on the parsed raw_body instead of substring-
matching against an escaped JSON string; raise the test-suite default
webhook rate limits in TestMain so the shared 192.0.2.1 IP bucket
doesn't fill across the suite and leak 429s into unrelated tests.
- Add regression coverage for the dedupe-after-failure path.
cd server && go test ./... is green locally.
Co-authored-by: multica-agent <github@multica.ai>
---------
Co-authored-by: multica-agent <github@multica.ai>
322 lines
11 KiB
SQL
322 lines
11 KiB
SQL
-- =====================
|
|
-- Autopilot CRUD
|
|
-- =====================
|
|
|
|
-- name: ListAutopilots :many
|
|
SELECT * FROM autopilot
|
|
WHERE workspace_id = $1
|
|
AND (sqlc.narg('status')::text IS NULL OR status = sqlc.narg('status'))
|
|
ORDER BY created_at DESC;
|
|
|
|
-- name: GetAutopilot :one
|
|
SELECT * FROM autopilot
|
|
WHERE id = $1;
|
|
|
|
-- name: GetAutopilotInWorkspace :one
|
|
SELECT * FROM autopilot
|
|
WHERE id = $1 AND workspace_id = $2;
|
|
|
|
-- name: CreateAutopilot :one
|
|
INSERT INTO autopilot (
|
|
workspace_id, title, description, assignee_id,
|
|
status, execution_mode, issue_title_template,
|
|
created_by_type, created_by_id
|
|
) VALUES (
|
|
$1, $2, sqlc.narg('description'), $3,
|
|
$4, $5, sqlc.narg('issue_title_template'),
|
|
$6, $7
|
|
) RETURNING *;
|
|
|
|
-- name: UpdateAutopilot :one
|
|
UPDATE autopilot SET
|
|
title = COALESCE(sqlc.narg('title'), title),
|
|
description = COALESCE(sqlc.narg('description'), description),
|
|
assignee_id = COALESCE(sqlc.narg('assignee_id')::uuid, assignee_id),
|
|
status = COALESCE(sqlc.narg('status'), status),
|
|
execution_mode = COALESCE(sqlc.narg('execution_mode'), execution_mode),
|
|
issue_title_template = sqlc.narg('issue_title_template'),
|
|
updated_at = now()
|
|
WHERE id = $1
|
|
RETURNING *;
|
|
|
|
-- name: DeleteAutopilot :exec
|
|
DELETE FROM autopilot WHERE id = $1;
|
|
|
|
-- name: UpdateAutopilotLastRunAt :exec
|
|
UPDATE autopilot SET last_run_at = now(), updated_at = now()
|
|
WHERE id = $1;
|
|
|
|
-- =====================
|
|
-- Autopilot Trigger CRUD
|
|
-- =====================
|
|
|
|
-- name: ListAutopilotTriggers :many
|
|
SELECT * FROM autopilot_trigger
|
|
WHERE autopilot_id = $1
|
|
ORDER BY created_at ASC;
|
|
|
|
-- name: GetAutopilotTrigger :one
|
|
SELECT * FROM autopilot_trigger
|
|
WHERE id = $1;
|
|
|
|
-- name: CreateAutopilotTrigger :one
|
|
INSERT INTO autopilot_trigger (
|
|
autopilot_id, kind, enabled, cron_expression, timezone,
|
|
next_run_at, webhook_token, label, provider
|
|
) VALUES (
|
|
$1, $2, $3, sqlc.narg('cron_expression'), sqlc.narg('timezone'),
|
|
sqlc.narg('next_run_at'), sqlc.narg('webhook_token'), sqlc.narg('label'),
|
|
COALESCE(sqlc.narg('provider')::text, 'generic')
|
|
) RETURNING *;
|
|
|
|
-- name: UpdateAutopilotTrigger :one
|
|
UPDATE autopilot_trigger SET
|
|
enabled = COALESCE(sqlc.narg('enabled')::boolean, enabled),
|
|
cron_expression = COALESCE(sqlc.narg('cron_expression'), cron_expression),
|
|
timezone = COALESCE(sqlc.narg('timezone'), timezone),
|
|
next_run_at = sqlc.narg('next_run_at'),
|
|
label = COALESCE(sqlc.narg('label'), label),
|
|
updated_at = now()
|
|
WHERE id = $1
|
|
RETURNING *;
|
|
|
|
-- name: DeleteAutopilotTrigger :exec
|
|
DELETE FROM autopilot_trigger WHERE id = $1;
|
|
|
|
-- name: AdvanceTriggerNextRun :exec
|
|
UPDATE autopilot_trigger
|
|
SET next_run_at = sqlc.narg('next_run_at'),
|
|
last_fired_at = now(),
|
|
updated_at = now()
|
|
WHERE id = $1;
|
|
|
|
-- name: GetWebhookTriggerByToken :one
|
|
-- Look up a webhook trigger by its public bearer token. Joined to autopilot
|
|
-- so the webhook handler can derive the workspace from the trigger's parent
|
|
-- without trusting any request header. The handler still re-loads the
|
|
-- Autopilot via GetAutopilot and cross-checks WorkspaceID matches the row's
|
|
-- autopilot_workspace_id.
|
|
SELECT t.*, a.workspace_id AS autopilot_workspace_id
|
|
FROM autopilot_trigger t
|
|
JOIN autopilot a ON a.id = t.autopilot_id
|
|
WHERE t.kind = 'webhook'
|
|
AND t.webhook_token = $1;
|
|
|
|
-- name: TouchAutopilotTriggerFiredAt :exec
|
|
-- Bumps last_fired_at after a webhook fires, regardless of whether the
|
|
-- dispatch succeeded, was admission-skipped, or even if Autopilot status
|
|
-- transitioned to paused/disabled at exactly the wrong moment. Disabled /
|
|
-- paused early-return paths in the handler never call this.
|
|
UPDATE autopilot_trigger
|
|
SET last_fired_at = now(),
|
|
updated_at = now()
|
|
WHERE id = $1;
|
|
|
|
-- name: RotateAutopilotTriggerWebhookToken :one
|
|
-- Rotates the bearer token for a webhook trigger. Restricted to kind='webhook'
|
|
-- so an accidental call against a schedule/api trigger is a no-op (returns no
|
|
-- rows) rather than corrupting unrelated state.
|
|
UPDATE autopilot_trigger
|
|
SET webhook_token = $2,
|
|
updated_at = now()
|
|
WHERE id = $1
|
|
AND kind = 'webhook'
|
|
RETURNING *;
|
|
|
|
-- name: SetAutopilotTriggerWebhookToken :one
|
|
-- Sets the webhook token at creation time. CreateAutopilotTrigger inserts the
|
|
-- row first (using its full 8-arg signature), then this query attaches the
|
|
-- token. Splitting the create + token-set keeps the existing CreateAutopilotTrigger
|
|
-- query usable by the schedule path without forcing every caller to think
|
|
-- about webhook_token.
|
|
UPDATE autopilot_trigger
|
|
SET webhook_token = $2,
|
|
updated_at = now()
|
|
WHERE id = $1
|
|
RETURNING *;
|
|
|
|
-- name: SetAutopilotTriggerSigningSecret :one
|
|
-- Writes the signing secret for a webhook trigger. Kept as a dedicated query
|
|
-- (not a field on UpdateAutopilotTrigger) so the request body for the
|
|
-- write-only endpoint only ever carries the secret value, with no risk of an
|
|
-- accidental log line leaking it alongside other fields. Restricted to
|
|
-- webhook triggers to avoid corrupting unrelated state.
|
|
UPDATE autopilot_trigger
|
|
SET signing_secret = sqlc.narg('signing_secret'),
|
|
updated_at = now()
|
|
WHERE id = $1
|
|
AND kind = 'webhook'
|
|
RETURNING *;
|
|
|
|
-- =====================
|
|
-- Autopilot Run Management
|
|
-- =====================
|
|
|
|
-- name: CreateAutopilotRun :one
|
|
INSERT INTO autopilot_run (
|
|
autopilot_id, trigger_id, source, status, trigger_payload
|
|
) VALUES (
|
|
$1, sqlc.narg('trigger_id'), $2, $3, sqlc.narg('trigger_payload')
|
|
) RETURNING *;
|
|
|
|
-- name: GetAutopilotRun :one
|
|
SELECT * FROM autopilot_run
|
|
WHERE id = $1;
|
|
|
|
-- name: ListAutopilotRuns :many
|
|
SELECT * FROM autopilot_run
|
|
WHERE autopilot_id = $1
|
|
ORDER BY created_at DESC
|
|
LIMIT $2 OFFSET $3;
|
|
|
|
-- name: UpdateAutopilotRunIssueCreated :one
|
|
UPDATE autopilot_run
|
|
SET status = 'issue_created', issue_id = $2
|
|
WHERE id = $1
|
|
RETURNING *;
|
|
|
|
-- name: UpdateAutopilotRunRunning :one
|
|
UPDATE autopilot_run
|
|
SET status = 'running', task_id = $2
|
|
WHERE id = $1
|
|
RETURNING *;
|
|
|
|
-- name: UpdateAutopilotRunCompleted :one
|
|
UPDATE autopilot_run
|
|
SET status = 'completed', completed_at = now(), result = sqlc.narg('result')
|
|
WHERE id = $1
|
|
RETURNING *;
|
|
|
|
-- name: UpdateAutopilotRunFailed :one
|
|
UPDATE autopilot_run
|
|
SET status = 'failed', completed_at = now(), failure_reason = $2
|
|
WHERE id = $1
|
|
RETURNING *;
|
|
|
|
-- name: UpdateAutopilotRunSkipped :one
|
|
-- Marks an autopilot_run as skipped without enqueueing any task. Used by the
|
|
-- pre-flight admission check when the assignee agent's runtime is offline:
|
|
-- creating an issue / task in that state would just pile a doomed job onto
|
|
-- agent_task_queue (the canonical "持续给离线 local agent 入队" symptom from
|
|
-- MUL-1899). Recording the skip + reason gives the UI / failure monitor / ops
|
|
-- a paper trail without polluting the failure ratio.
|
|
UPDATE autopilot_run
|
|
SET status = 'skipped', completed_at = now(), failure_reason = $2
|
|
WHERE id = $1
|
|
RETURNING *;
|
|
|
|
-- name: UpdateAutopilotRunSkippedWithResult :one
|
|
UPDATE autopilot_run
|
|
SET status = 'skipped',
|
|
completed_at = now(),
|
|
failure_reason = $2,
|
|
result = sqlc.narg('result')
|
|
WHERE id = $1
|
|
RETURNING *;
|
|
|
|
-- =====================
|
|
-- Scheduler Queries
|
|
-- =====================
|
|
|
|
-- name: ClaimDueScheduleTriggers :many
|
|
-- Atomically claim all due schedule triggers to prevent concurrent execution.
|
|
-- Joins the autopilot table to ensure only active autopilots are fired.
|
|
UPDATE autopilot_trigger t
|
|
SET next_run_at = NULL
|
|
FROM autopilot a
|
|
WHERE t.autopilot_id = a.id
|
|
AND t.kind = 'schedule'
|
|
AND t.enabled = true
|
|
AND t.next_run_at IS NOT NULL
|
|
AND t.next_run_at <= now()
|
|
AND a.status = 'active'
|
|
RETURNING t.*, a.workspace_id AS autopilot_workspace_id;
|
|
|
|
-- =====================
|
|
-- Task Queue (run_only mode)
|
|
-- =====================
|
|
|
|
-- name: CreateAutopilotTask :one
|
|
INSERT INTO agent_task_queue (agent_id, runtime_id, issue_id, status, priority, autopilot_run_id, trigger_summary)
|
|
VALUES ($1, $2, NULL, 'queued', $3, $4, sqlc.narg(trigger_summary))
|
|
RETURNING *;
|
|
|
|
-- =====================
|
|
-- Run lookup by linked entities
|
|
-- =====================
|
|
|
|
-- name: GetAutopilotRunByIssue :one
|
|
SELECT * FROM autopilot_run
|
|
WHERE issue_id = $1 AND status IN ('issue_created', 'running')
|
|
LIMIT 1;
|
|
|
|
-- name: FailAutopilotRunsByIssue :exec
|
|
-- Fails active autopilot runs linked to a given issue.
|
|
-- Must be called BEFORE issue deletion (ON DELETE SET NULL clears issue_id).
|
|
UPDATE autopilot_run
|
|
SET status = 'failed', completed_at = now(), failure_reason = 'linked issue was deleted'
|
|
WHERE issue_id = $1
|
|
AND status IN ('issue_created', 'running');
|
|
|
|
-- =====================
|
|
-- Scheduler Recovery
|
|
-- =====================
|
|
|
|
-- name: RecoverLostTriggers :many
|
|
-- Finds schedule triggers that were claimed (next_run_at = NULL) but never
|
|
-- advanced — typically due to a scheduler crash. Returns them so the scheduler
|
|
-- can recompute next_run_at.
|
|
SELECT t.*, a.workspace_id AS autopilot_workspace_id
|
|
FROM autopilot_trigger t
|
|
JOIN autopilot a ON t.autopilot_id = a.id
|
|
WHERE t.kind = 'schedule'
|
|
AND t.enabled = true
|
|
AND t.next_run_at IS NULL
|
|
AND t.cron_expression IS NOT NULL
|
|
AND a.status = 'active';
|
|
|
|
-- =====================
|
|
-- Failure-rate auto-pause
|
|
-- =====================
|
|
|
|
-- name: SelectAutopilotsExceedingFailureThreshold :many
|
|
-- Find active autopilots whose recent run failure rate exceeds the threshold.
|
|
-- Counts only "real" terminal runs (completed | failed). 'skipped' is
|
|
-- excluded from BOTH numerator and denominator: an admission-skipped run
|
|
-- (e.g. assignee runtime offline at dispatch time, MUL-1899) is neither a
|
|
-- success nor a failure, so it must not dilute the failure ratio (which
|
|
-- would let a 100%-failing autopilot mask itself behind a wall of skips)
|
|
-- nor inflate it. issue_created/running are still excluded so in-flight
|
|
-- work isn't penalised.
|
|
-- Used by the failure monitor to auto-pause sustained-failure autopilots
|
|
-- (the canonical example from MUL-1336 was an autopilot scheduled every 5 min
|
|
-- that 100% failed for days, burning ~1.5k useless tasks per week).
|
|
WITH stats AS (
|
|
SELECT autopilot_id,
|
|
count(*) FILTER (WHERE status IN ('completed', 'failed')) AS total,
|
|
count(*) FILTER (WHERE status = 'failed') AS failed
|
|
FROM autopilot_run
|
|
WHERE created_at >= sqlc.arg('since')::timestamptz
|
|
GROUP BY autopilot_id
|
|
)
|
|
SELECT a.id, a.workspace_id, a.title, a.assignee_id,
|
|
a.created_by_type, a.created_by_id,
|
|
s.total::bigint AS total_runs,
|
|
s.failed::bigint AS failed_runs
|
|
FROM autopilot a
|
|
JOIN stats s ON s.autopilot_id = a.id
|
|
WHERE a.status = 'active'
|
|
AND s.total >= sqlc.arg('min_runs')::bigint
|
|
AND s.failed::float8 / NULLIF(s.total, 0)::float8 >= sqlc.arg('fail_ratio_threshold')::float8
|
|
ORDER BY s.failed DESC, a.id ASC;
|
|
|
|
-- name: SystemPauseAutopilot :one
|
|
-- Atomically pauses an autopilot only if it is currently active. Returns no
|
|
-- rows when the autopilot was already paused/archived (or another worker
|
|
-- raced first), letting the caller treat that as a benign no-op rather than
|
|
-- an error.
|
|
UPDATE autopilot
|
|
SET status = 'paused', updated_at = now()
|
|
WHERE id = $1 AND status = 'active'
|
|
RETURNING *;
|