Deploy Linear Triage to Cloud Run

Deploy a signed Linear webhook receiver with durable Cloud Run workflows.

Introduction

30 Second Summary

An urgent issue can stall when the alert lives in one place while the decisions live somewhere else. A restart at the wrong moment can erase the team's progress.

In this project, you will deploy a signed webhook receiver built with FastAPI to Cloud Run. Durable LangGraph checkpoints in Cloud SQL let Linear approvals resume after the receiver instance disappears.

What You'll Build

You will watch INC-2041 start from the agent-triage label, pause for two Linear approvals, then finish with one Runbook published comment.

By the end of this project, you'll have:

  • A signed receiver with a live /health endpoint. Unsigned webhook requests stop before workflow work begins.
  • A restart-safe approval flow that resumes the matching LangGraph interrupt from Cloud SQL after a Cloud Run instance disappears.
  • An incident timeline with one findings comment, two completed approval sub-issues, and one Runbook published comment.
  • Secret Mission: Re-trigger a completed incident to prove the receiver creates no duplicate findings, approval sub-issues, or runbook comments.

Are there any prerequisites?

This project assumes you completed Triage Incidents with LangGraph, Observe Agent Workflows with Langfuse, and Connect the Triage Workflow to Linear. Your existing Google Cloud project, Linear workspace, Langfuse project, and ~/langgraph-triage-linear clone must remain available.

Before We Start

Before deployment begins, commit to a signed Linear webhook receiver that acknowledges each delivery before asynchronous processing starts. Because Cloud Run instances are disposable, Cloud SQL keeps LangGraph checkpoints available through shutdowns, redeployments, retries, and approval pauses.

Verify the Existing Workflow

The signed webhook receiver must extend the exact LangGraph workflow you already proved. A fresh clone or changed dependency could hide a regression before deployment.

In this step, you will resume the existing project in Google Cloud Shell. You will inspect the inherited Linear workflow before proving the baseline with its existing tests.

In this step, get ready to:
  • Resume the existing project environment in ~/langgraph-triage-linear.
  • Confirm the inherited code boundary remains unchanged.
  • Run the inherited test suite from the existing clone.
Resume the existing Cloud Shell session

The existing clone contains the workflow from your prerequisite projects. Its .venv preserves the proven Python environment without another installation.

  • Return to the Google Cloud console for your existing project.
  • Use the Cloud Shell control in the top toolbar to start the terminal.

You will see the Cloud Shell terminal connected to your existing Google Cloud project.

  • Move into the existing clone by running this command:
cd ~/langgraph-triage-linear

What Does This Command Do?

The command moves your terminal into the clone created during the prerequisite projects. Every check in this step now targets that inherited workflow.

  • Activate the existing .venv with the same activation command used in the prerequisite project.
  • Export the saved .env values with the same export command used in the prerequisite project.
  • Confirm the terminal prompt identifies the active virtual environment.
  • Confirm the export returns to the prompt without displaying credential values.

Your existing environment is active inside ~/langgraph-triage-linear. The saved configuration is available to the inherited workflow.

Environment Not Active?

  • Confirm the terminal is inside ~/langgraph-triage-linear.
  • Confirm the .venv directory still exists inside the clone.
  • Reuse the activation command from the prerequisite project without creating another environment.
  • Reuse the export command from the prerequisite project without printing the contents of .env.
  • Help me restore the existing project environment.
Inspect the inherited code boundary

The current files form the terminal-driven baseline for the receiver. Checking their responsibilities keeps solution-branch features outside this project.

  • Open app.py in the Cloud Shell editor.
  • Locate SqliteSaver.from_conn_string(DB_PATH) inside run_command().
  • Locate the inherited build_graph(checkpointer: SqliteSaver) signature.
  • Locate approval_issue_id inside the approval_gate interrupt.
  • Search app.py for report_failure().

You will find the inherited SQLite checkpoint structures. The search for report_failure() returns no match.

Why Preserve This Baseline?

These checks prove that app.py still matches the workflow from the prerequisite project. Later changes can be compared with a known working starting point.

  • Open requirements.txt in the Cloud Shell editor.
  • Compare its existing dependency pins with the versions below.

Which Pins Should Remain?

  • FastAPI remains pinned to fastapi==0.156.1.
  • Uvicorn remains pinned to uvicorn==0.54.0.
  • LangGraph remains pinned to langgraph==1.2.14.
  • LangGraph Checkpoint Postgres remains pinned to langgraph-checkpoint-postgres==3.2.0.
  • Psycopg remains pinned to psycopg[binary]==3.3.6.
  • Langfuse remains pinned to langfuse==4.17.0.
  • Inspect linear.py to confirm it remains the deterministic Linear GraphQL client.
  • Inspect seed_incidents.py to confirm it remains the incident fixture creator.
  • Inspect linear_webhook.py to confirm it manages the workspace webhook with LINEAR_WEBHOOK_SECRET.
  • Inspect Dockerfile to confirm it copies the repository into the image.
  • Confirm Dockerfile launches receiver:api with Uvicorn on the injected PORT.
  • Leave every inspected file unchanged.

What Does the Deployment Boundary Contain?

The supplied files preserve deterministic Linear operations. They also preserve the incident fixture and workspace webhook utility.

The Dockerfile prepares the repository for Cloud Run. Its entrypoint serves the future receiver on the port injected by the platform.

The solution branch includes features outside this receiver project. Keeping them out protects the starting boundary you just verified.

  • Keep fixer.py out of this project.
  • Keep linear_mcp.py out of this project.
  • Exclude open_fix_pr from this project.
  • Exclude recall_history from this project.
  • Keep GitHub credentials out of this project.
  • Preserve the existing dependencies without reinstalling or upgrading them.
Run the inherited test suite

The inherited suite is the clean checkpoint for this project. It tests the workflow before receiver code changes the execution path.

Before you run the suite, do you expect the unchanged workflow to pass?

  • Run the inherited tests from ~/langgraph-triage-linear with this command:
python3 -m pytest

What Does This Check Prove?

The pytest suite exercises the inherited workflow before receiver code is added. A passing result proves that the existing clone is a reliable starting point.

Baseline secured. You will see the test suite finish with a passing summary and no failed tests.

Inherited Tests Not Passing?

  • Confirm the terminal is still inside ~/langgraph-triage-linear.
  • Confirm the existing .venv remains active.
  • Export the saved .env values again with the command from the prerequisite project.
  • Compare any changed file with its inherited prerequisite version.
  • Avoid changing dependency pins to make a failure disappear.
  • Help me diagnose the failing inherited test suite.

This step changes no files. Use the checkpoint below to compare the inherited project boundary.

✔️ Awesome, I've got everything!

The inherited files remain unchanged. The existing test suite passes from the active project environment.

ⓧ I'd like to double check the full code

  • Dockerfile: Inherited deployment boundary remains unchanged: copies the repository and launches uvicorn receiver:api on the injected Cloud Run port.
  • app.py: Inherited prerequisite workflow file remains unchanged: it uses DB_PATH and SqliteSaver for terminal-driven checkpoints, build_graph(checkpointer: SqliteSaver), approval_gate approval_issue_id interrupts, and no report_failure().
  • linear.py: Inherited deterministic Linear GraphQL client remains unchanged.
  • linear_webhook.py: Inherited workspace webhook create/remove utility remains unchanged and uses LINEAR_WEBHOOK_SECRET.
  • requirements.txt: Inherited unchanged pinned dependency file, including fastapi==0.156.1, uvicorn==0.54.0, langgraph==1.2.14, langgraph-checkpoint-postgres==3.2.0, psycopg[binary]==3.3.6, and langfuse==4.17.0.
  • seed_incidents.py: Inherited Linear incident fixture creator remains unchanged.

Your inherited workflow is verified in its original terminal-driven state. Next, you will build the authenticated receiver shell that acknowledges Linear before slow workflow work begins.

Secure the Webhook Receiver

Your inherited workflow is verified. Incoming traffic from Linear now needs a trusted boundary.

An empty signing secret stops the receiver before authentication. Slow Linear calls or graph calls can cross Linear's five-second response limit.

