diff --git a/agentex/openapi.yaml b/agentex/openapi.yaml index aeff08d3..501c3631 100644 --- a/agentex/openapi.yaml +++ b/agentex/openapi.yaml @@ -2985,6 +2985,21 @@ paths: application/json: schema: $ref: '#/components/schemas/HTTPValidationError' + /linear/events: + post: + tags: + - Linear + summary: Linear agent webhook ingress for the @agentex app + operationId: linear_events_linear_events_post + responses: + '200': + description: Successful Response + content: + application/json: + schema: + additionalProperties: true + type: object + title: Response Linear Events Linear Events Post /deployment-history/{deployment_id}: get: tags: diff --git a/agentex/src/api/app.py b/agentex/src/api/app.py index 6366c0a7..9fc26ee1 100644 --- a/agentex/src/api/app.py +++ b/agentex/src/api/app.py @@ -37,6 +37,7 @@ deployment_history, deployments, events, + linear, messages, slack, spans, @@ -204,7 +205,7 @@ async def handle_unexpected(request, exc): fastapi_app.include_router(slack.router) fastapi_app.include_router(agent_task_tracker.router) fastapi_app.include_router(agent_api_keys.router) -fastapi_app.include_router(slack.router) +fastapi_app.include_router(linear.router) fastapi_app.include_router(deployment_history.router) fastapi_app.include_router(deployments.router) # Agent run schedules are feature-flagged (off by default, enabled in development). diff --git a/agentex/src/api/middleware_utils.py b/agentex/src/api/middleware_utils.py index af1abd9f..8f82ac75 100644 --- a/agentex/src/api/middleware_utils.py +++ b/agentex/src/api/middleware_utils.py @@ -17,9 +17,8 @@ WHITELISTED_ROUTES: set[str] = { "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/agents/register", "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/agents/forward", - # Slack can't present an SGP principal; the Slack signature is the auth, - # verified inside SlackGatewayUseCase against the app's signing secret. "/slack", + "/linear", "/docs", "/api", "/openapi.json", diff --git a/agentex/src/api/routes/linear.py b/agentex/src/api/routes/linear.py new file mode 100644 index 00000000..b32f2cf9 --- /dev/null +++ b/agentex/src/api/routes/linear.py @@ -0,0 +1,28 @@ +"""Linear gateway ingress — POST /linear/events. + +The webhook URL for the ``@agentex`` Linear agent app. Auth-whitelisted (like +/slack) because Linear can't present an SGP principal; the Linear-Signature is the +auth, verified in the use case against the app's webhook signing secret. Delegates +all logic to LinearGatewayUseCase. +""" + +from fastapi import APIRouter, BackgroundTasks, Request + +from src.domain.use_cases.linear_gateway_use_case import DLinearGatewayUseCase + +router = APIRouter(prefix="/linear", tags=["Linear"]) + + +@router.post("/events", summary="Linear agent webhook ingress for the @agentex app") +async def linear_events( + request: Request, + background: BackgroundTasks, + use_case: DLinearGatewayUseCase, +) -> dict: + # Read the raw body first (needed for HMAC signature verification), then parse. + body = await request.body() + payload = await request.json() + headers = {k.lower(): v for k, v in request.headers.items()} + return await use_case.handle_linear_event( + body=body, headers=headers, payload=payload, background=background + ) diff --git a/agentex/src/domain/use_cases/linear_gateway_use_case.py b/agentex/src/domain/use_cases/linear_gateway_use_case.py new file mode 100644 index 00000000..8308cbaf --- /dev/null +++ b/agentex/src/domain/use_cases/linear_gateway_use_case.py @@ -0,0 +1,714 @@ +"""Linear gateway — a platform-side ingress that fronts one Linear agent app +(``@agentex``) and routes each turn to the resolved agent runtime. + +Direct analog of the Slack gateway (``slack_gateway_use_case``): an external trigger +that does ``task/create`` (get-or-create by name) then ``event/send`` against an +arbitrary agent via ``AgentsACPUseCase.handle_rpc_request``, with an idempotency +marker so provider retries don't double-deliver. Here the trigger is a Linear +``AgentSessionEvent`` (the app was @mentioned / assigned) instead of a Slack event. + +Differences from Slack, all isolated to ingress/normalize/deliver: + - Inbound auth is Linear's ``Linear-Signature`` (hex HMAC-SHA256 over the raw body) + plus a ``webhookTimestamp`` freshness check. No url_verification handshake. + - Dedup is on the ``Linear-Delivery`` UUID. + - The reply is a Linear **agent activity** (``agentActivityCreate``): a ``thought`` + emitted immediately (Linear marks a session unresponsive without an activity + within ~10s), then a ``response`` (terminal) when the turn settles — instead of + a Slack ``chat.postMessage``. + - The Linear API token is minted via the OAuth **client_credentials** grant with + ``actor=app`` (the bot's own identity) and re-minted on a 401 — so no perishable + token is stored, only the static client id/secret. + +Identity: every Linear turn acts as the gateway's own SGP identity — a dedicated bot +service account (``LINEAR_GATEWAY_ACTING_BOT_API_KEY`` + ``LINEAR_GATEWAY_ACCOUNT_ID``, +env / k8s-secret only), forwarded as ``x-api-key`` and converted downstream to +``x-acting-user-api-key``. The bot is a first-class entity, not a proxy for the +invoking Linear user; its tasks are owned by it. Deliberately NOT per-user. +""" + +from __future__ import annotations + +import asyncio +import hashlib +import hmac +import os +import time +from dataclasses import dataclass +from typing import Annotated, Any + +import httpx +from fastapi import BackgroundTasks, Depends + +from src.adapters.crud_store.exceptions import DuplicateItemError, ItemDoesNotExist +from src.config.dependencies import ( + GlobalDependencies, + database_async_read_only_session_maker, + database_async_read_write_engine, + database_async_read_write_session_maker, +) +from src.domain.entities.agents import ACPType +from src.domain.entities.agents_rpc import ( + AgentRPCMethod, + CreateTaskRequestEntity, + SendEventRequestEntity, + SendMessageRequestEntity, +) +from src.domain.entities.task_messages import ( + MessageAuthor, + TextContentEntity, + TextFormat, +) +from src.domain.repositories.agent_repository import AgentRepository +from src.utils.logging import make_logger + +logger = make_logger(__name__) + +# The registered golden agent is "golden-agent" (hyphenated), same as the Slack path. +_DEFAULT_AGENT_NAME = "golden-agent" + +# Every Linear turn acts as the gateway's own SGP identity — a dedicated bot service +# account. Forwarded as x-api-key (+ x-selected-account-id); converted downstream to +# x-acting-user-api-key so the agent's tools act as the bot. Env / k8s-secret only. +_ACTING_BOT_API_KEY = os.getenv("LINEAR_GATEWAY_ACTING_BOT_API_KEY", "") +_ACTING_ACCOUNT_ID = os.getenv("LINEAR_GATEWAY_ACCOUNT_ID", "") + +# DEV ONLY. Skip Linear-Signature verification so a hand-crafted payload can be POSTed +# to /linear/events without a real signature. Never enable outside a dev environment — +# it lets anyone invoke agents through the gateway unauthenticated. +_DEV_SKIP_VERIFY = os.getenv("LINEAR_GATEWAY_DEV_SKIP_VERIFY", "").lower() in ( + "1", + "true", + "yes", +) + +# Linear webhook replay guard: reject a delivery whose ``webhookTimestamp`` (ms) is +# more than this far from now. Linear's guidance is 60s. +_WEBHOOK_MAX_AGE_MS = int(os.getenv("LINEAR_WEBHOOK_MAX_AGE_MS", "60000")) + +_MESSAGE_PAGE = 200 # per-poll page size when collecting the reply + +# Linear webhooks are at-least-once — dedup on the ``Linear-Delivery`` UUID via Redis +# with a short TTL so a retry can't start a duplicate turn. +_EVENT_DEDUP_TTL_SECONDS = int(os.getenv("LINEAR_EVENT_DEDUP_TTL", "600")) + +# Create-race handling (two concurrent first turns for one session): the loser's +# get-or-create raises DuplicateItemError; retry with a short backoff so a lagging +# read replica catches up. Mirrors the Slack gateway. +_CREATE_RACE_ATTEMPTS = max(1, int(os.getenv("LINEAR_CREATE_RACE_ATTEMPTS", "4"))) +_CREATE_RACE_BACKOFF_S = float(os.getenv("LINEAR_CREATE_RACE_BACKOFF_S", "0.25")) + +# Selector cascade default: golden-agent + this agent_config id when nothing else +# matches. Same shape as the Slack gateway. +_DEFAULT_CONFIG_ID = os.getenv( + "LINEAR_GATEWAY_DEFAULT_CONFIG_ID", "416f61d9-9587-46be-a1d8-0b0aba17eb6e" +) +# SGP API base for name -> config_id resolution. Empty (or no acting key) -> skipped. +_SGP_BASE_URL = os.getenv("LINEAR_GATEWAY_SGP_BASE_URL", "").rstrip("/") +_CONFIG_ID_CACHE: dict[tuple[str, str], str] = {} + +# Linear OAuth app (client_credentials, actor=app) — the gateway mints the app token +# from these to call the Linear API, and re-mints on 401. Static; never expire. +_CLIENT_ID = os.getenv("LINEAR_CLIENT_ID", "") +_CLIENT_SECRET = os.getenv("LINEAR_CLIENT_SECRET", "") +_WEBHOOK_SIGNING_SECRET = os.getenv("LINEAR_WEBHOOK_SIGNING_SECRET", "") + +_LINEAR_API_BASE = os.getenv("LINEAR_API_BASE", "https://api.linear.app").rstrip("/") +_LINEAR_SCOPES = "read,write,app:assignable,app:mentionable" +# Process-lifetime cache of the minted app token; cleared + re-minted on a 401. +_APP_TOKEN: dict[str, str] = {} + +_AGENT_ACTIVITY_CREATE = """ +mutation AgentActivityCreate($input: AgentActivityCreateInput!) { + agentActivityCreate(input: $input) { success } +} +""".strip() + + +# --------------------------------------------------------------------------- shaping + + +@dataclass +class InboundLinear: + session_id: str # AgentSession id — the conversation / task key + actor: str # who triggered it (for attribution) + text: str # the prompt text + selector: str | None # first token (candidate agent_config / agent name) + issue_id: str # issue context for the agent's Linear tools (may be "") + action: str # "created" | "prompted" + + +def _extract_prompt(payload: dict[str, Any], action: str) -> str: + """Pull the user's prompt out of an AgentSessionEvent. + + On ``prompted`` the follow-up message is in ``agentActivity.body``. On ``created`` + the trigger is a mention/delegation, so prefer the triggering comment body, then + the issue title/description, then the formatted ``promptContext``.""" + if action == "prompted": + return ((payload.get("agentActivity") or {}).get("body") or "").strip() + session = payload.get("agentSession") or {} + comment = session.get("comment") or {} + if comment.get("body"): + return str(comment["body"]).strip() + issue = session.get("issue") or {} + title, desc = issue.get("title") or "", issue.get("description") or "" + if title or desc: + return (f"{title}\n\n{desc}").strip() + return str( + session.get("promptContext") or payload.get("promptContext") or "" + ).strip() + + +def normalize(payload: dict[str, Any]) -> InboundLinear | None: + """Shape an ``AgentSessionEvent``. Returns None for events we ignore.""" + if payload.get("type") != "AgentSessionEvent": + return None + action = payload.get("action") or "" + if action not in ("created", "prompted"): + return None + session = payload.get("agentSession") or {} + session_id = session.get("id") or payload.get("agentSessionId") or "" + if not session_id: + return None + text = _extract_prompt(payload, action) + issue = session.get("issue") or {} + return InboundLinear( + session_id=session_id, + actor=(session.get("creator") or {}).get("name") + or (session.get("actor") or {}).get("name") + or "", + text=text, + selector=text.split(maxsplit=1)[0] if text else None, + issue_id=issue.get("id") or "", + action=action, + ) + + +def verify_signature( + signing_secret: str, signature: str, body: bytes, webhook_timestamp: Any +) -> bool: + """Linear webhook auth: hex HMAC-SHA256 over the RAW body with the webhook signing + secret, plus a ``webhookTimestamp`` (ms) freshness guard to prevent replay. + + Fail CLOSED on an empty signing secret: an empty HMAC key is publicly known, so + accepting it would let anyone forge a valid signature and drive agent turns as the + bot. An unconfigured secret must reject every delivery, not authenticate all of them.""" + if not signing_secret: + return False + try: + if abs(time.time() * 1000 - int(webhook_timestamp)) > _WEBHOOK_MAX_AGE_MS: + return False + except (TypeError, ValueError): + return False + expected = hmac.new(signing_secret.encode(), body, hashlib.sha256).hexdigest() + return hmac.compare_digest(expected, signature or "") + + +@dataclass +class Target: + agent_name: str + config_id: str | None = None + + def label(self) -> str: + return self.agent_name + + +def _strip_selector(text: str, selector: str) -> str: + """Drop the leading selector token once it has matched a target.""" + stripped = text.strip() + if stripped.lower().startswith(selector.lower()): + stripped = stripped[len(selector) :].lstrip() + return stripped + + +def _turn_content(inbound: InboundLinear, prompt: str) -> str: + """Prepend Linear context so the agent's Linear tools have the issue/session to act + on. Harmless when no Linear tool is enabled.""" + context = ( + f"[Linear context] session_id={inbound.session_id} issue_id={inbound.issue_id}. " + f"This message came from that Linear agent session; use your Linear tools with " + f"this issue_id to read or update the issue." + ) + return f"{context}\n\n{prompt}" + + +def _agent_text(messages) -> str | None: + """Join agent-authored text content from a list of task messages.""" + parts = [] + for m in messages: + content = getattr(m, "content", None) + if ( + content is not None + and getattr(content, "author", None) == MessageAuthor.AGENT + and isinstance(getattr(content, "content", None), str) + ): + text = content.content.strip() + if text: + parts.append(text) + return "\n\n".join(parts) if parts else None + + +# --------------------------------------------------------------------------- use case + + +class LinearGatewayUseCase: + """Runs every Linear turn as one dedicated bot identity (the API key above). No + constructor deps — the ACP use case is built per-turn, bound to the resolved + principal + delegation headers, in ``_dispatch``.""" + + async def handle_linear_event( + self, + *, + body: bytes, + headers: dict[str, str], + payload: dict[str, Any], + background: BackgroundTasks, + ) -> dict: + # 1. Verify the signature (unless dev-skip). No url_verification handshake. + if _DEV_SKIP_VERIFY: + logger.warning( + "LINEAR_GATEWAY_DEV_SKIP_VERIFY is ON — skipping signature verification. " + "DEV ONLY; the gateway is unauthenticated in this mode." + ) + elif not verify_signature( + _WEBHOOK_SIGNING_SECRET, + headers.get("linear-signature", ""), + body, + payload.get("webhookTimestamp"), + ): + logger.warning("linear signature verification failed") + return {"ok": False} # 200 to Linear, but drop it + + inbound = normalize(payload) + if inbound is None: + return {"ok": True} # event we don't act on + + # 2. Dedup Linear's at-least-once delivery on the Linear-Delivery UUID. + delivery_id = headers.get("linear-delivery") or payload.get("webhookId") + if await self._already_processed(delivery_id): + logger.info("[linear] duplicate delivery %s skipped", delivery_id) + return {"ok": True} + + # 3. ACK fast; run the turn out-of-band (it emits a `thought` within Linear's + # ~10s window before the longer dispatch). + background.add_task(self._run_turn, inbound) + return {"ok": True} + + async def _already_processed(self, delivery_id: str | None) -> bool: + """Dedup on the Linear-Delivery UUID via a Redis ``SET NX`` with a short TTL. + Returns True when the id was already seen. Fail-open (False) with no id / no + Redis / any error, so dedup can never drop a legitimate first delivery.""" + if not delivery_id: + return False + try: + pool = GlobalDependencies().redis_pool + except Exception: # noqa: BLE001 - deps not initialized (unit test) -> allow + return False + if pool is None: + return False + try: + import redis.asyncio as redis + + client = redis.Redis(connection_pool=pool) + first_time = await client.set( + f"linear:dedup:{delivery_id}", "1", nx=True, ex=_EVENT_DEDUP_TTL_SECONDS + ) + return not first_time + except Exception: # noqa: BLE001 - dedup is best-effort; never block a turn + logger.warning( + "[linear] delivery dedup check failed; processing anyway", exc_info=True + ) + return False + + async def _run_turn(self, inbound: InboundLinear) -> None: + try: + # Emit a `thought` first thing — Linear marks the session unresponsive + # without an activity within ~10s, and this shows the agent is working. + await self._emit(inbound, "thought", "On it…") + + principal, auth_headers = await self._acting_identity() + target, prompt = await self._resolve_target(inbound, auth_headers) + if not await self._authorize(target): + await self._emit( + inbound, "error", f"You're not authorized to run {target.label()}." + ) + return + + reply = await self._dispatch( + target, inbound, prompt, principal, auth_headers + ) + note = f"_via {target.label()}_" + await self._emit( + inbound, + "response", + f"{reply}\n\n{note}" + if reply + else f"_(no reply from {target.label()})_", + ) + except Exception: + logger.exception("linear gateway turn failed") + await self._emit( + inbound, "error", "Something went wrong handling that. Please retry." + ) + + async def _resolve_config_id( + self, name: str, auth_headers: dict[str, str] + ) -> str | None: + """Resolve an agent_config NAME -> id via SGP's directory, authenticated with the + acting identity's headers. Cached by (account, name). Fail-safe -> None (no SGP + base / no acting key / any error) so the caller falls back to the default id.""" + api_key = auth_headers.get("x-api-key") + if not (_SGP_BASE_URL and name and api_key): + return None + cache_key = (auth_headers.get("x-selected-account-id", ""), name) + if cache_key in _CONFIG_ID_CACHE: + return _CONFIG_ID_CACHE[cache_key] + try: + async with httpx.AsyncClient(timeout=10) as client: + resp = await client.get( + f"{_SGP_BASE_URL}/v5/agent_configs", + headers=auth_headers, + params={"name": name}, + ) + items = (resp.json() or {}).get("items") or [] + for cfg in items: + if cfg.get("name") == name and cfg.get("id"): + _CONFIG_ID_CACHE[cache_key] = cfg["id"] + return cfg["id"] + logger.warning("[linear] no agent_config named %r in SGP", name) + except Exception: # noqa: BLE001 - best-effort; fall back to the default id + logger.warning( + "[linear] config-id resolution failed for %r", name, exc_info=True + ) + return None + + async def _message_send_with_race_retry(self, acp, params, agent_id): + """Send a SYNC agent's ``message/send`` (get-or-creates the task, returns the + reply), retrying the whole call on the create-race with a short backoff so a + lagging replica catches up. Each raising attempt fails on the insert before + appending, so retries never duplicate the turn's message.""" + last: DuplicateItemError | None = None + for attempt in range(_CREATE_RACE_ATTEMPTS): + try: + return await acp.handle_rpc_request( + method=AgentRPCMethod.MESSAGE_SEND, params=params, agent_id=agent_id + ) + except DuplicateItemError as exc: + last = exc + if attempt < _CREATE_RACE_ATTEMPTS - 1: + await asyncio.sleep(_CREATE_RACE_BACKOFF_S) + raise last + + async def _resolve_task_after_race(self, task_service, task_name: str): + """Resolve the winner's task by name after an async-path create-race. Retry on + ItemDoesNotExist with backoff until the (possibly-lagging) replica catches up.""" + last: ItemDoesNotExist | None = None + for attempt in range(_CREATE_RACE_ATTEMPTS): + try: + return await task_service.get_task(name=task_name) + except ItemDoesNotExist as exc: + last = exc + if attempt < _CREATE_RACE_ATTEMPTS - 1: + await asyncio.sleep(_CREATE_RACE_BACKOFF_S) + raise last + + async def _dispatch( + self, + target: Target, + inbound: InboundLinear, + prompt: str, + principal: Any, + auth_headers: dict[str, str], + ) -> str | None: + """Create-or-resume a task on the resolved agent, then inject the turn. Keyed on + the Linear agent session (``linear:{session_id}``). ASYNC/AGENTIC agents get a + long-lived workflow (TASK_CREATE first turn, then event/send + poll); SYNC agents + get one message/send. Slack-origin gating in golden-agent auto-enables the + Linear-only tools off ``task_metadata.channel == "linear"`` — the gateway never + requests them.""" + # Local import avoids a domain -> temporal import cycle at module load. + from src.temporal.scheduled_agent_run_factory import ( + build_acp_use_case_for_principal, + ) + + acp = build_acp_use_case_for_principal( + GlobalDependencies(), principal, request_headers=auth_headers + ) + agent = await acp.agent_repository.get(name=target.agent_name) + task_name = f"linear:{inbound.session_id}" + content = TextContentEntity( + author=MessageAuthor.USER, + content=_turn_content(inbound, prompt), + format=TextFormat.MARKDOWN, + ) + # Target (agent + config) is bound at session creation and applied ONLY on the + # first-turn TASK_CREATE below. A Linear agent session is one conversation with + # one target, so a follow-up's leading token is treated as part of the message, + # not a re-route — the session keeps its original agent/config. (Same first-turn + # binding as the Slack gateway; switching target mid-session isn't supported.) + create_params: dict[str, Any] = {} + if target.config_id: + create_params["config_id"] = target.config_id + + if agent.acp_type == ACPType.SYNC: + send = SendMessageRequestEntity( + task_name=task_name, + content=content, + task_params=create_params, + stream=False, + ) + replies = await self._message_send_with_race_retry(acp, send, agent.id) + return _agent_text(replies or []) + + # ASYNC / AGENTIC: TASK_CREATE only on the first turn, then event/send + poll. + task = await self._existing_task(acp.task_service, task_name) + if task is None: + task_metadata = { + "channel": "linear", + "sender_id": target.label(), + "session_id": inbound.session_id, + "issue_id": inbound.issue_id, + } + if target.config_id: + task_metadata["config_id"] = target.config_id + try: + task = await acp.handle_rpc_request( + method=AgentRPCMethod.TASK_CREATE, + params=CreateTaskRequestEntity( + name=task_name, + params=create_params, + task_metadata=task_metadata, + ), + agent_id=agent.id, + ) + except DuplicateItemError: + task = await self._resolve_task_after_race(acp.task_service, task_name) + + seen = await self._seen_message_ids(acp.task_message_service, task.id) + await acp.handle_rpc_request( + method=AgentRPCMethod.EVENT_SEND, + params=SendEventRequestEntity(task_name=task_name, content=content), + agent_id=agent.id, + ) + return await self._collect_reply(acp.task_message_service, task.id, seen) + + async def _existing_task(self, task_service, name: str): + try: + return await task_service.get_task(name=name) + except ItemDoesNotExist: + return None + + async def _recent_messages(self, msg_service, task_id: str) -> list: + """Newest page of task messages in chronological order (fetch DESC so this turn's + reply is always in the window even on a long task, then reverse).""" + msgs = await msg_service.get_messages( + task_id=task_id, limit=_MESSAGE_PAGE, page_number=1, order_direction="desc" + ) + return list(reversed(msgs or [])) + + async def _seen_message_ids(self, msg_service, task_id: str) -> set[str]: + msgs = await self._recent_messages(msg_service, task_id) + return {m.id for m in msgs if getattr(m, "id", None)} + + async def _collect_reply( + self, + msg_service, + task_id: str, + seen: set[str], + *, + timeout_s: float = 120.0, + interval_s: float = 2.0, + quiescence_s: float = 6.0, + ) -> str | None: + """Poll for THIS turn's reply: new agent-authored text that settles (unchanged + for quiescence_s) or times out. Filters on ids not present before the event. + + Interim, same as the Slack gateway: reply attribution is by message-id snapshot, + not event-level correlation, so two prompts racing on the SAME session can each + collect the other's messages (swapped/combined replies). Linear sessions are + sequential in practice (prompt → wait → prompt), so the window is narrow; a + proper fix is event-correlated streaming, tracked as the shared follow-up.""" + waited, last, stable = 0.0, None, 0.0 + while waited < timeout_s: + await asyncio.sleep(interval_s) + waited += interval_s + msgs = await self._recent_messages(msg_service, task_id) + new = [m for m in msgs if getattr(m, "id", None) not in seen] + text = _agent_text(new) + if text and text == last: + stable += interval_s + if stable >= quiescence_s: + return text + elif text: + last, stable = text, 0.0 + return last + + async def _acting_identity(self) -> tuple[Any, dict[str, str]]: + """The gateway's bot identity. Bot API key + account from env / k8s-secret. + Verify the key -> principal, return the credential headers (delegated downstream + as x-acting-user-api-key). FAIL CLOSED when authz is enabled (AGENTEX_AUTH_URL + set) but the key is missing, so a misconfigured deploy never dispatches + unauthenticated. Only the authz-off local case runs with no principal.""" + api_key = _ACTING_BOT_API_KEY + if not api_key: + if os.getenv("AGENTEX_AUTH_URL"): + raise RuntimeError( + "LINEAR_GATEWAY_ACTING_BOT_API_KEY is unset while authz is enabled " + "(AGENTEX_AUTH_URL); refusing to dispatch a Linear turn without a " + "bot principal." + ) + return None, {} + account_id = _ACTING_ACCOUNT_ID + from src.adapters.authentication.adapter_agentex_authn_proxy import ( + AgentexAuthenticationProxy, + ) + from src.config.dependencies import resolve_environment_variable_dependency + from src.config.environment_variables import EnvVarKeys + + headers = {"x-api-key": api_key} + if account_id: + headers["x-selected-account-id"] = account_id + authn = AgentexAuthenticationProxy( + agentex_auth_url=resolve_environment_variable_dependency( + EnvVarKeys.AGENTEX_AUTH_URL + ), + environment=resolve_environment_variable_dependency(EnvVarKeys.ENVIRONMENT), + ) + principal = await authn.verify_headers(headers) + return principal, headers + + async def _resolve_target( + self, inbound: InboundLinear, auth_headers: dict[str, str] + ) -> tuple[Target, str]: + """Selector cascade for the leading token: SGP agent_config name -> golden-agent + + that config; else a registered agentex agent -> that runtime; else golden-agent + + the default config. The selector is stripped only when it matched.""" + selector = inbound.selector + if selector: + config_id = await self._resolve_config_id(selector, auth_headers) + if config_id: + return ( + Target(_DEFAULT_AGENT_NAME, config_id=config_id), + _strip_selector(inbound.text, selector), + ) + if await self._get_agent_by_name(selector) is not None: + return ( + Target(agent_name=selector), + _strip_selector(inbound.text, selector), + ) + return ( + Target(_DEFAULT_AGENT_NAME, config_id=_DEFAULT_CONFIG_ID or None), + inbound.text, + ) + + async def _get_agent_by_name(self, name: str): + """Runtime-registry tier: does an agent with this name exist? Credential-free + read (the directory is infrastructure); access control is enforced at dispatch.""" + engine = database_async_read_write_engine() + repo = AgentRepository( + database_async_read_write_session_maker(engine), + database_async_read_only_session_maker(engine), + ) + try: + return await repo.get(name=name) + except ItemDoesNotExist: + return None + + async def _authorize(self, target: Target) -> bool: + # The acting identity's own grants are enforced by the ACP use case's internal + # authz (agent.execute / task.*). No extra gate here. + return True + + async def _app_token(self, *, force_refresh: bool = False) -> str: + """Mint (and cache) the Linear API token via the OAuth client_credentials grant. + The token is INHERENTLY an app-actor token (the bot's own identity) — do NOT pass + ``actor=app`` here; that param belongs to the authorization_code flow and Linear + rejects/ignores it on client_credentials. Re-minted when ``force_refresh`` (a 401 + from the API means the ~30-day token expired). No perishable token is stored, only + the static client id/secret in env.""" + if not force_refresh and _APP_TOKEN.get("token"): + return _APP_TOKEN["token"] + if not (_CLIENT_ID and _CLIENT_SECRET): + logger.warning("[linear] no client id/secret configured; cannot mint token") + return "" + try: + async with httpx.AsyncClient(timeout=15) as client: + resp = await client.post( + f"{_LINEAR_API_BASE}/oauth/token", + data={ + "grant_type": "client_credentials", + "client_id": _CLIENT_ID, + "client_secret": _CLIENT_SECRET, + "scope": _LINEAR_SCOPES, + }, + ) + token = (resp.json() or {}).get("access_token") or "" + except Exception: # noqa: BLE001 - surfaced as an empty token -> activity no-op + logger.warning("[linear] app-token mint failed", exc_info=True) + return "" + if token: + _APP_TOKEN["token"] = token + return token + + async def _emit( + self, inbound: InboundLinear, activity_type: str, body: str + ) -> None: + """Post an agent activity (``thought`` / ``response`` / ``error``) onto the + session via ``agentActivityCreate``. Mints the app token, and re-mints + retries + once on a 401 (expired client_credentials token). Best-effort: logged, never + fatal to the turn.""" + for attempt in range(2): + token = await self._app_token(force_refresh=attempt == 1) + if not token: + # No token (missing creds or mint failure): we can't reach Linear, so the + # session is left without this activity — a real misconfiguration, hence + # WARNING. Do NOT log the body: for a `response` it's agent output that + # shouldn't leak into the logging pipeline. Length only. + logger.warning( + "[linear] no app token — dropped %s for session %s (%d chars); " + "check LINEAR_CLIENT_ID / LINEAR_CLIENT_SECRET", + activity_type, + inbound.session_id, + len(body), + ) + return + try: + async with httpx.AsyncClient(timeout=15) as client: + resp = await client.post( + f"{_LINEAR_API_BASE}/graphql", + headers={"Authorization": f"Bearer {token}"}, + json={ + "query": _AGENT_ACTIVITY_CREATE, + "variables": { + "input": { + "agentSessionId": inbound.session_id, + "content": {"type": activity_type, "body": body}, + } + }, + }, + ) + except Exception: # noqa: BLE001 - never let delivery break the turn + logger.warning( + "[linear] agentActivityCreate request failed", exc_info=True + ) + return + if resp.status_code == 401 and attempt == 0: + continue # token expired — re-mint and retry once + data = resp.json() if resp.content else {} + if resp.status_code == 200 and not data.get("errors"): + logger.info( + "[linear] emitted %s -> session %s", + activity_type, + inbound.session_id, + ) + else: + logger.warning( + "[linear] agentActivityCreate failed (%s): %s", + resp.status_code, + (data.get("errors") if data else resp.text)[:300] + if data or resp.text + else "", + ) + return + + +DLinearGatewayUseCase = Annotated[LinearGatewayUseCase, Depends(LinearGatewayUseCase)] diff --git a/agentex/tests/unit/use_cases/test_linear_gateway_use_case.py b/agentex/tests/unit/use_cases/test_linear_gateway_use_case.py new file mode 100644 index 00000000..fca73211 --- /dev/null +++ b/agentex/tests/unit/use_cases/test_linear_gateway_use_case.py @@ -0,0 +1,358 @@ +"""Unit tests for the Linear gateway — everything that does NOT require a real Linear +app: signature verification, AgentSessionEvent normalization, the handle_linear_event +control flow (dev-skip / drop / dedup / ack), dispatch (ACP mocked), acting identity, +and agentActivityCreate delivery with client-credentials token minting (httpx mocked). +""" + +import hashlib +import hmac +import time +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest +from fastapi import BackgroundTasks +from src.domain.entities.agents_rpc import AgentRPCMethod +from src.domain.use_cases import linear_gateway_use_case as lg +from src.domain.use_cases.linear_gateway_use_case import ( + LinearGatewayUseCase, + Target, + normalize, + verify_signature, +) + + +def _sig(secret: str, body: bytes) -> str: + return hmac.new(secret.encode(), body, hashlib.sha256).hexdigest() + + +def _now_ms() -> int: + return int(time.time() * 1000) + + +def _created_payload(comment="math-agent please help", session="sess_1"): + return { + "type": "AgentSessionEvent", + "action": "created", + "webhookTimestamp": _now_ms(), + "agentSession": { + "id": session, + "issue": {"id": "iss_1", "title": "Fix the bug", "description": "it broke"}, + "comment": {"body": comment}, + }, + } + + +@pytest.fixture(autouse=True) +def _no_runtime_agents(monkeypatch): + """Default: no selector names a registered agent, so resolution falls to + golden-agent (no DB hit). Non-golden routing tests override this.""" + monkeypatch.setattr( + LinearGatewayUseCase, "_get_agent_by_name", AsyncMock(return_value=None) + ) + + +@pytest.mark.unit +class TestVerifySignature: + def test_valid_signature_passes(self): + secret, body = "shh", b'{"x":1}' + assert verify_signature(secret, _sig(secret, body), body, _now_ms()) is True + + def test_wrong_signature_fails(self): + body = b'{"x":1}' + assert verify_signature("shh", _sig("other", body), body, _now_ms()) is False + + def test_stale_timestamp_fails(self): + secret, body = "shh", b'{"x":1}' + stale = _now_ms() - 10 * 60 * 1000 # 10 min old > 60s guard + assert verify_signature(secret, _sig(secret, body), body, stale) is False + + def test_nonnumeric_timestamp_fails(self): + assert verify_signature("shh", "deadbeef", b"{}", "not-a-number") is False + + def test_empty_secret_fails_closed(self): + # An empty HMAC key is publicly known — even a "correctly" computed signature + # over the empty key must be rejected, or an unconfigured secret authenticates + # every forged delivery. + body = b'{"x":1}' + assert verify_signature("", _sig("", body), body, _now_ms()) is False + + +@pytest.mark.unit +class TestNormalize: + def test_created_extracts_session_prompt_and_selector(self): + inb = normalize(_created_payload()) + assert inb.session_id == "sess_1" + assert inb.issue_id == "iss_1" + assert inb.action == "created" + assert inb.selector == "math-agent" # first token of the comment + assert "please help" in inb.text + + def test_created_falls_back_to_issue_when_no_comment(self): + p = _created_payload() + p["agentSession"].pop("comment") + inb = normalize(p) + assert "Fix the bug" in inb.text and "it broke" in inb.text + + def test_prompted_reads_agent_activity_body(self): + p = { + "type": "AgentSessionEvent", + "action": "prompted", + "agentSession": {"id": "sess_1"}, + "agentActivity": {"body": "any update?"}, + } + inb = normalize(p) + assert inb.action == "prompted" + assert inb.text == "any update?" + + def test_ignores_non_agent_session_events(self): + assert normalize({"type": "Issue", "action": "create"}) is None + + def test_ignores_unhandled_actions(self): + assert normalize({"type": "AgentSessionEvent", "action": "elicited"}) is None + + def test_ignores_missing_session_id(self): + assert ( + normalize( + {"type": "AgentSessionEvent", "action": "created", "agentSession": {}} + ) + is None + ) + + +@pytest.mark.unit +class TestHandleLinearEvent: + @pytest.mark.asyncio + async def test_dev_skip_verify_schedules_turn(self, monkeypatch): + monkeypatch.setattr(lg, "_DEV_SKIP_VERIFY", True) + monkeypatch.setattr( + LinearGatewayUseCase, "_already_processed", AsyncMock(return_value=False) + ) + bg = BackgroundTasks() + out = await LinearGatewayUseCase().handle_linear_event( + body=b"{}", headers={}, payload=_created_payload(), background=bg + ) + assert out == {"ok": True} + assert len(bg.tasks) == 1 # _run_turn scheduled + + @pytest.mark.asyncio + async def test_bad_signature_is_dropped(self, monkeypatch): + monkeypatch.setattr(lg, "_DEV_SKIP_VERIFY", False) + monkeypatch.setattr(lg, "_WEBHOOK_SIGNING_SECRET", "shh") + bg = BackgroundTasks() + out = await LinearGatewayUseCase().handle_linear_event( + body=b'{"x":1}', + headers={"linear-signature": "deadbeef"}, + payload=_created_payload(), + background=bg, + ) + assert out == {"ok": False} + assert len(bg.tasks) == 0 + + @pytest.mark.asyncio + async def test_duplicate_delivery_skipped(self, monkeypatch): + monkeypatch.setattr(lg, "_DEV_SKIP_VERIFY", True) + monkeypatch.setattr( + LinearGatewayUseCase, "_already_processed", AsyncMock(return_value=True) + ) + bg = BackgroundTasks() + out = await LinearGatewayUseCase().handle_linear_event( + body=b"{}", + headers={"linear-delivery": "dup-1"}, + payload=_created_payload(), + background=bg, + ) + assert out == {"ok": True} + assert len(bg.tasks) == 0 # not scheduled + + +@pytest.mark.unit +class TestActingIdentity: + @pytest.mark.asyncio + async def test_no_key_authz_off_is_dev_bypass(self, monkeypatch): + monkeypatch.setattr(lg, "_ACTING_BOT_API_KEY", "") + monkeypatch.delenv("AGENTEX_AUTH_URL", raising=False) + principal, headers = await LinearGatewayUseCase()._acting_identity() + assert principal is None + assert headers == {} + + @pytest.mark.asyncio + async def test_no_key_authz_on_fails_closed(self, monkeypatch): + monkeypatch.setattr(lg, "_ACTING_BOT_API_KEY", "") + monkeypatch.setenv("AGENTEX_AUTH_URL", "http://auth") + with pytest.raises(RuntimeError, match="refusing to dispatch"): + await LinearGatewayUseCase()._acting_identity() + + @pytest.mark.asyncio + async def test_sends_both_headers(self, monkeypatch): + monkeypatch.setattr(lg, "_ACTING_BOT_API_KEY", "ssk_test") + monkeypatch.setattr(lg, "_ACTING_ACCOUNT_ID", "acct_1") + fake_authn = MagicMock() + fake_authn.verify_headers = AsyncMock( + return_value=SimpleNamespace(user_id="u1") + ) + monkeypatch.setattr( + "src.adapters.authentication.adapter_agentex_authn_proxy.AgentexAuthenticationProxy", + MagicMock(return_value=fake_authn), + ) + monkeypatch.setattr( + "src.config.dependencies.resolve_environment_variable_dependency", + lambda _key: "http://auth", + ) + principal, headers = await LinearGatewayUseCase()._acting_identity() + assert headers == {"x-api-key": "ssk_test", "x-selected-account-id": "acct_1"} + assert principal.user_id == "u1" + + +def _fake_acp(existing_task=None, acp_type=None): + """Fake ACP use case (mirrors the Slack test harness). existing_task=None → task + doesn't exist (get_task raises → _dispatch will TASK_CREATE).""" + acp = MagicMock() + acp.agent_repository.get = AsyncMock( + return_value=SimpleNamespace(id="agt_1", acp_type=acp_type or lg.ACPType.ASYNC) + ) + created = SimpleNamespace(id="task_1", task_metadata=None) + acp.handle_rpc_request = AsyncMock(return_value=created) + acp.task_message_service.get_messages = AsyncMock(return_value=[]) + if existing_task is None: + acp.task_service.get_task = AsyncMock( + side_effect=lg.ItemDoesNotExist("no task") + ) + else: + acp.task_service.get_task = AsyncMock(return_value=existing_task) + return acp, created + + +@pytest.mark.unit +class TestDispatch: + @pytest.mark.asyncio + async def test_new_session_creates_task_then_sends_event(self, monkeypatch): + monkeypatch.setattr(lg, "_ACTING_BOT_API_KEY", "") # -> (None, {}) + monkeypatch.setattr(lg, "GlobalDependencies", MagicMock()) + acp, _ = _fake_acp(existing_task=None) + monkeypatch.setattr( + "src.temporal.scheduled_agent_run_factory.build_acp_use_case_for_principal", + MagicMock(return_value=acp), + ) + monkeypatch.setattr( + LinearGatewayUseCase, "_collect_reply", AsyncMock(return_value=None) + ) + inbound = lg.InboundLinear( + session_id="sess_1", + actor="Someone", + text="hello", + selector=None, + issue_id="iss_1", + action="created", + ) + await LinearGatewayUseCase()._dispatch( + Target("golden-agent", config_id=lg._DEFAULT_CONFIG_ID), + inbound, + "hello", + None, + {}, + ) + assert acp.handle_rpc_request.await_count == 2 + first, second = acp.handle_rpc_request.await_args_list + assert first.kwargs["method"] == AgentRPCMethod.TASK_CREATE + assert first.kwargs["params"].name == "linear:sess_1" + assert first.kwargs["params"].task_metadata["channel"] == "linear" + assert first.kwargs["params"].params["config_id"] == lg._DEFAULT_CONFIG_ID + assert second.kwargs["method"] == AgentRPCMethod.EVENT_SEND + sent = second.kwargs["params"].content.content + assert "hello" in sent and "issue_id=iss_1" in sent + + +class _FakeResp: + def __init__(self, status_code=200, json_data=None): + self.status_code = status_code + self._json = json_data or {} + self.content = b"x" + self.text = "" + + def json(self): + return self._json + + +class _FakeClient: + """Records POSTs and returns canned responses keyed on the URL suffix.""" + + def __init__(self, handler): + self._handler = handler + + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + return False + + async def post(self, url, **kw): + return self._handler(url, kw) + + +@pytest.mark.unit +class TestEmitAgentActivity: + @pytest.mark.asyncio + async def test_mints_token_then_posts_activity(self, monkeypatch): + monkeypatch.setattr(lg, "_APP_TOKEN", {}) + monkeypatch.setattr(lg, "_CLIENT_ID", "cid") + monkeypatch.setattr(lg, "_CLIENT_SECRET", "csecret") + calls = [] + + def handler(url, kw): + calls.append(url) + if url.endswith("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/oauth/token"): + return _FakeResp(200, {"access_token": "tok_123"}) + return _FakeResp(200, {"data": {"agentActivityCreate": {"success": True}}}) + + monkeypatch.setattr( + lg.httpx, "AsyncClient", lambda *a, **k: _FakeClient(handler) + ) + inbound = lg.InboundLinear("sess_1", "", "hi", None, "iss_1", "created") + await LinearGatewayUseCase()._emit(inbound, "response", "done") + assert any(u.endswith("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/oauth/token") for u in calls) + assert any(u.endswith("/graphql") for u in calls) + + @pytest.mark.asyncio + async def test_re_mints_on_401(self, monkeypatch): + monkeypatch.setattr(lg, "_APP_TOKEN", {"token": "stale"}) + monkeypatch.setattr(lg, "_CLIENT_ID", "cid") + monkeypatch.setattr(lg, "_CLIENT_SECRET", "csecret") + graphql_calls = {"n": 0} + minted = {"n": 0} + + def handler(url, kw): + if url.endswith("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/oauth/token"): + minted["n"] += 1 + return _FakeResp(200, {"access_token": "fresh"}) + graphql_calls["n"] += 1 + # first graphql call 401 (stale token), second succeeds + if graphql_calls["n"] == 1: + return _FakeResp(401, {}) + return _FakeResp(200, {"data": {"agentActivityCreate": {"success": True}}}) + + monkeypatch.setattr( + lg.httpx, "AsyncClient", lambda *a, **k: _FakeClient(handler) + ) + inbound = lg.InboundLinear("sess_1", "", "hi", None, "iss_1", "created") + await LinearGatewayUseCase()._emit(inbound, "response", "done") + assert graphql_calls["n"] == 2 # retried after 401 + assert minted["n"] == 1 # re-minted once + + @pytest.mark.asyncio + async def test_no_client_creds_is_noop(self, monkeypatch): + monkeypatch.setattr(lg, "_APP_TOKEN", {}) + monkeypatch.setattr(lg, "_CLIENT_ID", "") + monkeypatch.setattr(lg, "_CLIENT_SECRET", "") + called = {"n": 0} + + def handler(url, kw): + called["n"] += 1 + return _FakeResp(200, {}) + + monkeypatch.setattr( + lg.httpx, "AsyncClient", lambda *a, **k: _FakeClient(handler) + ) + inbound = lg.InboundLinear("sess_1", "", "hi", None, "iss_1", "created") + await LinearGatewayUseCase()._emit(inbound, "thought", "on it") + assert called["n"] == 0 # no token → no HTTP call