feat(linear): add flyteplugins-linear

Receive Linear webhooks in Flyte. Exports LinearProvider, a Provider subclass with its
defaults pre-wired, so wiring it up reads:

    WebhookAppEnvironment(providers=[LinearProvider()])

The receiver itself ships with flyte, at flyte.extras.webhooks; this package
contributes only what is specific to Linear -- which environment variable holds
the secret, how to verify a delivery, how to parse one into a WebhookEvent, and
typed constants for every event Linear sends.

examples/linear_webhooks.py runs with no Linear account at all: --local replays
this plugin's own SAMPLE_DELIVERY through the app, so you can watch a delivery
be verified, normalized, and dispatched before wiring anything up. The shared
conformance check exercises that same sample.

The package is named for the product rather than for webhooks, so client
methods can land here later when they earn their place -- returning
flyte.io.File instead of an inline megabyte payload, rendering into the task
report, or participating in caching and fan-out. Plain API passthrough belongs
in the vendor's own SDK, called directly from a task.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VKZrTNjjWVzTZUxDFbn4Nk
Signed-off-by: Niels Bantilan <niels.bantilan@gmail.com>

Author: cosmicBboy

plugins/linear/README.md

@@ -0,0 +1,83 @@
+# flyteplugins-linear
+
+Receive Linear webhooks in Flyte.
+
+```bash
+pip install "flyteplugins-linear[app]"
+```
+
+## Using it
+
+Hand a `LinearProvider()` to a `WebhookAppEnvironment` and register handlers with the
+typed constants in `events`:
+
+```python
+import flyte
+from flyte.extras.webhooks import DuplicateRun, WebhookAppEnvironment, idempotent_run
+from flyteplugins.linear import LinearProvider, events
+
+app_env = WebhookAppEnvironment(
+    name="linear-webhooks",
+    providers=[LinearProvider()],
+    secrets=[flyte.Secret("LINEAR_WEBHOOK_SECRET", as_env_var="LINEAR_WEBHOOK_SECRET")],
+)
+
+
+@app_env.on_event(events.Issue.CREATE)
+async def handle(event):
+    import flyte.remote as remote
+
+    task = remote.Task.get(name="my-env.my_task", auto_version="latest")
+    try:
+        run = await idempotent_run.aio(task, key=event.dedupe_key(), resource=event.resource_id)
+    except DuplicateRun as exc:
+        return {"skipped": str(exc)}
+    return {"run": run.name}
+
+
+flyte.serve(app_env)
+```
+
+Handlers must `await idempotent_run.aio(...)`. The blocking form stalls the
+app's event loop, and Linear times deliveries out in seconds.
+
+One app can serve several products at once — hand it one provider per product.
+
+## Try it
+
+`examples/linear_webhooks.py` runs two ways. The first needs no Linear account:
+
+```bash
+python examples/linear_webhooks.py --local   # replay a real sample delivery in-process
+python examples/linear_webhooks.py           # deploy the receiver to Flyte
+```
+
+`--local` posts this plugin's `SAMPLE_DELIVERY` through the app with FastAPI's
+test client, so you see a delivery verified, normalized, and dispatched — plus
+an unsigned one refused with a 401, and the same delivery replayed to show the
+dedupe key is stable.
+
+## Setup
+
+1. Store the secret and mount it on the app:
+   ```bash
+   flyte create secret LINEAR_WEBHOOK_SECRET --value <secret>
+   ```
+2. Point Linear at `<app-url>/webhook/linear`, from
+   Linear Settings → API → Webhooks (it shows the signing secret on creation).
+
+**Verification:** HMAC-SHA256 over the raw body (`X-Linear-Signature`).
+
+Comment and reaction payloads carry the team id only on the nested issue; the parser follows it, so a `scopes` allowlist can still attribute them.
+
+## Event constants
+
+`events` spells every event this plugin can dispatch, as `str` enums grouped by
+event type, so a typo fails at import rather than by silently never matching.
+Raw strings still work, for events the constants do not cover yet.
+
+## What this plugin does not do
+
+Call the Linear API. Use `gql` — Linear ships no Python SDK, and its API is a single GraphQL endpoint directly from your tasks — see
+`examples/external_saas_integrations`. This plugin owns only the part that is
+Flyte's: authenticating an inbound delivery and turning it into a run.