A signed webhook endpoint in FastAPI authenticates the raw body. The endpoint acknowledges verified deliveries before the temporary background handler runs.

In this step, get ready to:
  • Generate the signing secret for webhook authentication.
  • Build raw-body signature checks plus delivery deduplication.
  • Serve health checks while rejecting unsigned webhook requests.
Generate the webhook signing secret

An HMAC signature lets Linear prove that a request came from your webhook configuration. The receiver calculates its own signature from the same secret.

The generated secret stays inside .env. This keeps the credential out of receiver.py.

  • Generate the signing secret in the active Cloud Shell terminal by running this command:
python3 -c "import secrets; print(secrets.token_hex(32))"

What does this command create?

The command generates a random hexadecimal value for LINEAR_WEBHOOK_SECRET. Linear and the receiver later use this shared value for signature checks.

  • Copy the generated value from the terminal.
  • Open the existing .env file with the Cloud Shell file editor.
  • Replace the value after LINEAR_WEBHOOK_SECRET= with the generated value.
  • Save .env.
  • Confirm .env contains a non-empty LINEAR_WEBHOOK_SECRET entry without revealing the value.
  • Re-run the environment export command from the previous step.

Secret not reaching the receiver?

  • Check that LINEAR_WEBHOOK_SECRET uses that exact spelling in .env.
  • Keep the generated value on the same line as LINEAR_WEBHOOK_SECRET=.
  • Repeat the previous environment export command in the terminal that launches the receiver.
  • Help me check why my receiver cannot read LINEAR_WEBHOOK_SECRET from .env.
Build the authenticated receiver shell

Signature verification must use the request's raw bytes. Parsing or changing the body first could produce a different HMAC value.

  • In the Cloud Shell file editor, create receiver.py inside ~/langgraph-triage-linear.
  • Build the FastAPI application plus raw-body signature check by pasting this code into receiver.py:
import hashlib
import hmac
import json
import os
import threading
import time
from collections import OrderedDict, defaultdict
from contextlib import contextmanager

from fastapi import BackgroundTasks, FastAPI, Header, HTTPException, Request

import linear as linear_api
from linear import APPROVALS_TEAM, INCIDENTS_TEAM, TRIGGER_LABEL

MAX_AGE_MS = 60_000
DECISIONS = {"Approved": "approve", "Rejected": "reject"}

api = FastAPI()
_thread_locks: defaultdict[str, threading.Lock] = defaultdict(threading.Lock)
_locks_guard = threading.Lock()
_seen_deliveries: OrderedDict[str, None] = OrderedDict()


def verify_signature(raw_body: bytes, signature: str | None) -> None:
    secret = os.environ["LINEAR_WEBHOOK_SECRET"].encode()
    expected = hmac.new(secret, raw_body, hashlib.sha256).hexdigest()
    if not signature or not hmac.compare_digest(expected, signature):
        raise HTTPException(status_code=401, detail="Bad signature.")

What does this code do?

  • The imports prepare the receiver for hashing, request handling, delivery tracking, local thread locks, plus later workflow routing.
  • MAX_AGE_MS defines a 60-second replay window for incoming deliveries.
  • _seen_deliveries stores recent delivery IDs for retry detection.
  • verify_signature() calculates an HMAC-SHA256 digest from the raw request body.
  • hmac.compare_digest compares the calculated digest with the incoming signature.
  • HTTPException stops absent or invalid signatures with HTTP 401 plus the detail Bad signature..
  • Save receiver.py.
  • Confirm receiver.py appears inside ~/langgraph-triage-linear in the Cloud Shell file tree.

Receiver file not saving?

  • Check that the file is named receiver.py with lowercase letters.
  • Confirm the file sits directly inside ~/langgraph-triage-linear.
  • Help me find why receiver.py is missing from my project folder.

Linear identifies each delivery with the Linear-Delivery header. A small process-local cache can suppress an immediate retry of the same delivery.

  • Add delivery deduplication plus the temporary background handler by pasting this code below verify_signature():
def first_delivery(delivery_id: str | None) -> bool:
    if not delivery_id:
        return True
    if delivery_id in _seen_deliveries:
        return False
    _seen_deliveries[delivery_id] = None
    while len(_seen_deliveries) > 1000:
        _seen_deliveries.popitem(last=False)
    return True


def handle_delivery(payload: dict) -> None:
    print(json.dumps({"background_payload": payload}), flush=True)

How does retry handling work?

  • first_delivery() returns False when the same delivery ID appears again.
  • The OrderedDict retains at most 1,000 recent IDs.
  • handle_delivery() temporarily prints the parsed payload.
  • The temporary handler performs no Linear lookup or workflow execution.
  • Save receiver.py.
  • Confirm the inherited workflow still passes its tests by running:
python3 -m pytest

What does this check prove?

The passing suite confirms that the new receiver shell has not changed the inherited workflow. Your prerequisite behavior remains intact before the HTTP routes are added.

Tests no longer passing?

  • Confirm you changed only receiver.py in this substep.
  • Compare each import with the code block above.
  • Check the closing parentheses in verify_signature() plus handle_delivery().
  • Help me debug the failing tests after adding the receiver shell.
Return before background work begins

The webhook route verifies the raw body before parsing its payload. It queues handle_delivery() only after authentication plus replay checks succeed.

FastAPI runs the queued task after returning the response. This keeps Linear lookups and graph execution away from the acknowledgement path.

  • Add the webhook endpoint plus health endpoint by pasting this code below handle_delivery():
@api.post("/linear/webhook")
async def linear_webhook(
    request: Request,
    background: BackgroundTasks,
    linear_signature: str | None = Header(default=None),
    linear_delivery: str | None = Header(default=None),
):
    raw_body = await request.body()
    verify_signature(raw_body, linear_signature)
    payload = json.loads(raw_body)
    if abs(time.time() * 1000 - payload.get("webhookTimestamp", 0)) > MAX_AGE_MS:
        raise HTTPException(status_code=401, detail="Stale webhook.")
    print(json.dumps({"delivery": linear_delivery, "payload": payload}), flush=True)
    if not first_delivery(linear_delivery):
        return {"ok": True, "duplicate": True}

    background.add_task(handle_delivery, payload)
    return {"ok": True}


@api.get("/health")
def health():
    return {"ok": True}

What happens on each request?

  • linear_webhook() reads the raw request body before any parsing occurs.
  • verify_signature() rejects unsigned or incorrectly signed traffic first.
  • webhookTimestamp rejects deliveries outside the 60-second replay window with Stale webhook..
  • first_delivery() returns a safe acknowledgement for a repeated delivery ID.
  • background.add_task() schedules only the temporary logging handler.
  • health() provides a small liveness response at /health.
  • Save receiver.py.

Before you start the receiver, what response do you expect from its health endpoint?

  • Start the local receiver on port 8080 by running this command:
uvicorn receiver:api --port 8080

What does this command do?

Uvicorn loads the api object from receiver.py. The process keeps this terminal attached while the receiver listens on port 8080.

  • Keep this terminal tab running during the endpoint checks.
  • Create a second terminal tab in the existing Cloud Shell session.

Receiver not starting?

  • Check that receiver.py is saved inside ~/langgraph-triage-linear.
  • Confirm the first terminal still has the exported LINEAR_WEBHOOK_SECRET value.
  • Compare the reported line number with the matching section of receiver.py.
  • Help me debug why Uvicorn cannot start receiver:api on port 8080.

Before you send the health request, do you expect this route to require a webhook signature?

  • Check the health endpoint from the second terminal by running:
curl http://localhost:8080/health

What should you see?

You should see {"ok":true}. This proves the FastAPI application is running locally.

Health check not responding?

  • Return to the first terminal to confirm the Uvicorn process is still running.
  • Check that the route uses /health.
  • Confirm both commands use port 8080.
  • Help me debug why the local /health endpoint does not return {"ok":true}.

Before you send an unsigned webhook request, do you expect the receiver to parse its body or reject it first?

  • Send an unsigned POST request to the webhook endpoint by running:
curl -i -X POST http://localhost:8080/linear/webhook

What should you see?

You should see HTTP 401 with {"detail":"Bad signature."}. The receiver rejects the request before JSON parsing or any Linear lookup.

