Linear

import { createRegistry } from "@earendil-works/pi-durable";
import { pi } from "@rivet-dev/pi";
import { Hono } from "hono";
import { setup } from "rivetkit";
import { type LinearComment, linearRoutes, postReply } from "./linear";

const agent = pi({
	model: "anthropic/claude-opus-5-5",
	registry: createRegistry(),
	actions: {
		// Schedules the answer and returns, so the route can respond in time.
		receive: async (c, comment: LinearComment) => {
			await c.schedule.after(0, "reply", comment);
		},
		reply: async (c, comment: LinearComment): Promise<void> => {
			const self = c.client<typeof registry>().agent.getForId(c.actorId);
			// A message sent during a run waits as a follow-up.
			// The requestId stops a duplicate delivery from running twice.
			const result = await self.prompt(comment.text, { requestId: comment.commentId });
			await postReply(comment, result.status === "done" ? result.text : undefined);
		},
	},
});

export const registry = setup({ use: { agent } });

const app = new Hono();
app.all("/api/rivet/*", (c) => registry.handler(c.req.raw));
app.route("/linear", linearRoutes);

export default app;

import { createHmac, timingSafeEqual } from "node:crypto";
import { LinearClient } from "@linear/sdk";
import { Hono } from "hono";
import { createClient } from "rivetkit/client";
import type { registry } from "./server";

export type LinearComment = { commentId: string; issueId: string; text: string };

const trigger = process.env.LINEAR_TRIGGER ?? "@agent";
const linear = new LinearClient({ apiKey: process.env.LINEAR_API_KEY });
const agentUserId = linear.viewer.then((viewer) => viewer.id);
const client = createClient<typeof registry>();

export const linearRoutes = new Hono();

linearRoutes.post("/webhook", async (c) => {
	const body = await c.req.text();
	if (!verifyLinearSignature(body, c.req.header("linear-signature") ?? "")) {
		return c.text("invalid signature", 401);
	}

	const payload = JSON.parse(body);
	if (Math.abs(Date.now() - payload.webhookTimestamp) > 60 * 1000) return c.text("stale webhook", 401);

	const comment = payload.data;
	if (
		payload.type === "Comment" &&
		payload.action === "create" &&
		comment.issueId &&
		comment.body.includes(trigger) &&
		comment.userId !== (await agentUserId)
	) {
		const agent = client.agent.getOrCreate(["linear", comment.issueId]);
		await agent.receive({ commentId: comment.id, issueId: comment.issueId, text: comment.body });
	}
	return c.body(null, 200);
});

export async function postReply(comment: LinearComment, reply: string | undefined) {
	await linear.createComment({ issueId: comment.issueId, body: reply || "I finished without a reply." });
}

function verifyLinearSignature(body: string, signature: string): boolean {
	const secret = process.env.LINEAR_WEBHOOK_SECRET ?? "";
	const expected = createHmac("sha256", secret).update(body).digest("hex");
	return expected.length === signature.length && timingSafeEqual(Buffer.from(expected), Buffer.from(signature));
}