plugins/linear/examples/linear_webhooks.py

@@ -0,0 +1,121 @@
+"""Receive Linear webhooks in Flyte, and see one arrive without leaving your laptop.
+
+Two ways to run this. The second needs no Linear account at all:
+
+    python linear_webhooks.py --local   # replay a real sample delivery in-process
+    python linear_webhooks.py           # deploy the receiver to Flyte
+
+`--local` runs the app through FastAPI's test client and posts this plugin's
+`SAMPLE_DELIVERY` — an `Issue.create` delivery — signed with a throwaway secret. You see the
+delivery verified, normalized, and dispatched to a handler, which is the whole
+path a real webhook takes.
+
+To receive real events, deploy it and point Linear at `<app-url>/webhook/linear`
+from Linear Settings -> API -> Webhooks.
+
+Setup for the real thing:
+    flyte create secret LINEAR_WEBHOOK_SECRET --value <secret>
+
+Linear shows the signing secret once, when you create the webhook.
+"""
+
+import os
+import sys
+
+import flyte
+from flyte.extras.webhooks import WebhookAppEnvironment
+
+from flyteplugins.linear import DEFAULT_SECRET_ENV, SAMPLE_DELIVERY, LinearProvider, events
+
+image = flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages("flyteplugins-linear[app]")
+
+app_env = WebhookAppEnvironment(
+    name="linear-webhooks",
+    providers=[LinearProvider()],
+    image=image,
+    secrets=[flyte.Secret(DEFAULT_SECRET_ENV, as_env_var=DEFAULT_SECRET_ENV)],
+)
+
+
+@app_env.on_event(events.Issue.CREATE)
+async def on_primary(event):
+    """React to the event this plugin's sample delivery carries.
+
+    Returning a dict is enough to see the path working. To do real work, launch
+    a deployed task instead — see `launch_a_task` below.
+    """
+    return {
+        "saw": event.qualified_type,
+        "resource": event.resource_id,
+        "title": event.title,
+        # The key `idempotent_run` would dedupe on. Replaying the same delivery
+        # produces the same key, which is what makes a redelivery a no-op.
+        "dedupe_key": event.dedupe_key(),
+    }
+
+
+@app_env.on_event(events.Issue.UPDATE)
+async def on_secondary(event):
+    """A second handler, to show dispatch picking the right one per event."""
+    return {"saw": event.qualified_type, "resource": event.resource_id}
+
+
+async def launch_a_task(event):
+    """What a handler looks like once it does real work.
+
+    Not registered above, because it needs `linear-triage.triage_issue` deployed first
+    and a Flyte backend to launch into. Wire it up with:
+
+        @app_env.on_event(events.Issue.CREATE)
+
+    `idempotent_run` refuses to launch when a run carrying the same dedupe key
+    is already live or has succeeded, so Linear redelivering an event — which
+    it does on any non-2xx — never starts a second run.
+    """
+    import flyte.remote as remote
+    from flyte.extras.webhooks import DuplicateRun, idempotent_run
+
+    task = remote.Task.get(name="linear-triage.triage_issue", auto_version="latest")
+    try:
+        # Always `.aio`: the blocking form stalls the app's event loop, and
+        # webhook senders time deliveries out in seconds.
+        run = await idempotent_run.aio(task, key=event.dedupe_key(), issue_id=event.resource_id)
+    except DuplicateRun as exc:
+        return {"skipped": str(exc)}
+    return {"run": run.name}
+
+
+def _try_locally() -> None:
+    """Post this plugin's sample delivery to the app, in-process."""
+    from fastapi.testclient import TestClient
+
+    secret = os.environ.setdefault(DEFAULT_SECRET_ENV, "local-trial-secret")
+    build_headers, body = SAMPLE_DELIVERY
+    client = TestClient(app_env.app)
+
+    print("POST /webhook/linear  (signed with a throwaway secret)")
+    response = client.post("/webhook/linear", content=body, headers=build_headers(body, secret))
+    print(f"  {response.status_code}  {response.json()}\n")
+
+    print("the same delivery again — note the identical dedupe_key:")
+    again = client.post("/webhook/linear", content=body, headers=build_headers(body, secret))
+    print(f"  {again.status_code}  {again.json()}\n")
+
+    print("an unsigned delivery is refused:")
+    bad = client.post("/webhook/linear", content=body, headers={})
+    print(f"  {bad.status_code}  {bad.json()}\n")
+
+    print("normalized events the app has seen:")
+    for seen in client.get("/api/events").json():
+        print(f"  {seen['provider']}  {seen['qualified_type']}  resource={seen['resource_id']}")
+
+
+if __name__ == "__main__":
+    if "--local" in sys.argv:
+        _try_locally()
+    else:
+        flyte.init_from_config()
+        handle = flyte.serve(app_env)
+        handle.activate(wait=True)
+        print(f"Dashboard ready at {handle.endpoint}")
+        print(f"Point Linear at {handle.endpoint}/webhook/linear")