Unsigned request not returning 401?

  • Confirm linear_webhook() calls verify_signature() immediately after reading the raw body.
  • Check that verify_signature() raises HTTP 401 with Bad signature..
  • Restart Uvicorn after saving the latest receiver.py changes.
  • Help me debug why an unsigned webhook request is not returning Bad signature.

✔️ Awesome, I've got everything!

Great. Double-check that receiver.py is saved before moving on.

ⓧ I'd like to double check the full code

Compare your complete receiver.py with this checkpoint:

import hashlib
import hmac
import json
import os
import threading
import time
from collections import OrderedDict, defaultdict
from contextlib import contextmanager

from fastapi import BackgroundTasks, FastAPI, Header, HTTPException, Request

import linear as linear_api
from linear import APPROVALS_TEAM, INCIDENTS_TEAM, TRIGGER_LABEL

MAX_AGE_MS = 60_000
DECISIONS = {"Approved": "approve", "Rejected": "reject"}

api = FastAPI()
_thread_locks: defaultdict[str, threading.Lock] = defaultdict(threading.Lock)
_locks_guard = threading.Lock()
_seen_deliveries: OrderedDict[str, None] = OrderedDict()


def verify_signature(raw_body: bytes, signature: str | None) -> None:
    secret = os.environ["LINEAR_WEBHOOK_SECRET"].encode()
    expected = hmac.new(secret, raw_body, hashlib.sha256).hexdigest()
    if not signature or not hmac.compare_digest(expected, signature):
        raise HTTPException(status_code=401, detail="Bad signature.")


def first_delivery(delivery_id: str | None) -> bool:
    if not delivery_id:
        return True
    if delivery_id in _seen_deliveries:
        return False
    _seen_deliveries[delivery_id] = None
    while len(_seen_deliveries) > 1000:
        _seen_deliveries.popitem(last=False)
    return True


def handle_delivery(payload: dict) -> None:
    print(json.dumps({"background_payload": payload}), flush=True)


@api.post("/linear/webhook")
async def linear_webhook(
    request: Request,
    background: BackgroundTasks,
    linear_signature: str | None = Header(default=None),
    linear_delivery: str | None = Header(default=None),
):
    raw_body = await request.body()
    verify_signature(raw_body, linear_signature)
    payload = json.loads(raw_body)
    if abs(time.time() * 1000 - payload.get("webhookTimestamp", 0)) > MAX_AGE_MS:
        raise HTTPException(status_code=401, detail="Stale webhook.")
    print(json.dumps({"delivery": linear_delivery, "payload": payload}), flush=True)
    if not first_delivery(linear_delivery):
        return {"ok": True, "duplicate": True}

    background.add_task(handle_delivery, payload)
    return {"ok": True}


@api.get("/health")
def health():
    return {"ok": True}

That security boundary is working: the receiver answers health checks while rejecting unsigned webhook traffic before any Linear call. Next, you will route authenticated deliveries into durable workflow threads.

Route Deliveries to Durable Threads

Your FastAPI endpoint authenticates Linear deliveries before acknowledging them. Its temporary background handler still stops after printing each payload.

A Cloud Run instance can disappear while a workflow is paused. This step routes each event to its matching LangGraph thread. Durable PostgreSQL checkpoints preserve progress beyond the lifetime of one instance.

In this step, get ready to:
  • Classify incident triggers under local locks plus database-backed locks.
  • Route approval decisions to the matching paused workflow gate.
  • Preserve workflow checkpoints across local runs plus deployed instances.
Classify background deliveries

Two instances can briefly receive events for the same incident. A process-local lock coordinates work inside one instance. A database advisory lock extends that protection across instances.

The database connection uses Psycopg. Closing the connection releases its advisory lock when the thread finishes.

  • Switch back to receiver.py in the Cloud Shell editor.
  • Add this context manager immediately below first_delivery():
@contextmanager
def thread_lock(thread: str):
    with _locks_guard:
        lock = _thread_locks[thread]
    with lock:
        url = os.environ.get("DATABASE_URL")
        if not url:
            yield
            return
        import psycopg

        with psycopg.connect(url, autocommit=True) as conn:
            conn.execute("SELECT pg_advisory_lock(hashtext(%s))", (thread,))
            yield

How Does the Thread Lock Work?

  • The lock stored in _thread_locks prevents overlapping work for one thread inside the current process.
  • The DATABASE_URL check keeps local development on its existing path.
  • The advisory lock uses the incident thread identifier as its key. Every connected instance therefore competes for the same lock.
  • Add this incident classifier immediately below thread_lock():
def new_incident(payload: dict) -> str | None:
    data = payload.get("data") or {}
    if payload.get("type") != "Issue":
        return None
    linear = linear_api.client()
    if data.get("teamId") != linear.team(INCIDENTS_TEAM)["id"]:
        return None
    trigger = linear.label_id(TRIGGER_LABEL)
    if trigger not in (data.get("labelIds") or []):
        return None
    if payload.get("action") == "create":
        return data["id"]
    old = payload.get("updatedFrom") or {}
    if (
        payload.get("action") == "update"
        and "labelIds" in old
        and trigger not in (old["labelIds"] or [])
    ):
        return data["id"]
    return None

Which Incidents Can Start a Run?

  • The function ignores payloads whose type is different from Issue.
  • The team check limits new runs to issues from the Incidents team.
  • A creation event qualifies when the trigger label is present.
  • An update qualifies when the trigger label was absent from the previous label list.
  • Save receiver.py.
  • Compile the first routing changes by running:
python3 -m py_compile receiver.py

What Does This Check Prove?

The command checks the file's Python syntax without starting the receiver. A clean return to the prompt means both new constructs compiled successfully.

Seeing a Syntax Error?

  • Check that each yield remains inside its matching context manager.
  • Check the indentation inside the multi-line condition in new_incident().
  • Use the reported line with help me fix this receiver syntax error.

Incident triggers now have a thread boundary. Approval events need a narrower classifier so ordinary issue updates cannot resume a paused gate.

  • Add this approval classifier immediately below new_incident():
def approval_decision(payload: dict) -> tuple[str, str, str] | None:
    data = payload.get("data") or {}
    old_state = (payload.get("updatedFrom") or {}).get("stateId")
    if payload.get("type") != "Issue" or payload.get("action") != "update" or not old_state:
        return None
    if not data.get("parentId"):
        return None
    linear = linear_api.client()
    if old_state != linear.state_id(APPROVALS_TEAM, "Needs Approval"):
        return None
    for state, decision in DECISIONS.items():
        if data.get("stateId") == linear.state_id(APPROVALS_TEAM, state):
            return data["id"], data["parentId"], decision
    return None

Which Decisions Can Resume a Run?

  • The function accepts only issue updates with a parent issue.
  • The previous state must be Needs Approval.
  • The new state must match Approved or Rejected.
  • The returned tuple carries the sub-issue ID. It also carries the parent ID. Its final value is the graph decision.
  • Save receiver.py.
  • Compile the complete classification layer by running:
python3 -m py_compile receiver.py

What Does This Check Cover?

The compiler now reads the advisory lock plus both event classifiers. You should return to the prompt without an error.

Classifier Not Compiling?

  • Check that the return annotation contains three str entries.
  • Check that the return statement remains inside the state-matching condition.
  • Use the reported line with help me fix my approval classifier.
Route work through durable checkpoints

Classification identifies a delivery's purpose. The next functions open the saved workflow thread after the HTTP response has already returned.

A partial run continues from its checkpoint. A paused or completed run keeps its existing state. This prevents a fresh delivery from creating duplicate work.

  • Add this start route immediately below approval_decision():
def start_triage(issue_id: str) -> None:
    import app
    import telemetry

    thread, graph_input = app.issue_input(issue_id)
    with thread_lock(thread), app.open_checkpointer() as checkpointer:
        graph = app.build_graph(checkpointer)
        snapshot = graph.get_state({"configurable": {"thread_id": thread}})
        if snapshot.next and not snapshot.interrupts:
            print(f"{thread}: run stopped part-way, retrying from its last checkpoint.")
            app.run_graph(graph, thread, "resume", None)
        elif snapshot.values:
            print(f"{thread}: already has a run, not starting another.")
            return
        else:
            print(f"{thread}: {TRIGGER_LABEL} from Linear, starting.")
            app.run_graph(graph, thread, "start", graph_input)
    telemetry.shutdown()

