diff --git a/docs/contributing/architecture.md b/docs/contributing/architecture.md index b7d2737..9ba6c2e 100644 --- a/docs/contributing/architecture.md +++ b/docs/contributing/architecture.md @@ -57,8 +57,10 @@ system without knowing which operation or delivery host called them. Adding a new operation does not require adding a station to a shared lifecycle. The current worker registers three independent functions: polling, readiness -routing, and triage. Spec and implementation requests are typed but remain -disabled until they have their own consumers. +routing, and triage. The composition's enabled routes also choose the states the +poller observes: Backlog is always observed, and Open is added only when a Spec +or Implement consumer is registered. Spec and implementation requests are typed +but remain disabled until they have their own consumers. ### Automation construction contract diff --git a/docs/contributing/linear-automation.md b/docs/contributing/linear-automation.md index 0d3c490..fbddd8f 100644 --- a/docs/contributing/linear-automation.md +++ b/docs/contributing/linear-automation.md @@ -6,12 +6,12 @@ or a Linear webhook. ```text one-minute Inngest cron - -> list configured project Backlog revisions - -> linear/issue.revision-observed - -> reload complete Linear context - -> classify readiness - -> work/triage.requested - -> triage decision and Linear projection + -> list the configured project's observed states + -> Backlog: linear/issue.revision-observed + -> Open: linear/issue.readiness-check.requested + -> reload complete Linear context and classify readiness + -> work/triage.requested + -> work/spec.requested or work/implementation.requested ``` Linear remains the queue. A Backlog issue without an Agent action label needs @@ -19,6 +19,14 @@ triage. Successful triage moves it out of Backlog. Work identity includes the issue revision, so repeated observation of one revision converges while a later human change can request new work. +Open is observed only when the worker composition enables Spec or Implement and +registers the matching consumer. Each poll cycle gives an Open readiness check +new delivery identity because blocker state can change without changing the +blocked issue's revision. The router reloads current labels, state, and blockers, +then emits a work request whose identity comes from that current readiness +snapshot. Repeated checks of an unchanged snapshot therefore converge, while a +resolved blocker produces new work. + ## Workflow contract Statuses say who owns the next move. Agent action labels say what an agent @@ -81,9 +89,10 @@ secrets. } ``` -The initial worker composes one configured project. The standalone Linear read -operation accepts explicit team, project, and state IDs so another worker can -reuse it without adding a shared scheduler or project registry. +The initial worker composes one configured project. Its route map controls both +which consumers exist and which states the poller observes. The standalone +Linear read operation accepts explicit team, project, and state IDs so another +worker can reuse it without adding a shared scheduler or project registry. ## Run the local Compose stack @@ -178,13 +187,15 @@ project per Compose stack until app and function identities become project-aware The worker registers exactly three functions: -- the poller lists at most 250 matching issue revisions every minute and fails - visibly if that bound is exceeded; +- the poller lists at most 250 issue revisions per observed state every minute + and fails the whole poll visibly if any state exceeds that bound; - the readiness router reloads complete current context and emits a provider-neutral work request; and - the triage consumer invokes the configured agent and projects the decision. -Spec and Implement routes remain disabled. The poller accepts an explicit +Spec and Implement routes remain disabled, so the current composition observes +Backlog only. Enabling either route adds Open observation in the same composition +change that registers its consumer. The poller accepts an explicit `linear/poll.requested` event for deterministic smoke coverage and immediate operator checks, but cron is the only automatic trigger. diff --git a/lib/linear-automation/backlog-poller.test.ts b/lib/linear-automation/backlog-poller.test.ts deleted file mode 100644 index 86574ff..0000000 --- a/lib/linear-automation/backlog-poller.test.ts +++ /dev/null @@ -1,137 +0,0 @@ -import { Inngest } from "inngest"; -import { InngestTestEngine } from "@inngest/test"; -import { describe, expect, it, vi } from "vitest"; -import { - LINEAR_ISSUE_REVISION_EVENT_NAME, - LinearPollRequestedEvent, -} from "./events/linear-revision-events.ts"; -import { - createLinearBacklogPoller, - LINEAR_BACKLOG_LIST_STEP_ID, - LINEAR_BACKLOG_POLL_CRON, - LINEAR_BACKLOG_POLL_FUNCTION_ID, - LINEAR_BACKLOG_POLL_LIMIT, - LINEAR_BACKLOG_POLL_RETRIES, - LINEAR_BACKLOG_SEND_STEP_ID, - type LinearBacklogPollerLinear, -} from "./backlog-poller.ts"; - -const config = { - teamId: "team-1", - projectId: "project-1", - stateId: "state-backlog", -}; - -function client() { - return new Inngest({ - id: "linear-backlog-poller-test", - eventKey: "test", - fetch: async () => Response.json({ ids: ["sent-event"], status: 200 }), - }); -} - -function poller(linear: LinearBacklogPollerLinear) { - return createLinearBacklogPoller({ client: client(), linear, config }); -} - -function linear(result: Awaited>) { - const listIssueRevisions = vi.fn(async () => - Promise.resolve(result), - ); - return { - service: { listIssueRevisions } satisfies LinearBacklogPollerLinear, - listIssueRevisions, - }; -} - -function pollEvent() { - return LinearPollRequestedEvent.create({}, { id: "poll-test" }); -} - -describe("Linear Backlog poller", () => { - it("registers the one-minute cron and explicit poll trigger", () => { - const fake = linear({ revisions: [], truncated: false }); - - expect(poller(fake.service).opts).toMatchObject({ - id: LINEAR_BACKLOG_POLL_FUNCTION_ID, - concurrency: 1, - retries: LINEAR_BACKLOG_POLL_RETRIES, - triggers: [{ cron: LINEAR_BACKLOG_POLL_CRON }, LinearPollRequestedEvent], - }); - }); - - it("lists the exact configured Backlog and sends one event per revision", async () => { - const fake = linear({ - revisions: [ - { - id: "issue-1", - identifier: "FER-1", - updatedAt: "2026-07-20T20:00:00.000Z", - }, - { - id: "issue-2", - identifier: "FER-2", - updatedAt: "2026-07-20T20:01:00.000Z", - }, - ], - truncated: false, - }); - const output = await new InngestTestEngine({ - function: poller(fake.service), - events: [pollEvent()], - }).execute(); - - expect(output.error).toBeUndefined(); - expect(output.result).toMatchObject({ outcome: "observed", observed: 2 }); - expect(fake.listIssueRevisions).toHaveBeenCalledExactlyOnceWith({ - ...config, - limit: LINEAR_BACKLOG_POLL_LIMIT, - }); - expect(output.ctx.step.run).toHaveBeenCalledWith( - LINEAR_BACKLOG_LIST_STEP_ID, - expect.any(Function), - ); - const sent = vi.mocked(output.ctx.step.sendEvent).mock.calls[0]?.[1]; - expect(sent).toEqual([ - expect.objectContaining({ name: LINEAR_ISSUE_REVISION_EVENT_NAME }), - expect.objectContaining({ name: LINEAR_ISSUE_REVISION_EVENT_NAME }), - ]); - expect(output.ctx.step.sendEvent).toHaveBeenCalledWith(LINEAR_BACKLOG_SEND_STEP_ID, sent); - }); - - it("returns without a send step when the Backlog is empty", async () => { - const fake = linear({ revisions: [], truncated: false }); - const output = await new InngestTestEngine({ - function: poller(fake.service), - events: [pollEvent()], - }).execute(); - - expect(output.result).toEqual({ outcome: "empty", observed: 0 }); - expect(output.ctx.step.sendEvent).not.toHaveBeenCalled(); - }); - - it("fails visibly instead of sending a truncated Backlog", async () => { - const fake = linear({ revisions: [], truncated: true }); - const output = await new InngestTestEngine({ - function: poller(fake.service), - events: [pollEvent()], - }).execute(); - - expect(output.error).toMatchObject({ - message: expect.stringContaining(`${LINEAR_BACKLOG_POLL_LIMIT}-issue limit`), - }); - expect(output.ctx.step.sendEvent).not.toHaveBeenCalled(); - }); - - it("rejects incomplete poller configuration before creating the function", () => { - const fake = linear({ revisions: [], truncated: false }); - - expect(() => - createLinearBacklogPoller({ - client: client(), - linear: fake.service, - config: { ...config, projectId: "" }, - }), - ).toThrow(/projectId/); - }); -}); diff --git a/lib/linear-automation/backlog-poller.ts b/lib/linear-automation/backlog-poller.ts deleted file mode 100644 index 4912270..0000000 --- a/lib/linear-automation/backlog-poller.ts +++ /dev/null @@ -1,78 +0,0 @@ -import { cron, type Inngest, type InngestFunction } from "inngest"; -import { z } from "zod"; -import { - createLinearIssueRevisionObservedEvent, - LinearPollRequestedEvent, -} from "./events/linear-revision-events.ts"; -import { LinearError } from "../linear/error.ts"; -import type { LinearService } from "../linear/client.ts"; -import type { ListIssueRevisionsResult } from "../linear/types.ts"; - -export const LINEAR_BACKLOG_POLL_FUNCTION_ID = "poll-linear-backlog-v1"; -export const LINEAR_BACKLOG_POLL_RETRIES = 3; -export const LINEAR_BACKLOG_POLL_CRON = "* * * * *"; -export const LINEAR_BACKLOG_POLL_LIMIT = 250; -export const LINEAR_BACKLOG_LIST_STEP_ID = "list-linear-backlog-revisions-v1"; -export const LINEAR_BACKLOG_SEND_STEP_ID = "send-linear-revision-events-v1"; - -const LinearBacklogPollerConfigSchema = z - .object({ - teamId: z.string().trim().min(1), - projectId: z.string().trim().min(1), - stateId: z.string().trim().min(1), - }) - .strict(); - -export type LinearBacklogPollerLinear = Pick; - -export type LinearBacklogPollerConfig = Readonly>; - -export function createLinearBacklogPoller(input: { - client: Inngest.Any; - linear: LinearBacklogPollerLinear; - config: LinearBacklogPollerConfig; -}): InngestFunction.Any { - const config = LinearBacklogPollerConfigSchema.parse(input.config); - - return input.client.createFunction( - { - id: LINEAR_BACKLOG_POLL_FUNCTION_ID, - concurrency: 1, - retries: LINEAR_BACKLOG_POLL_RETRIES, - triggers: [cron(LINEAR_BACKLOG_POLL_CRON), LinearPollRequestedEvent], - }, - async ({ step }) => { - const result: ListIssueRevisionsResult = await step.run(LINEAR_BACKLOG_LIST_STEP_ID, () => - input.linear.listIssueRevisions({ - teamId: config.teamId, - projectId: config.projectId, - stateId: config.stateId, - limit: LINEAR_BACKLOG_POLL_LIMIT, - }), - ); - if (result.truncated) { - throw new LinearError( - "incomplete", - `Linear Backlog poll exceeded its ${LINEAR_BACKLOG_POLL_LIMIT}-issue limit.`, - ); - } - if (result.revisions.length === 0) { - return { outcome: "empty" as const, observed: 0 }; - } - - const events = result.revisions.map((revision) => - createLinearIssueRevisionObservedEvent({ - issueId: revision.id, - issueIdentifier: revision.identifier, - updatedAt: revision.updatedAt, - }), - ); - await step.sendEvent(LINEAR_BACKLOG_SEND_STEP_ID, events); - return { - outcome: "observed" as const, - observed: events.length, - eventIds: events.map((event) => event.id), - }; - }, - ); -} diff --git a/lib/linear-automation/events/linear-readiness-events.test.ts b/lib/linear-automation/events/linear-readiness-events.test.ts new file mode 100644 index 0000000..af55fe8 --- /dev/null +++ b/lib/linear-automation/events/linear-readiness-events.test.ts @@ -0,0 +1,47 @@ +import { describe, expect, it } from "vitest"; +import { + createLinearIssueReadinessCheckRequestedEvent, + LINEAR_ISSUE_READINESS_CHECK_EVENT_ID_PREFIX, + LINEAR_ISSUE_READINESS_CHECK_EVENT_NAME, + LINEAR_ISSUE_READINESS_CHECK_EVENT_VERSION, + LinearIssueReadinessCheckDataSchema, + LinearIssueReadinessCheckRequestedEvent, + linearIssueReadinessCheckEventId, +} from "./linear-readiness-events.ts"; + +const issue = { + issueId: "issue-1", + issueIdentifier: "FER-270", +}; + +describe("Linear readiness-check events", () => { + it("locks a strict versioned identity-only contract", () => { + expect(LinearIssueReadinessCheckRequestedEvent).toMatchObject({ + name: LINEAR_ISSUE_READINESS_CHECK_EVENT_NAME, + version: LINEAR_ISSUE_READINESS_CHECK_EVENT_VERSION, + }); + expect(LinearIssueReadinessCheckDataSchema.safeParse(issue).success).toBe(true); + expect( + LinearIssueReadinessCheckDataSchema.safeParse({ + ...issue, + updatedAt: "2026-07-22T01:00:00.000Z", + }).success, + ).toBe(false); + }); + + it("uses the poll cycle for delivery identity without adding it to event data", () => { + const first = linearIssueReadinessCheckEventId(issue, "poll-cycle-1"); + const retried = linearIssueReadinessCheckEventId({ ...issue }, "poll-cycle-1"); + const nextCycle = linearIssueReadinessCheckEventId(issue, "poll-cycle-2"); + + expect(first).toBe(retried); + expect(first).not.toBe(nextCycle); + expect(first).toMatch(new RegExp(`^${LINEAR_ISSUE_READINESS_CHECK_EVENT_ID_PREFIX}`)); + expect(createLinearIssueReadinessCheckRequestedEvent(issue, "poll-cycle-1")).toMatchObject({ + id: first, + name: LINEAR_ISSUE_READINESS_CHECK_EVENT_NAME, + v: LINEAR_ISSUE_READINESS_CHECK_EVENT_VERSION, + data: issue, + }); + }); +}); diff --git a/lib/linear-automation/events/linear-readiness-events.ts b/lib/linear-automation/events/linear-readiness-events.ts new file mode 100644 index 0000000..6b0f5a9 --- /dev/null +++ b/lib/linear-automation/events/linear-readiness-events.ts @@ -0,0 +1,54 @@ +import { createHash } from "node:crypto"; +import { eventType } from "inngest"; +import { z } from "zod"; + +export const LINEAR_ISSUE_READINESS_CHECK_EVENT_NAME = "linear/issue.readiness-check.requested"; +export const LINEAR_ISSUE_READINESS_CHECK_EVENT_VERSION = "1"; +export const LINEAR_ISSUE_READINESS_CHECK_EVENT_ID_PREFIX = "linear-issue-readiness-check-v1:"; + +const nonEmptyStringSchema = z.string().refine((value) => value.trim() !== ""); + +export const LinearIssueReadinessCheckDataSchema = z + .object({ + issueId: nonEmptyStringSchema, + issueIdentifier: nonEmptyStringSchema, + }) + .strict(); + +export type LinearIssueReadinessCheckData = Readonly< + z.infer +>; + +export const LinearIssueReadinessCheckRequestedEvent = eventType( + LINEAR_ISSUE_READINESS_CHECK_EVENT_NAME, + { + schema: LinearIssueReadinessCheckDataSchema, + version: LINEAR_ISSUE_READINESS_CHECK_EVENT_VERSION, + }, +); + +export function linearIssueReadinessCheckEventId( + data: LinearIssueReadinessCheckData, + pollCycleId: string, +): string { + const parsed = LinearIssueReadinessCheckDataSchema.parse(data); + const parsedPollCycleId = nonEmptyStringSchema.parse(pollCycleId); + const identity = [ + LINEAR_ISSUE_READINESS_CHECK_EVENT_NAME, + LINEAR_ISSUE_READINESS_CHECK_EVENT_VERSION, + parsed.issueId, + parsedPollCycleId, + ]; + const digest = createHash("sha256").update(JSON.stringify(identity)).digest("hex"); + return `${LINEAR_ISSUE_READINESS_CHECK_EVENT_ID_PREFIX}${digest}`; +} + +export function createLinearIssueReadinessCheckRequestedEvent( + data: LinearIssueReadinessCheckData, + pollCycleId: string, +) { + const parsed = LinearIssueReadinessCheckDataSchema.parse(data); + return LinearIssueReadinessCheckRequestedEvent.create(parsed, { + id: linearIssueReadinessCheckEventId(parsed, pollCycleId), + }); +} diff --git a/lib/linear-automation/issue-poller.test.ts b/lib/linear-automation/issue-poller.test.ts new file mode 100644 index 0000000..538533b --- /dev/null +++ b/lib/linear-automation/issue-poller.test.ts @@ -0,0 +1,243 @@ +import { Inngest } from "inngest"; +import { InngestTestEngine } from "@inngest/test"; +import { describe, expect, it, vi } from "vitest"; +import { LINEAR_ISSUE_READINESS_CHECK_EVENT_NAME } from "./events/linear-readiness-events.ts"; +import { + LINEAR_ISSUE_REVISION_EVENT_NAME, + LinearPollRequestedEvent, +} from "./events/linear-revision-events.ts"; +import { + createLinearIssuePoller, + LINEAR_ISSUE_LIST_STEP_ID, + LINEAR_ISSUE_POLL_CRON, + LINEAR_ISSUE_POLL_FUNCTION_ID, + LINEAR_ISSUE_POLL_LIMIT, + LINEAR_ISSUE_POLL_RETRIES, + LINEAR_ISSUE_SEND_STEP_ID, + type LinearIssuePollerLinear, +} from "./issue-poller.ts"; +import type { ListIssueRevisionsResult } from "../linear/types.ts"; + +const config = { + teamId: "team-1", + projectId: "project-1", + stateIds: { + backlog: "state-backlog", + open: "state-open", + }, +}; + +function client() { + return new Inngest({ + id: "linear-issue-poller-test", + eventKey: "test", + fetch: async () => Response.json({ ids: ["sent-event"], status: 200 }), + }); +} + +function poller( + linear: LinearIssuePollerLinear, + configOverride: Parameters[0]["config"] = config, +) { + return createLinearIssuePoller({ client: client(), linear, config: configOverride }); +} + +function linear(results: Readonly>) { + const listIssueRevisions = vi.fn( + async ({ stateId }) => results[stateId] ?? { revisions: [], truncated: false }, + ); + return { + service: { listIssueRevisions } satisfies LinearIssuePollerLinear, + listIssueRevisions, + }; +} + +function pollEvent(id = "poll-test") { + return LinearPollRequestedEvent.create({}, { id }); +} + +function revision(id: string, identifier: string, updatedAt: string) { + return { id, identifier, updatedAt }; +} + +describe("Linear issue poller", () => { + it("registers the one-minute cron and explicit poll trigger", () => { + const fake = linear({}); + + expect(poller(fake.service).opts).toMatchObject({ + id: LINEAR_ISSUE_POLL_FUNCTION_ID, + concurrency: 1, + retries: LINEAR_ISSUE_POLL_RETRIES, + triggers: [{ cron: LINEAR_ISSUE_POLL_CRON }, LinearPollRequestedEvent], + }); + }); + + it("keeps Backlog revision identity and adds per-cycle Open readiness checks", async () => { + const fake = linear({ + [config.stateIds.backlog]: { + revisions: [revision("issue-1", "FER-1", "2026-07-20T20:00:00.000Z")], + truncated: false, + }, + [config.stateIds.open]: { + revisions: [revision("issue-2", "FER-2", "2026-07-20T20:01:00.000Z")], + truncated: false, + }, + }); + const first = await new InngestTestEngine({ + function: poller(fake.service), + events: [pollEvent("poll-cycle-1")], + }).execute(); + const repeatedCycle = await new InngestTestEngine({ + function: poller(fake.service), + events: [pollEvent("poll-cycle-1")], + }).execute(); + const nextCycle = await new InngestTestEngine({ + function: poller(fake.service), + events: [pollEvent("poll-cycle-2")], + }).execute(); + + expect(first.error).toBeUndefined(); + expect(first.result).toMatchObject({ + outcome: "observed", + observed: 2, + revisions: 1, + readinessChecks: 1, + }); + expect(fake.listIssueRevisions).toHaveBeenCalledWith({ + teamId: config.teamId, + projectId: config.projectId, + stateId: config.stateIds.backlog, + limit: LINEAR_ISSUE_POLL_LIMIT, + }); + expect(fake.listIssueRevisions).toHaveBeenCalledWith({ + teamId: config.teamId, + projectId: config.projectId, + stateId: config.stateIds.open, + limit: LINEAR_ISSUE_POLL_LIMIT, + }); + expect(first.ctx.step.run).toHaveBeenCalledWith( + LINEAR_ISSUE_LIST_STEP_ID, + expect.any(Function), + ); + const firstSent = vi.mocked(first.ctx.step.sendEvent).mock.calls[0]?.[1]; + const repeatedSent = vi.mocked(repeatedCycle.ctx.step.sendEvent).mock.calls[0]?.[1]; + const nextSent = vi.mocked(nextCycle.ctx.step.sendEvent).mock.calls[0]?.[1]; + expect(firstSent).toEqual([ + expect.objectContaining({ name: LINEAR_ISSUE_REVISION_EVENT_NAME }), + expect.objectContaining({ name: LINEAR_ISSUE_READINESS_CHECK_EVENT_NAME }), + ]); + expect(first.ctx.step.sendEvent).toHaveBeenCalledWith(LINEAR_ISSUE_SEND_STEP_ID, firstSent); + expect(eventIds(firstSent)).toEqual(eventIds(repeatedSent)); + expect(eventIds(firstSent)[0]).toBe(eventIds(nextSent)[0]); + expect(eventIds(firstSent)[1]).not.toBe(eventIds(nextSent)[1]); + }); + + it("observes Backlog only when Open routes are not composed", async () => { + const fake = linear({ + [config.stateIds.backlog]: { + revisions: [revision("issue-1", "FER-1", "2026-07-20T20:00:00.000Z")], + truncated: false, + }, + }); + const backlogOnly = { ...config, stateIds: { backlog: config.stateIds.backlog } }; + const output = await new InngestTestEngine({ + function: poller(fake.service, backlogOnly), + events: [pollEvent()], + }).execute(); + + expect(fake.listIssueRevisions).toHaveBeenCalledExactlyOnceWith({ + teamId: config.teamId, + projectId: config.projectId, + stateId: config.stateIds.backlog, + limit: LINEAR_ISSUE_POLL_LIMIT, + }); + expect(output.result).toMatchObject({ revisions: 1, readinessChecks: 0 }); + }); + + it("merges duplicate records within each event kind", async () => { + const duplicate = revision("issue-1", "FER-1", "2026-07-20T20:00:00.000Z"); + const fake = linear({ + [config.stateIds.backlog]: { + revisions: [duplicate, duplicate], + truncated: false, + }, + [config.stateIds.open]: { + revisions: [duplicate, duplicate], + truncated: false, + }, + }); + const output = await new InngestTestEngine({ + function: poller(fake.service), + events: [pollEvent()], + }).execute(); + + expect(output.result).toMatchObject({ + observed: 2, + revisions: 1, + readinessChecks: 1, + }); + }); + + it("returns without a send step when every observed state is empty", async () => { + const output = await new InngestTestEngine({ + function: poller(linear({}).service), + events: [pollEvent()], + }).execute(); + + expect(output.result).toEqual({ outcome: "empty", observed: 0 }); + expect(output.ctx.step.sendEvent).not.toHaveBeenCalled(); + }); + + it.each([ + ["Backlog", config.stateIds.backlog], + ["Open", config.stateIds.open], + ])("fails the whole poll when the %s result is truncated", async (_name, stateId) => { + const fake = linear({ + [stateId]: { revisions: [], truncated: true }, + }); + const output = await new InngestTestEngine({ + function: poller(fake.service), + events: [pollEvent()], + }).execute(); + + expect(output.error).toMatchObject({ + message: expect.stringContaining(`${LINEAR_ISSUE_POLL_LIMIT}-issue limit`), + }); + expect(output.ctx.step.sendEvent).not.toHaveBeenCalled(); + }); + + it("rejects incomplete or ambiguous poller configuration", () => { + const fake = linear({}); + + expect(() => + createLinearIssuePoller({ + client: client(), + linear: fake.service, + config: { ...config, projectId: "" }, + }), + ).toThrow(/projectId/); + expect(() => + createLinearIssuePoller({ + client: client(), + linear: fake.service, + config: { + ...config, + stateIds: { + backlog: config.stateIds.backlog, + open: config.stateIds.backlog, + }, + }, + }), + ).toThrow(/State IDs must be unique/); + }); +}); + +function eventIds(events: unknown): unknown[] { + if (!Array.isArray(events)) throw new Error("Expected sent events"); + return events.map((event) => { + if (!event || typeof event !== "object" || !("id" in event)) { + throw new Error("Expected event ID"); + } + return event.id; + }); +} diff --git a/lib/linear-automation/issue-poller.ts b/lib/linear-automation/issue-poller.ts new file mode 100644 index 0000000..dc4e2d8 --- /dev/null +++ b/lib/linear-automation/issue-poller.ts @@ -0,0 +1,143 @@ +import { cron, type Inngest, type InngestFunction } from "inngest"; +import { z } from "zod"; +import { createLinearIssueReadinessCheckRequestedEvent } from "./events/linear-readiness-events.ts"; +import { + createLinearIssueRevisionObservedEvent, + LinearPollRequestedEvent, +} from "./events/linear-revision-events.ts"; +import { LinearError } from "../linear/error.ts"; +import type { LinearService } from "../linear/client.ts"; +import type { LinearIssueRevision, ListIssueRevisionsResult } from "../linear/types.ts"; + +export const LINEAR_ISSUE_POLL_FUNCTION_ID = "poll-linear-issues-v2"; +export const LINEAR_ISSUE_POLL_RETRIES = 3; +export const LINEAR_ISSUE_POLL_CRON = "* * * * *"; +export const LINEAR_ISSUE_POLL_LIMIT = 250; +export const LINEAR_ISSUE_LIST_STEP_ID = "list-linear-issue-observations-v2"; +export const LINEAR_ISSUE_SEND_STEP_ID = "send-linear-issue-observations-v2"; + +const LinearIssuePollerConfigSchema = z + .object({ + teamId: z.string().trim().min(1), + projectId: z.string().trim().min(1), + stateIds: z + .object({ + backlog: z.string().trim().min(1), + open: z.string().trim().min(1).optional(), + }) + .strict(), + }) + .strict() + .refine((config) => !config.stateIds.open || config.stateIds.open !== config.stateIds.backlog, { + path: ["stateIds"], + message: "State IDs must be unique", + }); + +export type LinearIssuePollerLinear = Pick; + +export type LinearIssuePollerConfig = Readonly>; + +type ListedState = Readonly<{ + kind: "revision" | "readiness-check"; + stateId: string; + result: ListIssueRevisionsResult; +}>; + +export function createLinearIssuePoller(input: { + client: Inngest.Any; + linear: LinearIssuePollerLinear; + config: LinearIssuePollerConfig; +}): InngestFunction.Any { + const config = LinearIssuePollerConfigSchema.parse(input.config); + + return input.client.createFunction( + { + id: LINEAR_ISSUE_POLL_FUNCTION_ID, + concurrency: 1, + retries: LINEAR_ISSUE_POLL_RETRIES, + triggers: [cron(LINEAR_ISSUE_POLL_CRON), LinearPollRequestedEvent], + }, + async ({ event, step }) => { + const states = observedStates(config); + const listed: ListedState[] = await step.run(LINEAR_ISSUE_LIST_STEP_ID, () => + Promise.all( + states.map(async (state) => ({ + ...state, + result: await input.linear.listIssueRevisions({ + teamId: config.teamId, + projectId: config.projectId, + stateId: state.stateId, + limit: LINEAR_ISSUE_POLL_LIMIT, + }), + })), + ), + ); + const truncated = listed.find((state) => state.result.truncated); + if (truncated) { + throw new LinearError( + "incomplete", + [ + `Linear ${truncated.kind} poll for state ${truncated.stateId}`, + `exceeded its ${LINEAR_ISSUE_POLL_LIMIT}-issue limit.`, + ].join(" "), + ); + } + + const revisions = uniqueRevisions( + listed + .filter((state) => state.kind === "revision") + .flatMap((state) => state.result.revisions), + ); + const readinessChecks = uniqueRevisions( + listed + .filter((state) => state.kind === "readiness-check") + .flatMap((state) => state.result.revisions), + ); + if (revisions.length === 0 && readinessChecks.length === 0) { + return { outcome: "empty" as const, observed: 0 }; + } + + const events = [ + ...revisions.map((revision) => + createLinearIssueRevisionObservedEvent({ + issueId: revision.id, + issueIdentifier: revision.identifier, + updatedAt: revision.updatedAt, + }), + ), + ...readinessChecks.map((revision) => + createLinearIssueReadinessCheckRequestedEvent( + { + issueId: revision.id, + issueIdentifier: revision.identifier, + }, + event.id, + ), + ), + ]; + await step.sendEvent(LINEAR_ISSUE_SEND_STEP_ID, events); + return { + outcome: "observed" as const, + observed: events.length, + revisions: revisions.length, + readinessChecks: readinessChecks.length, + eventIds: events.map((event) => event.id), + }; + }, + ); +} + +function observedStates( + config: LinearIssuePollerConfig, +): ReadonlyArray> { + return [ + { kind: "revision", stateId: config.stateIds.backlog }, + ...(config.stateIds.open + ? [{ kind: "readiness-check" as const, stateId: config.stateIds.open }] + : []), + ]; +} + +function uniqueRevisions(revisions: readonly LinearIssueRevision[]): LinearIssueRevision[] { + return [...new Map(revisions.map((revision) => [revision.id, revision])).values()]; +} diff --git a/lib/linear-automation/readiness-router.test.ts b/lib/linear-automation/readiness-router.test.ts index 9382001..676f1fd 100644 --- a/lib/linear-automation/readiness-router.test.ts +++ b/lib/linear-automation/readiness-router.test.ts @@ -1,6 +1,10 @@ import { Inngest } from "inngest"; import { InngestTestEngine, mockCtx } from "@inngest/test"; import { describe, expect, it, vi } from "vitest"; +import { + createLinearIssueReadinessCheckRequestedEvent, + LinearIssueReadinessCheckRequestedEvent, +} from "./events/linear-readiness-events.ts"; import { createLinearIssueRevisionObservedEvent, LinearIssueRevisionObservedEvent, @@ -147,6 +151,16 @@ function revisionEvent(updatedAt = UPDATED_AT) { }); } +function readinessCheckEvent(pollCycleId = "poll-cycle-1") { + return createLinearIssueReadinessCheckRequestedEvent( + { + issueId: "issue-1", + issueIdentifier: "FER-225", + }, + pollCycleId, + ); +} + function sentEvent(output: Awaited>) { const call = vi.mocked(output.ctx.step.sendEvent).mock.calls[0]; if (!call) throw new Error("Expected a sent work event"); @@ -166,7 +180,7 @@ describe("Linear readiness router", () => { id: LINEAR_READINESS_ROUTER_FUNCTION_ID, concurrency: 1, retries: LINEAR_READINESS_ROUTER_RETRIES, - triggers: [LinearIssueRevisionObservedEvent], + triggers: [LinearIssueRevisionObservedEvent, LinearIssueReadinessCheckRequestedEvent], }); }); @@ -210,9 +224,10 @@ describe("Linear readiness router", () => { ] as const)("refetches before an enabled %s dispatch", async (route, labelId, eventName) => { const current = issueContext({ stateId: readiness.stateIds.open, actionLabelId: labelId }); const linear = fakeLinear(current, current); + const event = readinessCheckEvent(); const output = await new InngestTestEngine({ function: router(linear.service), - events: [revisionEvent()], + events: [event], }).execute(); expect(output.error).toBeUndefined(); @@ -222,7 +237,21 @@ describe("Linear readiness router", () => { LINEAR_READINESS_CONFIRM_STEP_ID, expect.any(Function), ); - expect(sentEvent(output)).toMatchObject({ name: eventName }); + expect(sentEvent(output)).toMatchObject({ + name: eventName, + data: { causationEventId: event.id }, + }); + }); + + it("does not turn an Open readiness check into a new Backlog triage request", async () => { + const linear = fakeLinear(issueContext()); + const output = await new InngestTestEngine({ + function: router(linear.service), + events: [readinessCheckEvent()], + }).execute(); + + expect(output.result).toMatchObject({ outcome: "ignore", reason: "not-open" }); + expect(output.ctx.step.sendEvent).not.toHaveBeenCalled(); }); it.each(["2026-07-19T00:59:00.000Z", "2026-07-19T01:01:00.000Z"])( @@ -268,7 +297,7 @@ describe("Linear readiness router", () => { ...readiness, enabledRoutes: { ...readiness.enabledRoutes, spec: false }, }), - events: [revisionEvent()], + events: [readinessCheckEvent()], }).execute(); expect(output.result).toMatchObject({ outcome: "wait", reason: "route-disabled" }); @@ -285,7 +314,7 @@ describe("Linear readiness router", () => { const linear = fakeLinear(initial, changed); const output = await new InngestTestEngine({ function: router(linear.service), - events: [revisionEvent()], + events: [readinessCheckEvent()], }).execute(); expect(output.result).toEqual({ @@ -297,6 +326,39 @@ describe("Linear readiness router", () => { expect(output.ctx.step.sendEvent).not.toHaveBeenCalled(); }); + it("rechecks blocker truth without relying on the blocked issue revision", async () => { + const blocked = issueContext({ + stateId: readiness.stateIds.open, + actionLabelId: readiness.agentActionLabelIds.spec, + blockerStateId: readiness.stateIds.inProgress, + }); + const unblocked = issueContext({ + stateId: readiness.stateIds.open, + actionLabelId: readiness.agentActionLabelIds.spec, + blockerStateId: readiness.stateIds.done, + }); + const blockedOutput = await new InngestTestEngine({ + function: router(fakeLinear(blocked).service), + events: [readinessCheckEvent("poll-cycle-1")], + }).execute(); + const unblockedOutput = await new InngestTestEngine({ + function: router(fakeLinear(unblocked, unblocked).service), + events: [readinessCheckEvent("poll-cycle-2")], + }).execute(); + const repeatedOutput = await new InngestTestEngine({ + function: router(fakeLinear(unblocked, unblocked).service), + events: [readinessCheckEvent("poll-cycle-3")], + }).execute(); + + expect(blockedOutput.result).toMatchObject({ outcome: "wait", reason: "blocked" }); + expect(blockedOutput.ctx.step.sendEvent).not.toHaveBeenCalled(); + expect(unblockedOutput.result).toMatchObject({ outcome: "dispatched", route: "spec" }); + expect(sentEvent(unblockedOutput).id).toBe(sentEvent(repeatedOutput).id); + expect(sentEvent(unblockedOutput).data.causationEventId).not.toBe( + sentEvent(repeatedOutput).data.causationEventId, + ); + }); + it("ignores a claimed issue without reusing an earlier route", async () => { const linear = fakeLinear(issueContext({ stateId: readiness.stateIds.inProgress })); const output = await new InngestTestEngine({ diff --git a/lib/linear-automation/readiness-router.ts b/lib/linear-automation/readiness-router.ts index 76b05c5..2c7f71b 100644 --- a/lib/linear-automation/readiness-router.ts +++ b/lib/linear-automation/readiness-router.ts @@ -1,8 +1,11 @@ import type { Inngest, InngestFunction } from "inngest"; import { + LINEAR_ISSUE_READINESS_CHECK_EVENT_NAME, + LinearIssueReadinessCheckRequestedEvent, +} from "./events/linear-readiness-events.ts"; +import { + LINEAR_ISSUE_REVISION_EVENT_NAME, LinearIssueRevisionObservedEvent, - linearIssueRevisionEventId, - type LinearIssueRevisionData, } from "./events/linear-revision-events.ts"; import { createWorkRequestedEvent } from "./events/work-events.ts"; import { @@ -38,6 +41,12 @@ type LoadedReadiness = | Readonly<{ kind: "current"; observed: ObservedReadiness }> | Readonly<{ kind: "stale"; issueId: string }>; +type ReadinessObservation = Readonly<{ + issueId: string; + issueIdentifier: string; + expectedUpdatedAt?: string; +}>; + export function createLinearReadinessRouter(input: { client: Inngest.Any; linear: LinearReadinessRouterLinear; @@ -50,11 +59,22 @@ export function createLinearReadinessRouter(input: { id: LINEAR_READINESS_ROUTER_FUNCTION_ID, concurrency: 1, retries: LINEAR_READINESS_ROUTER_RETRIES, - triggers: [LinearIssueRevisionObservedEvent], + triggers: [LinearIssueRevisionObservedEvent, LinearIssueReadinessCheckRequestedEvent], }, async ({ event, step }) => { + const observation: ReadinessObservation = + event.name === LINEAR_ISSUE_REVISION_EVENT_NAME + ? { + issueId: event.data.issueId, + issueIdentifier: event.data.issueIdentifier, + expectedUpdatedAt: event.data.updatedAt, + } + : { + issueId: event.data.issueId, + issueIdentifier: event.data.issueIdentifier, + }; const loaded = await step.run(LINEAR_READINESS_LOAD_STEP_ID, () => - loadReadiness(input.linear, event.data, config.readiness), + loadReadiness(input.linear, observation, config.readiness), ); if (loaded.kind === "stale") { return { @@ -65,6 +85,18 @@ export function createLinearReadinessRouter(input: { } const { observed } = loaded; + if ( + event.name === LINEAR_ISSUE_READINESS_CHECK_EVENT_NAME && + observed.decision.kind === "dispatch" && + observed.decision.route === "triage" + ) { + return { + outcome: "ignore" as const, + reason: "not-open" as const, + issueId: observed.issueId, + snapshotGeneration: observed.decision.snapshotGeneration, + }; + } if (observed.decision.kind !== "dispatch") { return { outcome: observed.decision.kind, @@ -78,7 +110,7 @@ export function createLinearReadinessRouter(input: { observed.decision.route === "triage" ? observed : await step.run(LINEAR_READINESS_CONFIRM_STEP_ID, async () => { - const confirmed = await loadReadiness(input.linear, event.data, config.readiness); + const confirmed = await loadReadiness(input.linear, observation, config.readiness); return confirmed.kind === "current" ? confirmed.observed : null; }); if (!ready) { @@ -100,7 +132,7 @@ export function createLinearReadinessRouter(input: { const request = createWorkRequestedEvent(route, { issueId: ready.issueId, issueIdentifier: ready.issueIdentifier, - causationEventId: linearIssueRevisionEventId(event.data), + causationEventId: event.id, snapshotGeneration: ready.decision.snapshotGeneration, }); await step.sendEvent(LINEAR_READINESS_SEND_STEP_ID, request); @@ -118,23 +150,23 @@ export function createLinearReadinessRouter(input: { async function loadReadiness( linear: LinearReadinessRouterLinear, - revision: LinearIssueRevisionData, + observation: ReadinessObservation, config: LinearReadinessConfig, ): Promise { - const context = await linear.getIssueContext(revision.issueId); - if (context.id !== revision.issueId) { + const context = await linear.getIssueContext(observation.issueId); + if (context.id !== observation.issueId) { throw new LinearError( "invalid-response", - `Linear readiness read returned issue ${context.id}, expected ${revision.issueId}.`, + `Linear readiness read returned issue ${context.id}, expected ${observation.issueId}.`, ); } - if (context.identifier !== revision.issueIdentifier) { + if (context.identifier !== observation.issueIdentifier) { throw new LinearError( "invalid-response", - `Linear readiness read returned ${context.identifier}, expected ${revision.issueIdentifier}.`, + `Linear readiness read returned ${context.identifier}, expected ${observation.issueIdentifier}.`, ); } - if (context.updatedAt !== revision.updatedAt) { + if (observation.expectedUpdatedAt && context.updatedAt !== observation.expectedUpdatedAt) { return { kind: "stale", issueId: context.id }; } return { diff --git a/lib/linear-automation/worker.test.ts b/lib/linear-automation/worker.test.ts index 507b0f1..d212f01 100644 --- a/lib/linear-automation/worker.test.ts +++ b/lib/linear-automation/worker.test.ts @@ -7,16 +7,14 @@ import { import { describe, expect, it, vi } from "vitest"; import type { Agent } from "../agent/contract.ts"; import type { LinearAutomationSettings } from "./config.ts"; -import { - LINEAR_BACKLOG_POLL_FUNCTION_ID, - type LinearBacklogPollerLinear, -} from "./backlog-poller.ts"; +import { LINEAR_ISSUE_POLL_FUNCTION_ID, type LinearIssuePollerLinear } from "./issue-poller.ts"; import { createLinearAutomationFunctions, LINEAR_AUTOMATION_APP_ID, LINEAR_AUTOMATION_ENABLED_ROUTES, LINEAR_AUTOMATION_MAX_WORKER_CONCURRENCY, linearAutomationCodexEnvironment, + linearAutomationObservedStateIds, parseLinearAutomationWorkerEnvironment, startLinearAutomationWorker, verifyLinearAutomationCodexAuthentication, @@ -65,7 +63,7 @@ function app() { ensureBlockedByRelation: never, updateIssueLabels: never, updateIssueState: never, - } satisfies LinearTriageService & LinearBacklogPollerLinear; + } satisfies LinearTriageService & LinearIssuePollerLinear; const agent = { name: "codex", run: async () => { @@ -206,7 +204,7 @@ describe("Linear automation worker", () => { const functions = app().functions; expect(functions.map((fn) => fn.opts.id)).toEqual([ - LINEAR_BACKLOG_POLL_FUNCTION_ID, + LINEAR_ISSUE_POLL_FUNCTION_ID, LINEAR_READINESS_ROUTER_FUNCTION_ID, LINEAR_TRIAGE_FUNCTION_ID, ]); @@ -216,6 +214,23 @@ describe("Linear automation worker", () => { implement: false, }); expect(app().readiness.enabledRoutes).toBe(LINEAR_AUTOMATION_ENABLED_ROUTES); + expect(linearAutomationObservedStateIds(app().readiness)).toEqual({ + backlog: settings.readiness.stateIds.backlog, + }); + }); + + it("observes Open only when a composed route can consume it", () => { + const readiness = app().readiness; + + expect( + linearAutomationObservedStateIds({ + ...readiness, + enabledRoutes: { ...readiness.enabledRoutes, spec: true }, + }), + ).toEqual({ + backlog: settings.readiness.stateIds.backlog, + open: settings.readiness.stateIds.open, + }); }); it("reports liveness separately from Connect readiness and closes cleanly", async () => { diff --git a/lib/linear-automation/worker.ts b/lib/linear-automation/worker.ts index 0282d08..7b94ba0 100644 --- a/lib/linear-automation/worker.ts +++ b/lib/linear-automation/worker.ts @@ -13,7 +13,11 @@ import { import { z } from "zod"; import type { Agent } from "../agent/contract.ts"; import { resolveLinearAutomationSettings, type LinearAutomationSettings } from "./config.ts"; -import { createLinearBacklogPoller, type LinearBacklogPollerLinear } from "./backlog-poller.ts"; +import { + createLinearIssuePoller, + type LinearIssuePollerConfig, + type LinearIssuePollerLinear, +} from "./issue-poller.ts"; import { createLinearReadinessRouter } from "./readiness-router.ts"; import type { LinearReadinessConfig } from "./readiness.ts"; import { createLinearTriageFunction, type LinearTriageService } from "./triage-consumer.ts"; @@ -175,7 +179,7 @@ async function checkCodexLogin( export function createLinearAutomationFunctions(input: { client: Inngest.Any; - linear: LinearTriageService & LinearBacklogPollerLinear; + linear: LinearTriageService & LinearIssuePollerLinear; agent: Agent; settings: LinearAutomationSettings; }): LinearAutomationFunctions { @@ -184,13 +188,13 @@ export function createLinearAutomationFunctions(input: { ...input.settings.readiness, enabledRoutes: LINEAR_AUTOMATION_ENABLED_ROUTES, }); - const poller = createLinearBacklogPoller({ + const poller = createLinearIssuePoller({ client: input.client, linear: input.linear, config: { teamId: readiness.teamId, projectId: readiness.projectId, - stateId: readiness.stateIds.backlog, + stateIds: linearAutomationObservedStateIds(readiness), }, }); const router = createLinearReadinessRouter({ @@ -220,6 +224,17 @@ export function createLinearAutomationFunctions(input: { }); } +export function linearAutomationObservedStateIds( + readiness: LinearReadinessConfig, +): LinearIssuePollerConfig["stateIds"] { + return Object.freeze({ + backlog: readiness.stateIds.backlog, + ...(readiness.enabledRoutes.spec || readiness.enabledRoutes.implement + ? { open: readiness.stateIds.open } + : {}), + }); +} + export async function startLinearAutomationWorker(input: { app: LinearAutomationFunctions; host: string; diff --git a/scripts/smoke-linear-automation.ts b/scripts/smoke-linear-automation.ts index 0dc40e4..f0a47fb 100644 --- a/scripts/smoke-linear-automation.ts +++ b/scripts/smoke-linear-automation.ts @@ -22,10 +22,10 @@ import { type LinearAutomationWorker, } from "../lib/linear-automation/worker.ts"; import { - LINEAR_BACKLOG_POLL_FUNCTION_ID, - LINEAR_BACKLOG_POLL_LIMIT, - type LinearBacklogPollerLinear, -} from "../lib/linear-automation/backlog-poller.ts"; + LINEAR_ISSUE_POLL_FUNCTION_ID, + LINEAR_ISSUE_POLL_LIMIT, + type LinearIssuePollerLinear, +} from "../lib/linear-automation/issue-poller.ts"; import { LINEAR_READINESS_ROUTER_FUNCTION_ID } from "../lib/linear-automation/readiness-router.ts"; import { LINEAR_TRIAGE_FUNCTION_ID, @@ -244,7 +244,7 @@ function issueContext(): LinearIssueContext { }; } -function fakeLinear(): LinearTriageService & LinearBacklogPollerLinear { +function fakeLinear(): LinearTriageService & LinearIssuePollerLinear { return { listIssueRevisions: async (input) => { projection.pollInputs.push(input); @@ -450,7 +450,7 @@ try { "--app-id", LINEAR_AUTOMATION_APP_ID, ]); - assert(registered.includes(LINEAR_BACKLOG_POLL_FUNCTION_ID), "poller registration missing"); + assert(registered.includes(LINEAR_ISSUE_POLL_FUNCTION_ID), "poller registration missing"); assert(registered.includes(LINEAR_READINESS_ROUTER_FUNCTION_ID), "router registration missing"); assert(registered.includes(LINEAR_TRIAGE_FUNCTION_ID), "triage registration missing"); @@ -486,7 +486,7 @@ try { teamId: settings.readiness.teamId, projectId: settings.readiness.projectId, stateId: settings.readiness.stateIds.backlog, - limit: LINEAR_BACKLOG_POLL_LIMIT, + limit: LINEAR_ISSUE_POLL_LIMIT, }), ), "poller used unexpected Linear scope",