plugins/linear/pyproject.toml

@@ -0,0 +1,59 @@
+[project]
+name = "flyteplugins-linear"
+dynamic = ["version"]
+description = "Receive Linear webhooks in Flyte."
+readme = "README.md"
+authors = [{ name = "Flyte Contributors" }]
+requires-python = ">=3.10"
+# The webhook receiver lives in flyte itself, at flyte.extras.webhooks.
+dependencies = ["flyte"]
+
+[project.optional-dependencies]
+app = ["fastapi>=0.115", "uvicorn>=0.30"]
+
+[build-system]
+requires = ["setuptools", "setuptools_scm"]
+build-backend = "setuptools.build_meta"
+
+[dependency-groups]
+dev = [
+    "pytest>=8.3.5",
+    "pytest-asyncio>=0.26.0",
+    "fastapi>=0.115",
+    "uvicorn>=0.30",
+    "httpx>=0.27",
+]
+
+[tool.setuptools]
+include-package-data = true
+
+[tool.setuptools.packages.find]
+where = ["src"]
+include = ["flyteplugins*"]
+
+[tool.setuptools_scm]
+root = "../../"
+
+[tool.pytest.ini_options]
+norecursedirs = []
+log_cli = true
+log_cli_level = 20
+markers = []
+asyncio_mode = "auto"
+asyncio_default_fixture_loop_scope = "function"
+
+[tool.coverage.run]
+branch = true
+
+[tool.ruff]
+line-length = 120
+
+[tool.ruff.lint]
+select = ["E", "W", "F", "I", "PLW", "YTT", "ASYNC", "C4", "T10", "EXE", "ISC", "LOG", "PIE", "Q", "RSE", "FLY", "PGH", "PLC", "PLE", "FURB", "RUF"]
+ignore = ["PGH003", "PLC0415"]
+
+[tool.ruff.lint.per-file-ignores]
+"examples/*" = ["E402"]
+
+[tool.uv.sources]
+flyte = { path = "../../", editable = true }

plugins/linear/src/flyteplugins/linear/__init__.py