How Does a Start Stay Idempotent?

  • The imports remain inside the background function. The acknowledgement path therefore avoids loading the graph stack.
  • A snapshot with unfinished work resumes from its saved checkpoint.
  • A snapshot with stored values already represents a paused or completed run. The function logs already has a run, not starting another. before returning.
  • An empty snapshot starts the graph with the issue-derived input.
  • Add this approval route immediately below start_triage():
def resume_approval(sub_issue_id: str, parent_id: str, decision: str) -> None:
    from langgraph.types import Command

    import app
    import telemetry

    thread = linear_api.client().issue(parent_id)["identifier"]
    with thread_lock(thread), app.open_checkpointer() as checkpointer:
        graph = app.build_graph(checkpointer)
        config = {"configurable": {"thread_id": thread}}
        gates = [
            pending.id
            for pending in graph.get_state(config).interrupts
            if pending.value.get("approval_issue_id") == sub_issue_id
        ]
        if not gates:
            print(f"{thread}: no gate waiting on sub-issue {sub_issue_id}.")
            return
        print(f"{thread}: {decision} from Linear, resuming.")
        app.run_graph(graph, thread, "webhook", Command(resume={gates[0]: decision}))
    telemetry.shutdown()

How Does the Decision Reach the Right Gate?

  • The parent issue identifier becomes the durable graph thread ID.
  • The list comprehension checks each pending interrupt's approval_issue_id.
  • The matching interrupt ID becomes the key passed to Command.
  • A repeated decision finds no matching gate. The function logs no gate waiting on sub-issue before returning.
  • Save receiver.py.
  • Compile both workflow routes by running:
python3 -m py_compile receiver.py

What Does This Check Cover?

The compiler now reads the start path plus the approval-resume path. You should return to the prompt without an error.

Workflow Route Not Compiling?

  • Check that the dictionary braces around thread_id are balanced.
  • Check that the gate comprehension remains inside square brackets.
  • Use the reported line with help me fix my workflow routing functions.
  • Find the temporary handle_delivery(payload) function that logs background_payload.
  • Replace that temporary function with this dispatcher:
def handle_delivery(payload: dict) -> None:
    if issue_id := new_incident(payload):
        start_triage(issue_id)
    elif decision := approval_decision(payload):
        resume_approval(*decision)

Why Use One Dispatcher?

The endpoint still schedules only handle_delivery() as background work. The dispatcher sends a valid incident trigger to start_triage().

A valid approval transition goes to resume_approval(). Unrelated deliveries finish without starting graph work.

  • Save receiver.py.
  • Compile the completed receiver by running:
python3 -m py_compile receiver.py

What Does This Check Prove?

The command reads the complete receiver without starting Uvicorn or contacting Linear. A clean return confirms that the dispatcher fits the existing endpoint.

Dispatcher Not Compiling?

  • Check that the elif aligns with the first if.
  • Check that the starred argument remains inside resume_approval(*decision).
  • Use the reported line with help me fix the background dispatcher.

✔️ Awesome, I've got everything!

Great. Confirm that you saved receiver.py before updating the checkpoint boundary.

ⓧ I'd like to double check the full code

Compare your complete receiver.py with this reference.

import hashlib
import hmac
import json
import os
import threading
import time
from collections import OrderedDict, defaultdict
from contextlib import contextmanager

from fastapi import BackgroundTasks, FastAPI, Header, HTTPException, Request

import linear as linear_api
from linear import APPROVALS_TEAM, INCIDENTS_TEAM, TRIGGER_LABEL

MAX_AGE_MS = 60_000
DECISIONS = {"Approved": "approve", "Rejected": "reject"}

api = FastAPI()
_thread_locks: defaultdict[str, threading.Lock] = defaultdict(threading.Lock)
_locks_guard = threading.Lock()
_seen_deliveries: OrderedDict[str, None] = OrderedDict()


def verify_signature(raw_body: bytes, signature: str | None) -> None:
    secret = os.environ["LINEAR_WEBHOOK_SECRET"].encode()
    expected = hmac.new(secret, raw_body, hashlib.sha256).hexdigest()
    if not signature or not hmac.compare_digest(expected, signature):
        raise HTTPException(status_code=401, detail="Bad signature.")


def first_delivery(delivery_id: str | None) -> bool:
    if not delivery_id:
        return True
    if delivery_id in _seen_deliveries:
        return False
    _seen_deliveries[delivery_id] = None
    while len(_seen_deliveries) > 1000:
        _seen_deliveries.popitem(last=False)
    return True


@contextmanager
def thread_lock(thread: str):
    with _locks_guard:
        lock = _thread_locks[thread]
    with lock:
        url = os.environ.get("DATABASE_URL")
        if not url:
            yield
            return
        import psycopg

        with psycopg.connect(url, autocommit=True) as conn:
            conn.execute("SELECT pg_advisory_lock(hashtext(%s))", (thread,))
            yield


def new_incident(payload: dict) -> str | None:
    data = payload.get("data") or {}
    if payload.get("type") != "Issue":
        return None
    linear = linear_api.client()
    if data.get("teamId") != linear.team(INCIDENTS_TEAM)["id"]:
        return None
    trigger = linear.label_id(TRIGGER_LABEL)
    if trigger not in (data.get("labelIds") or []):
        return None
    if payload.get("action") == "create":
        return data["id"]
    old = payload.get("updatedFrom") or {}
    if (
        payload.get("action") == "update"
        and "labelIds" in old
        and trigger not in (old["labelIds"] or [])
    ):
        return data["id"]
    return None


def approval_decision(payload: dict) -> tuple[str, str, str] | None:
    data = payload.get("data") or {}
    old_state = (payload.get("updatedFrom") or {}).get("stateId")
    if payload.get("type") != "Issue" or payload.get("action") != "update" or not old_state:
        return None
    if not data.get("parentId"):
        return None
    linear = linear_api.client()
    if old_state != linear.state_id(APPROVALS_TEAM, "Needs Approval"):
        return None
    for state, decision in DECISIONS.items():
        if data.get("stateId") == linear.state_id(APPROVALS_TEAM, state):
            return data["id"], data["parentId"], decision
    return None


def start_triage(issue_id: str) -> None:
    import app
    import telemetry

    thread, graph_input = app.issue_input(issue_id)
    with thread_lock(thread), app.open_checkpointer() as checkpointer:
        graph = app.build_graph(checkpointer)
        snapshot = graph.get_state({"configurable": {"thread_id": thread}})
        if snapshot.next and not snapshot.interrupts:
            print(f"{thread}: run stopped part-way, retrying from its last checkpoint.")
            app.run_graph(graph, thread, "resume", None)
        elif snapshot.values:
            print(f"{thread}: already has a run, not starting another.")
            return
        else:
            print(f"{thread}: {TRIGGER_LABEL} from Linear, starting.")
            app.run_graph(graph, thread, "start", graph_input)
    telemetry.shutdown()


def resume_approval(sub_issue_id: str, parent_id: str, decision: str) -> None:
    from langgraph.types import Command

    import app
    import telemetry

    thread = linear_api.client().issue(parent_id)["identifier"]
    with thread_lock(thread), app.open_checkpointer() as checkpointer:
        graph = app.build_graph(checkpointer)
        config = {"configurable": {"thread_id": thread}}
        gates = [
            pending.id
            for pending in graph.get_state(config).interrupts
            if pending.value.get("approval_issue_id") == sub_issue_id
        ]
        if not gates:
            print(f"{thread}: no gate waiting on sub-issue {sub_issue_id}.")
            return
        print(f"{thread}: {decision} from Linear, resuming.")
        app.run_graph(graph, thread, "webhook", Command(resume={gates[0]: decision}))
    telemetry.shutdown()


def handle_delivery(payload: dict) -> None:
    if issue_id := new_incident(payload):
        start_triage(issue_id)
    elif decision := approval_decision(payload):
        resume_approval(*decision)


@api.post("/linear/webhook")
async def linear_webhook(
    request: Request,
    background: BackgroundTasks,
    linear_signature: str | None = Header(default=None),
    linear_delivery: str | None = Header(default=None),
):
    raw_body = await request.body()
    verify_signature(raw_body, linear_signature)
    payload = json.loads(raw_body)
    if abs(time.time() * 1000 - payload.get("webhookTimestamp", 0)) > MAX_AGE_MS:
        raise HTTPException(status_code=401, detail="Stale webhook.")
    print(json.dumps({"delivery": linear_delivery, "payload": payload}), flush=True)
    if not first_delivery(linear_delivery):
        return {"ok": True, "duplicate": True}

    background.add_task(handle_delivery, payload)
    return {"ok": True}


@api.get("/health")
def health():
    return {"ok": True}

How to Use This Reference

Compare the function order plus every identifier with your saved file. The endpoint still acknowledges before handle_delivery() performs any Linear lookup or graph work.

The receiver now requests its checkpointer from app.py. Local commands can keep using SQLite. A deployed receiver selects the Postgres checkpointer when DATABASE_URL exists.

  • Switch to app.py in the Cloud Shell editor.
  • Add these imports beside the existing imports:
# Add with the existing imports.
import threading
from contextlib import contextmanager

Why Add These Imports?

The lock coordinates one-time checkpoint table setup inside the process. The context manager gives local runs plus deployed runs one checkpointer interface.

  • Add this checkpoint state plus context manager immediately before build_graph():
# Add immediately before build_graph().
_postgres_ready = threading.Lock()
_postgres_set_up = False


@contextmanager
def open_checkpointer():
    global _postgres_set_up
    url = os.environ.get("DATABASE_URL")
    if not url:
        with SqliteSaver.from_conn_string(DB_PATH) as checkpointer:
            yield checkpointer
        return
    from langgraph.checkpoint.postgres import PostgresSaver

    with PostgresSaver.from_conn_string(url) as checkpointer:
        with _postgres_ready:
            if not _postgres_set_up:
                checkpointer.setup()
                _postgres_set_up = True
        yield checkpointer

How Does the Checkpointer Switch Work?

  • An absent DATABASE_URL keeps terminal-driven runs on the inherited local database.
  • A present database URL lazily imports PostgresSaver.
  • The first database-backed call runs checkpointer.setup() under the process lock.
  • Every caller receives the selected checkpointer through the same context manager.
  • Save app.py.
  • Compile the new checkpoint boundary by running:
python3 -m py_compile app.py

What Does This Check Prove?

The command compiles the inherited workflow plus the new checkpoint context manager. You should return to the prompt without an error.

Checkpointer Not Compiling?

  • Check that global _postgres_set_up remains inside open_checkpointer().
  • Check that each yield checkpointer remains inside its matching context manager.
  • Use the reported line with help me fix the durable checkpointer.

The next edits connect the inherited graph to either checkpointer. They also surface stopped runs directly on the incident.

  • Find the inherited build_graph(checkpointer: SqliteSaver) signature.
  • Replace only that signature with this reference:
# Replace only the signature. Keep the inherited graph body unchanged.
def build_graph(checkpointer):

Why Remove the SQLite Annotation?

The graph can now accept either supported checkpointer. Its inherited graph body remains unchanged.

  • Add this function immediately below the inherited display_failure() function:
# Add immediately below display_failure().
def report_failure(graph, config, error):
    values = graph.get_state(config).values
    if not values.get("issue_id"):
        return
    try:
        linear_client().comment(
            values["issue_id"],
            f"**Run stopped:** `{str(error)[:300]}`\n\n"
            "Its progress is saved. To retry from the last checkpoint, remove the "
            "`agent-triage` label and add it again.",
        )
    except Exception as comment_error:
        print(f"Couldn't post the failure to Linear: {comment_error}")

How Does Failure Stay Visible?

  • The function reads the latest saved graph state.
  • A state without an issue_id cannot be linked back to Linear.
  • The comment starts with Run stopped:.
  • The comment tells responders how to retry from the durable checkpoint.
  • Save app.py.
  • Compile the graph signature plus failure reporter by running:
python3 -m py_compile app.py

What Does This Check Cover?

The compiler now reads the flexible graph signature plus the Linear failure reporter. You should return to the prompt without an error.

Failure Reporter Not Compiling?

  • Check that the quoted guidance lines remain inside linear_client().comment().
  • Check that the exception block aligns with the try block.
  • Use the reported line with help me fix the failure reporter.

The final two edits are fiddly because they sit inside existing functions. Keep the inherited function bodies in place while changing only the checkpoint boundary plus failure path.

  • Find the existing exception handler inside run_graph().
  • Add the failure-report call so the handler matches this reference:
# In run_graph(), extend the existing exception handler.
    except Exception as error:
        display_failure(graph, config, error)
        report_failure(graph, config, error)

Why Report After Displaying the Failure?

The inherited terminal output remains the first failure signal. The next line mirrors that failure onto the Linear incident.

  • Find the direct SQLite context manager inside run_command().
  • Replace the beginning of that function with this reference:
# In run_command(), replace the direct SQLite context manager.
def run_command(args, config):
    with open_checkpointer() as checkpointer:
        graph = build_graph(checkpointer)
        # Keep the rest of the inherited function unchanged and indented here.

How Does the Terminal Path Stay Compatible?

The terminal command now obtains its checkpointer through open_checkpointer(). The inherited command behavior stays inside the new context.

  • Keep the rest of the inherited run_command() body beneath graph = build_graph(checkpointer).
  • Indent that inherited body inside the new context manager.
  • Save app.py.
  • Compile the completed workflow file by running:
python3 -m py_compile app.py

What Does This Check Prove?

The compiler reads the inherited workflow plus every durable-checkpoint change. A clean return confirms that Python accepted the updated file.

Workflow File Not Compiling?

  • Check that report_failure() aligns with display_failure() inside the exception handler.
  • Check that the inherited run_command() body is indented inside the new context manager.
  • Use the reported line with help me fix the final app.py error.

✔️ Awesome, I've got everything!

Excellent. Confirm that you saved app.py with the inherited workflow body intact.

ⓧ I'd like to double check the cumulative changes

Compare each cumulative change below with its matching location in app.py.

# Add with the existing imports.
import threading
from contextlib import contextmanager

Import Placement

These two imports belong with the existing imports at the top of the inherited file.

# Add immediately before build_graph().
_postgres_ready = threading.Lock()
_postgres_set_up = False


@contextmanager
def open_checkpointer():
    global _postgres_set_up
    url = os.environ.get("DATABASE_URL")
    if not url:
        with SqliteSaver.from_conn_string(DB_PATH) as checkpointer:
            yield checkpointer
        return
    from langgraph.checkpoint.postgres import PostgresSaver

    with PostgresSaver.from_conn_string(url) as checkpointer:
        with _postgres_ready:
            if not _postgres_set_up:
                checkpointer.setup()
                _postgres_set_up = True
        yield checkpointer

Checkpointer Placement

The checkpoint state plus context manager belong immediately before the existing graph builder.

# Replace only the signature. Keep the inherited graph body unchanged.
def build_graph(checkpointer):

Graph Signature

Only the graph builder's signature changes. Its inherited body stays in place.

# Add immediately below display_failure().
def report_failure(graph, config, error):
    values = graph.get_state(config).values
    if not values.get("issue_id"):
        return
    try:
        linear_client().comment(
            values["issue_id"],
            f"**Run stopped:** `{str(error)[:300]}`\n\n"
            "Its progress is saved. To retry from the last checkpoint, remove the "
            "`agent-triage` label and add it again.",
        )
    except Exception as comment_error:
        print(f"Couldn't post the failure to Linear: {comment_error}")

Failure Reporter Placement

The new reporter belongs immediately below the inherited terminal failure display function.

# In run_graph(), extend the existing exception handler.
    except Exception as error:
        display_failure(graph, config, error)
        report_failure(graph, config, error)

Exception Handler Change

The existing exception handler gains one call immediately after its terminal failure display.

# In run_command(), replace the direct SQLite context manager.
def run_command(args, config):
    with open_checkpointer() as checkpointer:
        graph = build_graph(checkpointer)
        # Keep the rest of the inherited function unchanged and indented here.