@@ -0,0 +1,32 @@
+"""Linear webhooks for Flyte.
+
+Hand a `LinearProvider()` to a `WebhookAppEnvironment` and register handlers with the
+typed constants in `events`. Calling the Linear API is not this plugin's job —
+Linear's API is a single GraphQL endpoint, so use `gql` from your tasks. See
+`examples/external_saas_integrations`.
+"""
+
+import hashlib
+import hmac
+
+from . import events
+from ._provider import DEFAULT_SECRET_ENV, LinearProvider, parse, verify
+
+__all__ = ["DEFAULT_SECRET_ENV", "SAMPLE_DELIVERY", "LinearProvider", "events", "parse", "verify"]
+
+
+def _sample_headers(body: bytes, secret: str) -> dict[str, str]:
+    return {"X-Linear-Signature": hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()}
+
+
+#: A real `Issue.create` delivery, trimmed to the fields the parser reads.
+SAMPLE_DELIVERY = (
+    _sample_headers,
+    (
+        b'{"action": "create", "type": "Issue", "webhookId": "wh-000",'
+        b' "createdAt": "2024-01-01T00:00:00.000Z",'
+        b' "data": {"id": "00000000-0000-0000-0000-000000000000", "title": "A bug",'
+        b' "teamId": "team-000", "updatedAt": "2024-01-01T00:00:00.000Z",'
+        b' "url": "https://linear.app/acme/issue/ENG-1"}}'
+    ),
+)

plugins/linear/src/flyteplugins/linear/_provider.py

@@ -0,0 +1,90 @@
+"""Linear webhook verification and payload normalization."""
+
+from __future__ import annotations
+
+from typing import Any, Mapping
+
+from flyte.extras.webhooks import (
+    Provider,
+    WebhookEvent,
+    constant_time_equals,
+    hex_hmac_sha256,
+    json_body,
+    lower_headers,
+)
+
+#: Environment variable this provider reads its secret from by default.
+DEFAULT_SECRET_ENV = "LINEAR_WEBHOOK_SECRET"
+
+
+def verify(body: bytes, headers: Mapping[str, str], secret: str) -> bool:
+    """Verify the `X-Linear-Signature` HMAC over the raw body."""
+    signature = lower_headers(headers).get("x-linear-signature")
+    if not signature:
+        return False
+    return constant_time_equals(hex_hmac_sha256(secret, body), signature.strip())
+
+
+def _team_id(data: dict[str, Any]) -> str | None:
+    """Find the team id, which Comment and Reaction payloads nest on the issue.
+
+    Without these fallbacks a `scopes` allowlist drops every non-Issue event as
+    unattributable.
+    """
+    issue = data.get("issue") or {}
+    for candidate in (
+        data.get("teamId"),
+        (data.get("team") or {}).get("id"),
+        issue.get("teamId"),
+        (issue.get("team") or {}).get("id"),
+    ):
+        if candidate:
+            return str(candidate)
+    return None
+
+
+def parse(headers: Mapping[str, str], body: bytes) -> WebhookEvent:
+    """Normalize a Linear delivery into a `WebhookEvent`."""
+    payload = json_body(body)
+    data = payload.get("data") or {}
+    return WebhookEvent(
+        provider="linear",
+        event_type=payload.get("type", "Unknown"),
+        action=payload.get("action", "unknown"),
+        delivery_id=str(payload.get("webhookId") or ""),
+        resource_id=data.get("id"),
+        # `updatedAt` is on the entity; `createdAt` is the delivery time and the
+        # only timestamp on payloads whose entity carries none.
+        occurred_at=data.get("updatedAt") or payload.get("createdAt"),
+        scope=_team_id(data),
+        title=data.get("title"),
+        url=data.get("url") or payload.get("url"),
+        actor=(data.get("creator") or {}).get("name"),
+        payload=payload,
+    )
+
+
+class LinearProvider(Provider):
+    """Linear's webhook provider, with its defaults pre-wired.
+
+    ```python
+    from flyte.extras.webhooks import WebhookAppEnvironment
+    from flyteplugins.linear import LinearProvider
+
+    app_env = WebhookAppEnvironment(name="webhooks", providers=[LinearProvider()])
+    ```
+
+    Args:
+        secret_env: Environment variable holding the secret, mounted from a
+            `flyte.Secret`. Override only if you store it under a non-standard
+            name; the default is what the docs and examples assume.
+    """
+
+    def __init__(self, *, secret_env: str = DEFAULT_SECRET_ENV) -> None:
+        super().__init__(
+            name="linear",
+            secret_env=secret_env,
+            verify=verify,
+            parse=parse,
+            setup_hint="Linear Settings -> API -> Webhooks",
+        )