Command Boundary Change

The command path now uses the shared checkpoint context. Every inherited line below the graph construction remains in the function.

Run the offline suite

The complete test suite checks the inherited workflow plus the new delivery routes. Its coverage includes duplicate-start protection plus approval-gate matching.

Before you run the suite, do you expect the local SQLite path to behave the same after adding the deployed Postgres path?

  • Run the full offline test suite from ~/langgraph-triage-linear with this command:
python3 -m pytest

What Does the Full Suite Verify?

The tests exercise the inherited workflow plus the receiver changes without requiring a deployed database. They confirm that the new routing behavior fits the existing graph.

You should see the test session finish with every test passing. You should see no failures.

Seeing a Test Failure?

  • Read the first failing test before changing a later function.
  • Compare the named function with its matching reference above.
  • Check that build_graph() no longer has the SqliteSaver annotation.
  • Use the first failure with help me diagnose my offline test failure.

That coordination boundary is now proven. Your receiver can classify deliveries without duplicating a paused or completed workflow.

Your event routes now survive beyond one receiver process. Next, you will connect those durable checkpoints to Cloud SQL and redeploy the receiver on Cloud Run.

Deploy Cloud SQL Checkpoints

Your receiver now routes each valid Linear delivery to the correct workflow thread. Its checkpoints still need a home beyond the local SQLite file.

A Cloud Run instance can disappear between webhook deliveries. Cloud SQL gives each paused workflow a durable PostgreSQL checkpoint store.

In this step, you will provision the database. Secret Manager will hold the receiver's credentials.

You will also deploy the receiver with a least-privilege service identity. The deployed health endpoint will prove that the new runtime boundary starts successfully.

In this step, get ready to:
  • Provision a Cloud SQL PostgreSQL 16 instance for durable checkpoints.
  • Store exactly four runtime values as version-pinned secrets.
  • Deploy the receiver with its database connection and scoped permissions.
Provision the durable database

Your Cloud Shell session needs access to the five managed services used by this deployment. Enabling their APIs lets the existing project create each resource.

  • Enable the required APIs by running this command:
gcloud services enable \
  run.googleapis.com \
  cloudbuild.googleapis.com \
  artifactregistry.googleapis.com \
  sqladmin.googleapis.com \
  secretmanager.googleapis.com

What does this command enable?

  • The Cloud Run API manages the deployed receiver.
  • The Cloud Build API builds the repository from source.
  • The Artifact Registry API stores the resulting container image.
  • The Cloud SQL Admin API provisions the PostgreSQL instance.
  • The Secret Manager API stores the four runtime values.

You should see each service finish with a successful operation status.

Cloud SQL incurs charges

Cloud SQL is a paid persistent resource. The cleanup section includes options to stop or delete triage-pg after the project.

Creating the database instance takes several minutes. A quiet terminal during provisioning does not mean the command has stalled.

  • Create the triage-pg instance by running this command:
gcloud sql instances create triage-pg \
  --database-version=POSTGRES_16 \
  --edition=ENTERPRISE \
  --tier=db-f1-micro \
  --region=us-central1

Why specify the edition?

PostgreSQL 16 can default to Enterprise Plus. The ENTERPRISE setting makes the shared-core db-f1-micro tier compatible with this instance.

The us-central1 region keeps the database beside the receiver you deploy later in this step.

  • Confirm the new instance configuration by running this command:
gcloud sql instances describe triage-pg

What should you see?

The description should identify POSTGRES_16 as the database version. It should also show the ENTERPRISE edition.

The settings should show db-f1-micro in us-central1.

Instance creation failed?

  • Check that billing is enabled for the existing Google Cloud project.
  • Confirm that your account can create Cloud SQL resources in the selected project.
  • Check whether an organization policy blocks resources in us-central1.
  • Help me diagnose my Cloud SQL instance creation failure.
Create version-pinned runtime secrets

The receiver needs a database password before it can build its connection URL. Keeping the generated password in a shell variable prevents it from appearing in the command output.

  • Generate the database password in PW by running these commands:
PW="$(python3 -c 'import secrets; print(secrets.token_hex(32))')"
test -n "$PW" && echo "Database password is ready."

What do these commands protect?

The first command generates a hexadecimal password inside PW. The second command confirms that the variable contains a value without printing the secret.

  • Create the triage database and its triage user by running these commands:
gcloud sql databases create triage --instance=triage-pg
gcloud sql users create triage --instance=triage-pg --password="$PW"

What do these database objects do?

The triage database holds the durable LangGraph checkpoint tables. The matching user gives the receiver password-based database access.

The checkpoint tables are created when open_checkpointer() first runs against this database.

  • Build the Cloud SQL connection URL by running these commands:
export DATABASE_URL="postgresql://triage:${PW}@/triage?host=/cloudsql/${GOOGLE_CLOUD_PROJECT}:us-central1:triage-pg"
test -n "$DATABASE_URL" && echo "DATABASE_URL is ready."

How does this URL reach Cloud SQL?

The URL authenticates as the triage user. It selects the triage database.

The host points to the Cloud SQL Unix socket mounted inside Cloud Run. The final command confirms that the URL exists without exposing its password.

Secret creation is a credential-handling point. The next commands pass each value through standard input so the secret itself stays out of the command text.

  • Create the four secrets with their first versions by running these commands:
printf '%s' "$LINEAR_API_KEY" | gcloud secrets create linear-api-key --data-file=-
printf '%s' "$LANGFUSE_SECRET_KEY" | gcloud secrets create langfuse-secret-key --data-file=-
printf '%s' "$LINEAR_WEBHOOK_SECRET" | gcloud secrets create linear-webhook-secret --data-file=-
printf '%s' "$DATABASE_URL" | gcloud secrets create database-url --data-file=-

Why create the values this way?

Supplying data during secret creation creates version 1 immediately. Cloud Run can reference that fixed version during startup.

A fixed version keeps a redeployed revision tied to the configuration you verified. It also avoids an unreviewed secret rotation changing a running service.

  • List the saved secrets by running this command:
gcloud secrets list --format='table(name)'

What should the list contain?

You should see linear-api-key, langfuse-secret-key, linear-webhook-secret, and database-url.

Each newly created secret contains version 1.

That is the sensitive configuration secured. Your four runtime values now have version-pinned homes outside the container.

Secret creation failed?

  • Confirm the earlier environment export populated the Linear and Langfuse values.
  • Check that the Secret Manager API finished enabling before retrying the failed command.
  • Confirm that your account can create secrets in the existing project.
  • Help me troubleshoot my four Secret Manager creation commands.
Deploy and verify the receiver

A service identity controls which Google Cloud APIs the deployed container can access. You need the identity from the real service before granting its runtime roles.

The first source deployment creates that service boundary without secret mappings. No webhook points at this deployment during the identity check.

The source build can take several minutes while Cloud Build creates the image. The terminal may stay on the same operation while the build runs.

  • Switch back to the Cloud Shell terminal from earlier.
  • Accept the repository creation prompt if the source deployment asks for confirmation.
  • Create the source-deployed service by running this command:
gcloud run deploy triage-receiver \
  --source . \
  --region=us-central1 \
  --memory=1Gi \
  --max-instances=1 \
  --no-cpu-throttling \
  --set-cloudsql-instances="${GOOGLE_CLOUD_PROJECT}:us-central1:triage-pg" \
  --no-invoker-iam-check \
  --set-env-vars="GOOGLE_CLOUD_PROJECT=${GOOGLE_CLOUD_PROJECT},LANGFUSE_PUBLIC_KEY=${LANGFUSE_PUBLIC_KEY},LANGFUSE_BASE_URL=${LANGFUSE_BASE_URL},LANGFUSE_TRACING_ENVIRONMENT=${LANGFUSE_TRACING_ENVIRONMENT}"

What does this deployment configure?

The source deployment builds the unchanged Dockerfile. That Dockerfile starts uvicorn receiver:api on the port injected by Cloud Run.

The revision receives 1Gi of memory. It can scale to a maximum of one instance.

The --no-cpu-throttling flag selects instance-based billing for post-response background work. The command does not configure a minimum instance.

The Cloud SQL connection mounts triage-pg inside the container. The invoker IAM check is disabled so Linear can reach the public webhook later.

Source deployment failed?

  • Confirm that you are still inside ~/langgraph-triage-linear before retrying the deployment.
  • Check that the earlier API enablement command completed successfully.
  • Confirm that your account has permission to deploy source builds to Cloud Run.
  • Help me diagnose my triage-receiver source deployment failure.
  • Capture the actual deployed service identity by running these commands:
export SERVICE_ACCOUNT="$(gcloud run services describe triage-receiver --region=us-central1 --format='value(spec.template.spec.serviceAccountName)')"
printf '%s\n' "$SERVICE_ACCOUNT"

Why inspect this identity?

The printed email is the identity used by the running revision. Granting roles to this observed identity avoids guessing which project service account Cloud Run selected.

Permission changes can feel broad. These commands grant only secret reading and Cloud SQL connection access to the identity you just inspected.

  • Grant the two required runtime roles by running these commands:
gcloud projects add-iam-policy-binding "$GOOGLE_CLOUD_PROJECT" \
  --member="serviceAccount:${SERVICE_ACCOUNT}" \
  --role="roles/secretmanager.secretAccessor"
gcloud projects add-iam-policy-binding "$GOOGLE_CLOUD_PROJECT" \
  --member="serviceAccount:${SERVICE_ACCOUNT}" \
  --role="roles/cloudsql.client"

What can the service identity access?

The roles/secretmanager.secretAccessor role lets the revision read its four referenced secret versions.

The roles/cloudsql.client role lets the revision connect to Cloud SQL. These commands do not grant the Editor role.

  • Create the secret-enabled revision by running this command:
gcloud run services update triage-receiver \
  --region=us-central1 \
  --set-secrets="LINEAR_API_KEY=linear-api-key:1,LANGFUSE_SECRET_KEY=langfuse-secret-key:1,LINEAR_WEBHOOK_SECRET=linear-webhook-secret:1,DATABASE_URL=database-url:1"

How are the secrets mapped?

The update maps exactly four environment variables to version 1 of their matching secrets. Existing non-secret environment variables remain on the service.

When a workflow first opens DATABASE_URL, PostgresSaver creates the required checkpoint tables. Later revisions reuse those durable tables.

  • Inspect the completed service configuration by running this command:
gcloud run services describe triage-receiver --region=us-central1

What should the revision show?

The service description should show 1Gi of memory. It should show a maximum of one instance with no minimum instance configured.

The description should show instance-based CPU allocation. It should also show the triage-pg Cloud SQL connection.

The secret mappings should reference version 1 for all four secret environment variables. The four Langfuse and Google Cloud configuration values should remain ordinary environment variables.

The service should use the identity stored in SERVICE_ACCOUNT. Public invocation should use the disabled invoker IAM check.

Before the health request, predict whether the receiver can start with its new database boundary.

  • Request the deployed health endpoint by running these commands:
export SERVICE_URL="$(gcloud run services describe triage-receiver --region=us-central1 --format='value(status.url)')"
curl "${SERVICE_URL}/health"

What does this health check prove?

The first command reads the deployed service URL. The second command requests the receiver's /health path.

A successful response proves that Cloud Run retrieved the pinned secrets. It also proves that the container started with the deployed configuration.

You should see {"ok":true}. That is the deployment boundary working with durable database configuration.

Health request failed?

  • Check the latest Cloud Run revision status for a startup failure.
  • Confirm that SERVICE_ACCOUNT received both required roles.
  • Confirm that each secret mapping points to version 1.
  • Help me diagnose my deployed receiver health failure.

✔️ My deployment matches

Your receiver is healthy. Its next workflow run can create durable checkpoints in Cloud SQL.

ⓧ I'd like to double check the deployment

Compare your deployed resources with this final state.

  • The Cloud SQL instance is named triage-pg.
  • The instance uses PostgreSQL 16 with the ENTERPRISE edition.
  • The instance uses the db-f1-micro tier in us-central1.
  • The instance contains the triage database with the triage user.
  • Secret Manager contains exactly the four required secrets.
  • Each required secret has version 1.
  • The triage-receiver service uses 1Gi of memory.
  • The service has a maximum of one instance with no configured minimum.
  • The service uses instance-based billing for background processing.
  • The service has the triage-pg Cloud SQL connection.
  • The service maps only the four required secret environment variables.
  • The service keeps the four required non-secret environment variables as ordinary values.
  • The deployed service identity has Secret Manager Secret Accessor and Cloud SQL Client access.
  • The deployed service identity was not granted Editor.
  • The service disables the invoker IAM check for public webhook delivery.
  • The /health endpoint returns {"ok":true}.

Your receiver now has a durable runtime boundary. Next up, you will start and resume the entire incident workflow from Linear.

Run Triage from Linear

Your Cloud Run service is healthy. That proves the container can answer requests.

Now Linear becomes the control surface for the entire run. You will start the durable LangGraph workflow from an incident. You will also resume its approval gates without terminal decision commands.

In this step, get ready to:
  • Register the deployed receiver as a signed Linear webhook.
  • Start one fresh INC-2041 triage run from Linear.
  • Approve both gates and verify one durable outcome.
Register the webhook and trigger one run

A signed webhook connects issue activity in your workspace to the deployed receiver. The registration helper uses the same signing secret that Cloud Run reads from Secret Manager.

  • Return to the deployed triage-receiver service details from earlier.
  • Copy the service URL.
  • Append /linear/webhook to the copied URL.
  • Record the complete endpoint here: https://triage-receiver-.../linear/webhook.
  • Switch back to the Google Cloud Shell terminal from earlier.
  • Prepare the existing clone and export its saved environment by running:
cd ~/langgraph-triage-linear
source .venv/bin/activate
set -a
source .env
set +a
pwd

What do these commands do?

  • The first command returns your shell to the existing ~/langgraph-triage-linear clone.
  • The source .venv/bin/activate command activates the inherited Python environment.
  • The set -a command marks subsequently loaded values for export.
  • The source .env command loads the existing configuration into the active shell.
  • The set +a command stops automatic exporting.
  • The pwd command confirms your current folder.

You should see a path ending in /langgraph-triage-linear.

Environment values missing?

  • Confirm that .env still exists inside ~/langgraph-triage-linear.
  • Confirm that the saved LINEAR_WEBHOOK_SECRET matches the value stored as Secret Manager version 1.
  • Use this if the environment still does not load: Help me diagnose why my existing .env values are not exported in Google Cloud Shell.
  • Register the deployed webhook endpoint by running:
python linear_webhook.py [[RECEIVER_WEBHOOK_URL="https://triage-receiver-.../linear/webhook"]]

What does this command do?

  • The linear_webhook.py helper registers one issue webhook across the public teams in your existing workspace.
  • The helper names the webhook triage-receiver.
  • The helper signs deliveries with the exported LINEAR_WEBHOOK_SECRET value.

You should see confirmation that the webhook named triage-receiver points to your deployed endpoint. That connection now carries incident events to Cloud Run.

Webhook registration failed?

  • Confirm that the endpoint ends with /linear/webhook.
  • Confirm that your existing Linear API key has workspace administration access.
  • Use this if the helper still fails: Help me troubleshoot linear_webhook.py without changing my signing secret.

The seed helper can add the trigger label immediately. This run leaves the label off so you can create one deliberate delivery from the Linear interface.

  • Seed a fresh INC-2041 incident without its trigger label by running:
python seed_incidents.py INC-2041 --no-label

Why leave off the label?

The --no-label flag creates the incident without agent-triage. This gives you control over the exact moment the webhook starts the run.

You should see a new issue identifier followed by INC-2041 and its Linear URL.

Incident creation failed?

  • Confirm that LINEAR_API_KEY was loaded from the existing .env file.
  • Confirm that your existing workspace still contains the incident fixture used by seed_incidents.py.
  • Use this if the helper does not create the issue: Help me troubleshoot seed_incidents.py for INC-2041.
  • Follow the Linear issue URL printed by the seed command.
  • Confirm that the new incident does not have the agent-triage label.

Before you add the label, do you expect the receiver to start once or react again to its own Linear updates?

  • Select the label control in the incident header.
  • Search for agent-triage.
  • Select agent-triage once.

Cold starts can trigger a retry

A cold Cloud Run instance can take longer than Linear's five-second delivery limit. Linear may mark the first attempt as failed while the late background task still finishes.

Delivery deduplication and durable thread state keep that retry safe. Leave the label in place while the workflow runs.

  • Watch the parent incident until it shows one findings comment.
  • Wait for the payments approval sub-issue.
  • Wait for the platform-data approval sub-issue.

The receiver has now acknowledged the trigger and started background workflow work. INC-2041 should have exactly one findings comment and two approval sub-issues.

Approve both workflow gates

Each approval sub-issue carries the identifier of one paused interrupt. The receiver compares that issue with approval_issue_id before resuming the matching gate.

  • Select the payments approval sub-issue.
  • Open its current Needs Approval status.
  • Select Approved.

You should see the payments sub-issue in the Approved state. Its webhook resumes only the interrupt associated with that sub-issue.

  • Return to the INC-2041 parent incident.
  • Select the platform-data approval sub-issue.
  • Open its current Needs Approval status.
  • Select Approved.

You should see the platform-data sub-issue in the Approved state. With both matching gates resolved, the graph can finish from its saved checkpoints.

Approval did not resume the run?

  • Confirm that each sub-issue moved directly from Needs Approval to Approved.
  • Confirm that you changed the generated approval sub-issues under the current INC-2041 parent.
  • Use this if one gate remains paused: Help me diagnose why an Approved Linear sub-issue did not resume its matching LangGraph interrupt.
Inspect the durable outcome

The final incident state proves more than successful model output. It shows that Cloud SQL preserved the workflow across event-driven starts and resumes.

  • Return to the INC-2041 parent incident.
  • Confirm that its status is Done.
  • Count the findings comments.
  • Confirm that there is exactly one findings comment.
  • Count the comments beginning Runbook published.
  • Confirm that there is exactly one final runbook comment.
  • Open the Langfuse trace attachment from the incident.
  • Confirm that the trace contains the completed INC-2041 workflow.

Before you read the service logs, how many workflow start lines do you expect after the receiver processes its own Linear writes?

  • Switch back to the Google Cloud Shell terminal from earlier.
  • Read the recent triage-receiver logs by running:
gcloud run services logs read triage-receiver --region=us-central1 --limit=100

What does this command show?

  • The command reads recent entries from the deployed triage-receiver service.
  • The --region=us-central1 flag targets the deployed region.
  • The --limit=100 flag keeps the output focused on recent activity.

You should find one line ending in agent-triage from Linear, starting. You should also find two approval resume lines.

That is the durable result: workflow-generated Linear updates did not create a second start. Your incident moved from trigger to publication entirely through Linear.

Seeing a stopped run?

A recoverable model failure may produce 429 RESOURCE_EXHAUSTED and a Linear comment beginning Run stopped:. The comment confirms that the workflow saved its progress.

  • Remove the agent-triage label from the same incident.
  • Add the agent-triage label once more.
  • Wait for the receiver to retry from the saved Cloud SQL checkpoint.
  • Use this if the retry does not continue: Help me diagnose a stopped INC-2041 run that should resume from its Cloud SQL checkpoint.

✔️ Awesome, I've got everything!

Great work. INC-2041 is complete with one durable run and no self-triggered duplicate.

ⓧ I'd like to double check the result

  • The Linear webhook points to the deployed triage-receiver endpoint ending in /linear/webhook.
  • The webhook uses the same signing secret stored as Secret Manager version 1.
  • The Cloud SQL checkpoint tables contain the completed INC-2041 workflow state.
  • INC-2041 is in Done with exactly one findings comment.
  • The payments and platform-data sub-issues are Approved.
  • INC-2041 has exactly one comment beginning Runbook published.
  • The completed workflow is available through its Langfuse trace attachment.
  • The Cloud Run logs contain one workflow start caused by the agent-triage trigger.

Secret mission

Prove a Completed Incident Is Idempotent

Send a fresh trigger to completed INC-2041. Prove durable Postgres state blocks another workflow run. Confirm every Linear side effect remains unchanged.

Clean Up Your Resources

Clean Up Your Resources

This project created a paid Cloud SQL instance in Google Cloud. Decide whether to keep the project active, pause it for later, or delete its resources.

Cost warning

Cloud SQL is the main ongoing cost in this project. Stopping triage-pg reduces its compute charges.

Retained storage can still incur charges. Other project services can add usage charges. Choose Pause or Delete unless you plan to continue soon.

Resources you used:

  • A Linear webhook targeting the deployed triage-receiver /linear/webhook endpoint.
  • A Cloud Run service named triage-receiver in us-central1.
  • A Cloud SQL instance named triage-pg running PostgreSQL 16 on the db-f1-micro tier. It contains the triage database plus the durable LangGraph checkpoint tables.
  • Four Secret Manager secrets named linear-api-key, langfuse-secret-key, linear-webhook-secret, and database-url. Each secret is pinned to version 1.
  • Source-build images created for triage-receiver through Artifact Registry and Cloud Build.

Keep everything running

No action is needed. Choose this if you plan to keep testing the deployed receiver or revisit its durable workflow soon.

  • Leave the Linear webhook registered for the triage-receiver endpoint.
  • Leave the triage-receiver Cloud Run service deployed with no minimum instances.
  • Leave the triage-pg Cloud SQL instance running so new workflow deliveries can reach its checkpoints.
  • Leave the four Secret Manager secrets pinned to version 1.
  • Review your Google Cloud charges while the paid database remains available.

Pause - I'll come back to this later

Pause incoming workflow events while preserving the deployed service, secrets, and durable checkpoints. Cloud SQL storage charges can continue while its compute is stopped.

  • Remove the webhook targeting /linear/webhook from the webhook settings in your existing Linear workspace.
  • Stop triage-pg from the Cloud SQL instance list for us-central1.
  • Leave the triage-receiver Cloud Run service deployed with no minimum instances.
  • Leave the four Secret Manager secrets and source-build images in place for your return.
  • Start triage-pg from the same Cloud SQL instance list when you return.
  • Register the Linear webhook again with the supplied linear_webhook.py helper after the database is running.

Delete - I don't want to use this again

Deleting these resources is deliberately final. Your completed incident stays in Linear while the deployed receiver plus its durable checkpoints are removed.

  • Delete the webhook targeting the triage-receiver /linear/webhook endpoint from the webhook settings in your existing Linear workspace.
  • Match the source-build image referenced by the current triage-receiver revision to its image in Artifact Registry.
  • Delete only the matched source-build image versions from the Artifact Registry repository list.
  • Delete triage-receiver from the Cloud Run service list for us-central1.
  • Delete triage-pg from the Cloud SQL instance list for us-central1. This also removes the triage database, the triage user, and the durable checkpoint tables.
  • Delete linear-api-key, langfuse-secret-key, linear-webhook-secret, and database-url from the Secret Manager secrets list.
  • Keep the existing Google Cloud project, Linear workspace, Langfuse project, and ~/langgraph-triage-linear clone because they came from the prerequisite projects.
  • Confirm the webhook, matched images, Cloud Run service, Cloud SQL instance, and four secrets no longer appear in their respective resource lists.

Nice Work!

Nice Work!

You did it! You deployed a signed Linear webhook receiver on Google Cloud Run that turns issue updates into a restart-safe incident workflow.

You've learned how to:

  • Built a signed FastAPI receiver that authenticates Linear webhook deliveries. Immediate acknowledgement moves verified work into background processing. Delivery deduplication prevents retries from repeating the same background task.
  • Protected each incident thread with process-local locks plus PostgreSQL advisory locks. Durable LangGraph checkpoints in Cloud SQL preserved paused work across Cloud Run instance shutdowns.
  • Drove INC-2041 entirely from Linear. The agent-triage trigger started the workflow. Two matching approval interrupts resumed before one final Runbook published comment appeared.
  • Completed the optional Secret Mission by re-triggering INC-2041. The completed PostgreSQL checkpoint rejected the fresh start without duplicating findings, approval sub-issues, or runbook publication.

Ready to quiz yourself?