onetaskgraph_linear/

lib.rs

//! A read/write source over Linear's published GraphQL API.
//!
//! Linear `Issue` maps to [`Task`], `Project` to [`Project`], `Document` to [`Document`],
//! `IssueLabel` and `ProjectLabel` to [`Label`], and `WorkflowState.name` is preserved
//! while its `type` (`backlog`, `unstarted`, `started`, `completed`, or `canceled`) maps to
//! the normalized status category. Issue `relations`/`inverseRelations` and
//! project relations provide native dependency traversal in both directions.
//!
//! Label, workflow-state, project, and orphan filters are sent in the
//! `issues(filter:)`/`projects(filter:)` variables. Pagination uses Relay `first` and
//! `after`.
//!
//! Every issue, project and document reports its own Linear web address as its
//! [`Location`], as a link rather than a path — the counterpart of a folder of Markdown
//! reporting the path of the file behind an item. It does not replace the `url` field
//! those types already carry; it is the same address said in the shape a reader can act on.
//!
//! # What this source declares, field by field
//!
//! One verdict per field of [`Capabilities`]. A field is *supported and proven* when this
//! source applies it and a shared journey drives it against the real binary; the shared
//! table is `crates/onetaskgraph/tests/e2e/fixtures.rs`, the journeys are beside it, and
//! `every_row_declares_exactly_what_its_plugin_reports` is what keeps this list and
//! [`capabilities`](TaskSource::capabilities) from parting.
//!
//! | Field | Verdict |
//! | --- | --- |
//! | `projects` | **Supported and proven.** `issues(filter:{project:{id:{eq:…}}})`. |
//! | `documents` | **Supported and proven.** Linear's own first-class `Document`, read through `documents(first:,after:,filter:)` and `document(id:)`, written through `documentCreate`/`documentUpdate` and taken back by `documentDelete`. See the ruling below on what a Linear document cannot hold. |
//! | `comments` | **Supported and proven,** as the issue's own comments: read oldest first through `issue(id:){comments(last:,before:)}`, added with `commentCreate`, edited with `commentUpdate` and removed with `commentDelete` — each of the last two only once `comment(id:)` has placed the comment on that very issue. See the ruling below on the order and on the author. |
//! | `priority` | **Supported,** as Linear's own `Issue.priority`: read on every issue, written by `issueCreate`/`issueUpdate` through `IssueCreateInput.priority`/`IssueUpdateInput.priority`, and set on its own by an `issueUpdate` carrying nothing else. See the ruling below on the scale. |
//! | `filter_by_priority` | **Supported and proven.** `issues(filter:{priority:{in:[…]}})` over Linear's own `0`–`4` scale, confirmed against each issue read. |
//! | `filter_by_comment_activity` | **Supported and proven.** `comments:{some:{or:[{createdAt:{gte:…}},{updatedAt:{gte:…}}]}}` — the issues with a comment created or last edited at or after the instant, over the same two fields a comment read reports. |
//! | `filter_by_metadata` | **Supported and proven.** `description:{contains:"\"<value>\""}` for each match — the value as every JSON encoder writes it, which a slot holding it contains however it spaces or spells its keys — and every candidate confirmed over the parsed slot, so prose carrying the phrase and a slot holding another value are both kept out. A value with a character an encoder may escape is not sent, and the confirmation decides alone. |
//! | `filter_by_origin` | **Supported and proven,** on the same terms, for the slot's `onetaskgraph.origin`. |
//! | `orphan_tasks` | **Supported and proven.** `issues(filter:{project:{null:true}})`. |
//! | `filter_by_label` | **Supported and proven.** `labels:{some:{name:{eqIgnoreCase:…}}}` for what an item must carry — one per label, gathered under `or:` where any one of them will do — and `labels:{every:{name:{neqIgnoreCase:…}}}` for what it must not. Linear's `StringComparator` has no case-insensitive list operator; see the note beside `filter`. |
//! | `filter_by_status` | **Supported and proven,** and spelled twice. An issue narrows with `state:{type:{in:[…]}}` over `WorkflowState.type`; a project narrows with `status:{type:{in:[…]}}` over `ProjectStatusType`, a different member of a different filter over a different vocabulary. See the ruling below. |
//! | `search_title` | **Supported and proven.** A task query's text is `title:{containsIgnoreCase:…}`, every candidate confirmed by the contract's case-insensitive substring rule; a project or document query's text is applied by that same rule over the page Linear answered. |
//! | `search_content` | **Supported and proven,** on the same terms, `description:{containsIgnoreCase:…}`, confirmed over the visible content — the trailing metadata slot Linear's comparator also reads is not part of what the rule confirms. A `title-or-content` search sends the two under one `or`. |
//! | `task_dependencies` | **Supported and proven,** in both directions: `relations` and `inverseRelations`. |
//! | `project_dependencies` | **Supported and proven,** in both directions, by the project relations of the same shape. Linear types every one of them `dependency`; see the ruling below on the edge that has no spelling here. |
//! | `max_page_size` | **Supported and proven.** 100; every read pages with Relay `first`/`after`. Linear's connection maximum is 250 and its complexity budget is the tighter bound — see [`MAX_PAGE_SIZE`]. |
//!
//! ## Ruling: the follow-up searches are native, and what each rests on
//!
//! Six predicates a follow-up search sends — metadata, origin, comment activity, priority,
//! and the two text searches — are each sent to Linear as a narrowing of `issues(filter:)`
//! and confirmed in process before a row is returned. Each narrowing is a candidate set that
//! cannot miss a row the contract's predicate keeps, which is what makes sending it sound;
//! the confirmation is what makes the answer exact. Every member below is pinned in
//! `tests/fixtures/schema.graphql`, and each rests on one observation of the real API,
//! against the scratch team `TES` on 2026-10-02, which `drive_follow_ups` in `tests/live.rs`
//! asserts again — the comparators by name, and every narrowing through this source with a
//! decoy whose prose carries the searched phrases:
//!
//! - **`IssueFilter.description.contains` reads the whole stored description, the metadata
//!   slot included, and is case-sensitive.** An issue whose slot held
//!   `"caller.key":"needle-…"` was returned for `contains` of that exact phrase, and of the
//!   `"onetaskgraph.origin":"…"` pair beside it; the same description's prose, upper-cased,
//!   was returned for `containsIgnoreCase` and *not* for `contains`. So a metadata match or an
//!   origin is sent as its value in quotes, `"<value>"` — the bytes any JSON encoder writes a
//!   string as, which a slot holding it contains whether it is the code span this source
//!   writes, the multi-line slot it wrote before, or one a person spaced by hand — and
//!   confirmed over the parsed slot. A value holding a character an encoder may escape is not
//!   sent, and the confirmation decides alone.
//! - **`IssueFilter.title.containsIgnoreCase` and `description.containsIgnoreCase` match
//!   regardless of case** — the title `… Alpha Title` was returned for `alpha TITLE`, and not
//!   for `contains` of it. They are the two text searches; the content search is confirmed
//!   over the visible content, because the comparator also reads the slot.
//! - **`IssueFilter.priority.in` narrows by Linear's own number** — an issue at `2` was
//!   returned for `in:[2]` and not for `in:[3]`.
//! - **`IssueFilter.comments.some` with `createdAt`/`updatedAt` `gte` narrows by a comment's
//!   own times** — an issue whose comment had been edited a moment earlier was returned for an
//!   instant before the edit and not for one a day later, and editing the comment moved its
//!   `updatedAt` while leaving `createdAt`. The comment read reports those same two fields,
//!   so the filter and the contract's rule read one value.
//!
//! ## Ruling: what Linear does to an HTML comment, settled
//!
//! A follow-up tool marks what it writes with HTML comments, and this source keeps its own
//! metadata in one, so what survives is a fact this crate records rather than assumes.
//! Observed on 2026-10-02 against the scratch team `TES`, each written and read back by id,
//! and asserted again — the probe text and its stored form exactly — by `drive_follow_ups` in
//! `tests/live.rs`, a leg of its `real_linear_applies_every_declared_capability_and_leaves_no_residue`:
//!
//! - **A comment's `body` keeps every HTML comment byte for byte** — on one line or across
//!   several, a bare `-->` closing line, domain-like text and JSON included.
//! - **An issue's `description` and a document's `content` do not.** Linear stores both as
//!   Markdown and normalizes the text *inside* an HTML comment exactly as it normalizes prose:
//!   a domain-like token or a URL is autolinked — `example.com` comes back
//!   `[example.com](<http://example.com>)`, and a key such as `caller.live` likewise — `[` and
//!   `]` come back `\[` and `\]`, `~` comes back `\~`, `_y_` comes back `*y*`, the backslash
//!   of `\"` is dropped, a backslash before a letter is doubled, and a line opening `-->` comes
//!   back `\-->`. Text with none of those in it — `{"k":"v"}`, `caller.key`, `sha256:…`,
//!   `gh:I_kwDO…` — comes back as written. An autolink can run on past the token, swallowing
//!   what follows it up to the next delimiter.
//! - **One thing in those two fields comes back byte for byte: a code span.** An HTML comment
//!   on one line whose payload is inside backticks — ``<!-- probe `{…}` -->`` — came back
//!   identical with every one of the payloads above inside it, and re-writing what Linear
//!   handed back changed nothing more. So this source writes its own slot that way (see
//!   `METADATA_OPEN_SPAN`), and a marker meant to survive an issue's description or a
//!   document's content belongs in one too.
//!
//! ## Ruling: a Linear document carries no label, and that is Linear's
//!
//! Unlike the two searches above, this one *is* a property of the remote service. The
//! types of Linear's published schema carrying a `labels` field are `Issue`, `Project`,
//! `Team`, `Initiative` and `Organization`; `Document` is not among them, re-observed
//! 2026-09-01 and pinned in `tests/fixtures/schema.graphql`. So this source reports a
//! document's labels as none and **refuses by name** a document write carrying one, rather
//! than dropping it or standing a slot up beside a first-class type. The shared journey
//! table's row says so, and the shared document journeys drive that claim.
//!
//! Two predicates therefore reach a fetched page rather than the `documents(filter:)`
//! variables, and both are still *applied* — which is what `Native` means here, and why
//! the declaration stays honest. Labels, for the reason above. And orphans, because
//! `DocumentFilter.project` is a `ProjectFilter` where `IssueFilter.project` is a
//! `NullableProjectFilter`: only the nullable one carries `null:`, so Linear cannot be
//! asked for the documents belonging to no project. The page-by-page walk asks for only
//! what is still owed, so neither predicate can make a read return more than the caller
//! asked for, and neither can drop a document the walk already fetched.
//!
//! ## Ruling: a comment is read backwards, and its author is Linear's to record
//!
//! **The order.** The contract owes a task's comments oldest first, across pages, and Linear's
//! `Issue.comments` takes no sort direction — only `orderBy`, whose members are `createdAt`
//! (the default) and `updatedAt`. Linear's pagination documentation says results are "ordered
//! by `createdAt`" and that "to get most recently updated resources, you can alternatively
//! order by `updatedAt`", which reads that ordering as newest first. So this source walks the
//! connection from its far end: `last` with `before`, each page reversed, the next page's
//! cursor being `startCursor` while `hasPreviousPage` holds. Reversing within a page and
//! walking backwards across them is what makes the whole walk oldest first rather than each
//! page alone. **That direction is inferred from the documentation's wording rather than
//! observed against the real API,** which is the one reading here a live run has not yet
//! confirmed; if Linear is found to list oldest first, the correction is this walk's
//! direction and nothing else.
//!
//! **The author.** Linear records the user whose credential made the request as a comment's
//! author, and this source authenticates with an API key. `CommentCreateInput.createAsUser`
//! exists but is, in Linear's own words, "only available to OAuth applications creating
//! comments in `actor=app` mode", which a key is not. So a comment carrying an author is
//! **refused before any request is sent**, naming why and what to do instead, rather than
//! posted under a name other than the one it was given. An author read back is the user's
//! `displayName`, which Linear keeps unique within a workspace, and is absent when Linear
//! names no user — a comment an integration or a bot wrote.
//!
//! **What "no such comment" means.** An edit or a removal first asks `comment(id:)` which
//! issue the comment is on, and answers "no such comment" — no mutation sent — unless it is
//! the task's own issue: a comment on another issue, on no issue at all, or trashed, is not a
//! comment this task has. The body is Linear's `body`, which its schema describes as markdown
//! derived from a rich-text document, so what an add or an edit answers with is what Linear
//! now holds rather than an echo of what was sent.
//!
//! ## Ruling: a project's filter is not an issue's, and neither is its status
//!
//! Linear's `IssueFilter` and `ProjectFilter` read as one filter over two kinds of row.
//! They are two input types, and this source built one object for both until 2026-09-04,
//! which put two members into `projects(filter:)` that Linear does not have there. It
//! refused the first outright — `Field "team" is not defined by type "ProjectFilter". Did
//! you mean "lead"?` — and would have refused the second next.
//!
//! A project has no team; it has the teams it is accessible from, so the configured team
//! reaches `accessibleTeams:{some:{key:{eqIgnoreCase:…}}}`. And a project's status is not
//! an issue's state: the counterpart of `IssueFilter.state` is `ProjectFilter.status`,
//! while `ProjectFilter.state` exists and is a bare `StringComparator` over something else.
//! The two do not even share a vocabulary — `ProjectStatus.type` is the `ProjectStatusType`
//! enum, `backlog`, `planned`, `started`, `paused`, `completed`, `canceled`, where a
//! workflow state is `backlog`, `unstarted`, `started`, `completed`, `canceled`, `triage`.
//! So `planned` is where `unstarted` would be, `paused` reads as in progress and has no
//! issue counterpart, and a filter spelled in the other level's words matches nothing while
//! being refused by nothing.
//!
//! **Neither of those could be caught by reading a document, and that is the general
//! lesson.** A filter is built at runtime and handed over as `$filter`, so it appears in no
//! operation this crate declares, and the two pinned-schema checks that parse those
//! operations could not see it — Linear was the only reader, one refusal per round trip.
//! `every_variables_object_this_source_sends_conforms_to_the_pinned_schema` closes that:
//! it drives this source's whole surface, records what really went out, and walks every
//! variables object against the pinned type of the argument it stands at.
//!
//! ## Ruling: a Linear project relation is always an ordering
//!
//! This one is Linear's too, and the validator says so in as many words. Asked on
//! 2026-09-04 for a project relation typed `related` — and separately `blocks` and
//! `dependsOn` — the real API refused each with `Argument Validation Error` and
//! `constraints: {"isEnum": "type must be one of the following values: dependency"}`. That
//! enumeration has one member and it is a timeline dependency, which is why the input
//! carries an anchor at each end at all.
//!
//! So a project edge carrying no ordering has nowhere here to land, and this source
//! **refuses it by name** before the write rather than sending a value Linear will reject
//! or quietly promoting it to a dependency it does not mean. `DependencyKind::Related`
//! keeps its issue-level spelling, `related`, because `IssueRelationCreateInput` really
//! does take it: the two relations are different relations with different vocabularies,
//! and each level's read accepts only its own.
//!
//! Which end of a project relation waits is carried by the two anchors and not by the two
//! id slots — measured, not reasoned, from Linear's own `ProjectFilter.hasBlockedByRelations`
//! against relations written both ways round. `tests/fixtures/README.md` records the whole
//! probe, and `write_relations` records why the pair this source sends is the oriented one.
//!
//! Caller metadata is canonical JSON in a trailing
//! ``<!-- onetaskgraph.metadata `…` -->`` Markdown comment in the item's description, on one
//! line with the JSON in a code span — the one spelling Linear keeps byte for byte, see the
//! ruling above; the multi-line spelling this source wrote before is still read. The visible
//! description is returned unchanged without that slot. Writes put the same canonical
//! encoding back beside the visible description, and use Linear issue/project relations for
//! same-source dependencies. Only cross-source far ends use the reserved
//! `onetaskgraph.depends_on` metadata key.
//!
//! ## Ruling: a status is written by `status_mapping`, and by type where it names none
//!
//! A Linear team can hold several workflow states of one type — `Todo` and `Queued` are both
//! `unstarted`, `Proposed` and `Backlog` both `backlog` — and a category says nothing about
//! which of them it means. `status_mapping` is what does: a category it names is written as
//! the state of exactly that name by every write — `set_task_status`, the targeted update and
//! `write_task`, so `task create` and every copy — and never as another state of the same
//! type. A name the team does not have is refused before any write, naming the state, the team
//! and the category, after one read of the team's states through
//! [`graphql::TEAM_WORKFLOW_STATES`]. A category it sets to `null` is refused as disabled
//! before any request. Two categories mapped to one name are refused when the configuration is
//! read, because that state could read back as only one of them.
//!
//! On a read, an issue at a state the mapping names is that category under the state's own
//! name; every other state reads by its type, as it always has — the review states only
//! people write, `Triage` among them, which nothing here ever writes. `filter_by_status`
//! returns exactly the issues whose status reads as each category asked for: those at the
//! state the mapping names for it, and those at an unmapped state whose type falls back to
//! it — never one at a state mapped to another category. So `--status queued` returns the
//! issues at `Queued` and never one at `Todo`; `--status in-progress` returns those at
//! `In Progress` and at an unmapped `started` state such as `In Review`, and never one at
//! `Needs Attention` when that is mapped to `unknown`; and `--status unknown` also returns
//! those at a state of a type none of the five categories stands for, `Triage` among them.
//!
//! A category the mapping does not mention keeps the behaviour of a source without the key
//! exactly: `set_task_status` and the targeted update write the team's first state of the
//! type `workflow_state_types` gives — `backlog`, `todo`, `in-progress`, `done` and
//! `cancelled` — and refuse `draft`, `queued` and `unknown`, which no type stands for, before
//! any request; `write_task` resolves the state by the status's own name, as it always has. A
//! task already in the category asked for keeps the state it is in and nothing is written:
//! moving an issue from `In Review` to `In Progress` because it was asked to be in progress is
//! a change nobody asked for.
//!
//! ## Ruling: `project` scopes a source to one project
//!
//! With `project` set, every issue read carries `project:{id:{eq:…}}` beside the team, a
//! project read carries `id:{eq:…}` and a document read the same project, and a read by id
//! of anything filed elsewhere answers as no such item — so a status, a content, a metadata
//! or a comment write to it is answered the same way. A task or a document written with no
//! project is placed in that one, and one naming another is refused naming both. A project
//! write other than to that project itself is refused: a project this source created would be
//! one none of its reads could find.
//!
//! ## Ruling: a narrow metadata write moves only the slot, and a task carries delivery
//!
//! `set_task_metadata`, `set_project_metadata` and `set_document_metadata` read the item and
//! send one update of its long-form field — `description`, `description` and `content` —
//! that differs from what Linear holds only inside the trailing metadata slot: every byte
//! above it is kept as it was. A key already holding the value sends nothing. The answer is
//! the item read back. `set_task_rendering` and `set_document_rendering` replace the content
//! and the slot's `onetaskgraph.template` entry together in one such update, every other slot
//! entry kept; this source keeps no template answers. So a copy's `onetaskgraph.copies` link
//! is recorded on a Linear item rather than reported unrecorded.
//!
//! A task's `delivers` and `delivered_by` live in that same slot under
//! `onetaskgraph.delivers` and `onetaskgraph.delivered_by`, each a list of qualified ids,
//! with the shape and the rules the GitHub Projects plugin keeps: neither may name the task
//! itself or name one task twice, which is refused by name before anything is sent; a write
//! lands the typed lists in place of any caller metadata of those names; and
//! `set_delivered_by` is one update of the slot. A project or a document naming either key is
//! refused, because only a task delivers or is delivered.
//!
//! ## Ruling: a priority is Linear's own, and content shares a field with the slot
//!
//! A task's priority is `Issue.priority`, on Linear's scale: `0` none, `1` urgent, `2` high,
//! `3` normal — this contract's `medium` — and `4` low. Linear declares the field `Float!`
//! while `IssueCreateInput.priority` and `IssueUpdateInput.priority` are `Int`, so a read
//! accepts `2` and `2.0` alike and refuses anything that is not one of the five as a
//! malformed response naming the field. A copy sends it on a create and on an update, `0`
//! included, so a task moved back to no priority is not left holding its old one.
//! `set_task_priority` reads the issue first — no such issue, or a trashed one, is `None`
//! with nothing written — then sends `issueUpdate` with `priority` alone and answers with
//! the priority the mutation's own payload reports.
//!
//! `set_task_content` sends `issueUpdate` with `description` alone, and that description is
//! the given content followed by the issue's metadata slot exactly as it was stored, so the
//! slot, and every key in it, is untouched. What a later read reports as the content is the
//! given bytes, trailing whitespace included: a read of an issue carrying a slot takes off only
//! the one blank line that sets the slot off, and a write whose content would not read back as
//! itself is refused before it is sent.
//!
//! Fixture provenance is recorded in `tests/fixtures/README.md`. The live journey in
//! `tests/live.rs` drives every field of the table above against Linear itself: it builds its own fixture
//! on the scratch team `LINEAR_WRITE_TEAM` names — two projects, one issue filed under
//! each, one filed under neither, two labels and two workflow states — because that shape
//! is what tells an honoured predicate from an ignored one, and a workspace where every
//! issue carries the label answers a filter the same way either way. Everything the lane
//! creates it deletes whether its assertions passed or failed, and it clears residue named
//! the way it names its own before it starts. A failed live cleanup is reported as a test
//! failure and may require manual deletion from that scratch team.
#![deny(missing_docs)]

use chrono::{DateTime, Utc};
use onetaskgraph_plugin_api::{
    Capabilities, Comment, CommentBody, Cursor, DependencyEdge, DependencyEndpoint, DependencyKind,
    DependencySupport, Direction, Document, DocumentQuery, Health, ItemKind, ItemWrite, Label,
    LabelFilter, Location, MetadataKey, NativeId, NewComment, Page, PageRequest, Priority, Project,
    ProjectFilter, ProjectQuery, Repository, SecretResolver, SourceError, SourceName, SourcePlugin,
    Status, StatusCategory, Support, Task, TaskQuery, TaskRef, TaskSource, TaskUpdate,
    TaskUpdateOutcome, TextFields, TextQuery, UpdatedField, WriteSupport,
};
use schemars::{Schema, schema_for};
use secrecy::{ExposeSecret, SecretString};
use serde::Deserialize;
use serde_json::{Value, json};

/// The plugin kind a `linear` source's `plugin:` field names.
pub const KIND: &str = "linear";

/// The largest page this source will ask Linear for, and the capability it declares.
///
/// **Not Linear's connection maximum, which is 250, because a connection maximum is not
/// the only thing bounding a page.** Linear also scores each document for complexity and
/// refuses one over 10000 with HTTP 400 and `The query is too complex.` — and the
/// `projects` document this source sends scores 17475 at `first: 250`, because its nested
/// `labels` connection, which names no `first` of its own, is charged Linear's default of
/// 50 per node. Measured against the real API on 2026-09-04: the largest `first` that
/// document is accepted at is **143**, exactly, and the filter it carries adds nothing.
/// The `issues` document is accepted at 250, so this is the tighter of the two and a
/// single declared maximum has to be the tighter one.
///
/// 100 rather than 143 because 143 is the cliff. A field added to either selection moves
/// it, and a page size chosen at the edge of a budget nobody here controls fails in the
/// live lane rather than in a check. This leaves 30% of the budget spare.
///
/// Nothing offline can hold this: complexity is scored by Linear's own runtime and appears
/// in no schema, so `every_variables_object_this_source_sends_conforms_to_the_pinned_schema`
/// cannot see it. What guards it is the live journey, which walks a real `projects` page at
/// exactly this size.
pub const MAX_PAGE_SIZE: u32 = 100;
const DEFAULT_ENDPOINT: &str = "https://api.linear.app/graphql";

/// Exact GraphQL query documents issued by this plugin.
///
/// Fixture servers consume these constants so their recognized contract cannot drift
/// from the production requests.
pub mod graphql {
    /// Check the authenticated viewer.
    pub const VIEWER: &str = "query { viewer { id } }";
    /// Fetch one issue.
    pub const ISSUE: &str = "query($id:String!){ issue(id:$id){ id identifier title description url createdAt updatedAt archivedAt state{name type} priority labels{nodes{id name color}} project{id} } }";
    /// Fetch one project.
    pub const PROJECT: &str = "query($id:String!){ project(id:$id){ id name description url createdAt updatedAt archivedAt status{name type} labels{nodes{id name color}} } }";
    /// List issues.
    pub const ISSUES: &str = "query($first:Int!,$after:String,$filter:IssueFilter){ issues(first:$first,after:$after,filter:$filter){ nodes{id identifier title description url createdAt updatedAt state{name type} priority labels{nodes{id name color}} project{id}} pageInfo{hasNextPage endCursor} } }";
    /// List projects.
    pub const PROJECTS: &str = "query($first:Int!,$after:String,$filter:ProjectFilter){ projects(first:$first,after:$after,filter:$filter){ nodes{id name description url createdAt updatedAt status{name type} labels{nodes{id name color}}} pageInfo{hasNextPage endCursor} } }";
    /// List issue labels.
    pub const LABELS: &str = "query($first:Int,$after:String){ issueLabels(first:$first,after:$after){ nodes{id name color} pageInfo{hasNextPage endCursor} } }";
    /// Fetch issue dependency relations.
    pub const ISSUE_RELATIONS: &str = "query($id:String!,$first:Int!,$after:String){ issue(id:$id){ description relations(first:$first,after:$after){nodes{id type relatedIssue{id}} pageInfo{hasNextPage endCursor}} inverseRelations(first:$first,after:$after){nodes{id type issue{id}} pageInfo{hasNextPage endCursor}} } }";
    /// Fetch project dependency relations.
    pub const PROJECT_RELATIONS: &str = "query($id:String!,$first:Int!,$after:String){ project(id:$id){ description relations(first:$first,after:$after){nodes{id type relatedProject{id}} pageInfo{hasNextPage endCursor}} inverseRelations(first:$first,after:$after){nodes{id type project{id}} pageInfo{hasNextPage endCursor}} } }";
    /// Resolve the configured team key to Linear's backend id.
    pub const TEAM: &str =
        "query($key:String!){ teams(filter:{key:{eqIgnoreCase:$key}}){nodes{id}} }";
    /// Resolve an issue workflow-state display name.
    ///
    /// `$team` is an `ID!` and `$name` a `String!` because that is what each one's
    /// *location* declares, not because of what this source passes: both carry a Linear
    /// identifier string. `WorkflowStateFilter.team` is a `NullableTeamFilter`, whose `id`
    /// is an `IDComparator`, whose `eq` is an `ID`; the sibling `name` reaches a
    /// `StringComparator.eqIgnoreCase`, which is a `String`.
    ///
    /// That distinction is what the live lane was refused for on 2026-09-04, with HTTP 400
    /// and `Variable "$team" of type "String!" used in position expecting type "ID".`
    /// GraphQL admits a variable at a location only when the variable's type is the
    /// location's type or that type's non-null form, and `String` is not `ID` however the
    /// value is spelled — so `String!` there fails validation before any field is read,
    /// while `ID!` is the non-null form of the location's own type and is accepted.
    ///
    /// It reached Linear because a variable inside an inline filter literal is not a root
    /// argument, and the pinned-schema checks only compared root arguments. They now walk
    /// into these literals too, so this class of drift fails here rather than in the live
    /// lane.
    pub const ISSUE_STATE: &str = "query($name:String!,$team:ID!){ workflowStates(filter:{name:{eqIgnoreCase:$name},team:{id:{eq:$team}}}){nodes{id}} }";
    /// Find the configured team's workflow states of one `WorkflowState.type`, so a task's
    /// status can be set by category alone.
    ///
    /// `name` is selected beside `id` because the status a narrow status write answers with
    /// is the one Linear now holds, and a category alone does not say which of the team's
    /// states of that type it is. `$type` is a `String!` at `StringComparator.eq`, which is a
    /// `String`, and `$team` an `ID!` for the reason recorded on [`ISSUE_STATE`].
    pub const ISSUE_STATE_OF_TYPE: &str = "query($type:String!,$team:ID!){ workflowStates(filter:{type:{eq:$type},team:{id:{eq:$team}}}){nodes{id name}} }";
    /// Every workflow state of the configured team, with its type.
    ///
    /// What a category `status_mapping` names is written through — the state is found by its
    /// name here, so a name the team lacks is refused naming the state, the team and the
    /// category rather than as an unexplained empty lookup — and what `sources fields` reports
    /// each mapped state against. A team holds a few dozen states at most, inside the one page
    /// Linear answers an unpaged connection with. `$team` is an `ID!` for the reason recorded
    /// on [`ISSUE_STATE`].
    pub const TEAM_WORKFLOW_STATES: &str =
        "query($team:ID!){ workflowStates(filter:{team:{id:{eq:$team}}}){nodes{id name type}} }";
    /// List the workspace's project statuses, so one can be resolved by display name.
    ///
    /// Unlike `teams`, `workflowStates` and the two label connections, Linear's
    /// `projectStatuses` accepts no `filter` argument: asking for one is refused outright
    /// with `Unknown argument "filter" on field "Query.projectStatuses"`. The display name
    /// is therefore matched locally over the whole connection, which a workspace holds few
    /// enough of to answer in one page.
    // llmlint: ignore[changed_behavior_has_e2e] The uncovered case the rule names — a status
    // on a later page — is not a test that is missing but a document this repository has no
    // evidence Linear would accept: `tests/fixtures/schema.graphql` pins `after` alone,
    // because Linear's own refusal is where that correction came from, and its
    // `ProjectStatusConnection` declares `nodes` and no `pageInfo`. Selecting a cursor field
    // to page on would fail `pinned_schema_checks_selected_fields_arguments_and_fixture_keys`
    // here and risk, against Linear, the same `GRAPHQL_VALIDATION_FAILED` this document was
    // changed to stop sending. Reading one page is not what changed either: `teams`,
    // `workflowStates` and `projectLabels` resolve a display name through the same `one_id`
    // over the same unpaged connections, and did before this change. What did change is
    // driven end to end — the CLI journey
    // `linear_project_and_task_copies_write_native_relations_and_record_only_cross_source_edges`
    // copies a project whose status is resolved this way, and
    // `a_project_status_is_matched_locally_because_linear_narrows_that_connection_for_nobody`
    // holds the match, the ambiguity and the absence against a real HTTP server.
    pub const PROJECT_STATUS: &str = "query{ projectStatuses{nodes{id name}} }";
    /// Resolve an issue-label display name.
    pub const ISSUE_LABEL: &str =
        "query($name:String!){ issueLabels(filter:{name:{eqIgnoreCase:$name}}){nodes{id}} }";
    /// Resolve a project-label display name.
    pub const PROJECT_LABEL: &str =
        "query($name:String!){ projectLabels(filter:{name:{eqIgnoreCase:$name}}){nodes{id}} }";
    /// Create an issue.
    pub const ISSUE_CREATE: &str =
        "mutation($input:IssueCreateInput!){ issueCreate(input:$input){success issue{id}} }";
    /// Update an issue.
    pub const ISSUE_UPDATE: &str = "mutation($id:String!,$input:IssueUpdateInput!){ issueUpdate(id:$id,input:$input){success issue{id}} }";
    /// Set an issue's priority on its own, and read back the priority Linear now holds.
    ///
    /// The same `issueUpdate` as [`ISSUE_UPDATE`], selecting `priority` in the payload
    /// because a narrow priority write answers with what the source reads back rather than
    /// an echo of what it sent. A document of its own rather than a wider [`ISSUE_UPDATE`],
    /// so every other issue write keeps asking for exactly what it reads.
    pub const ISSUE_PRIORITY_UPDATE: &str = "mutation($id:String!,$input:IssueUpdateInput!){ issueUpdate(id:$id,input:$input){success issue{id priority}} }";
    /// Create a project.
    pub const PROJECT_CREATE: &str =
        "mutation($input:ProjectCreateInput!){ projectCreate(input:$input){success project{id}} }";
    /// Update a project.
    pub const PROJECT_UPDATE: &str = "mutation($id:String!,$input:ProjectUpdateInput!){ projectUpdate(id:$id,input:$input){success project{id}} }";
    /// Create a native issue dependency.
    pub const ISSUE_RELATION_CREATE: &str = "mutation($input:IssueRelationCreateInput!){ issueRelationCreate(input:$input){success issueRelation{id}} }";
    /// Create a native project dependency.
    pub const PROJECT_RELATION_CREATE: &str = "mutation($input:ProjectRelationCreateInput!){ projectRelationCreate(input:$input){success projectRelation{id}} }";
    /// Delete a native issue dependency before replacing its full edge set.
    pub const ISSUE_RELATION_DELETE: &str =
        "mutation($id:String!){ issueRelationDelete(id:$id){success} }";
    /// Delete a native project dependency before replacing its full edge set.
    pub const PROJECT_RELATION_DELETE: &str =
        "mutation($id:String!){ projectRelationDelete(id:$id){success} }";
    /// Delete an issue, so a copy that could not finish can take back what it created.
    pub const ISSUE_DELETE: &str = "mutation($id:String!){ issueDelete(id:$id){success} }";
    /// Delete a project, for the same reason and on the same terms.
    pub const PROJECT_DELETE: &str = "mutation($id:String!){ projectDelete(id:$id){success} }";
    /// Fetch one document.
    pub const DOCUMENT: &str = "query($id:String!){ document(id:$id){ id title content url createdAt updatedAt archivedAt project{id} } }";
    /// List documents.
    ///
    /// `first` is an `Int` rather than an `Int!` because that is what Linear's `documents`
    /// connection declares, unlike its `issues` one.
    pub const DOCUMENTS: &str = "query($first:Int,$after:String,$filter:DocumentFilter){ documents(first:$first,after:$after,filter:$filter){ nodes{id title content url createdAt updatedAt project{id}} pageInfo{hasNextPage endCursor} } }";
    /// Create a document.
    pub const DOCUMENT_CREATE: &str = "mutation($input:DocumentCreateInput!){ documentCreate(input:$input){success document{id}} }";
    /// Update a document.
    pub const DOCUMENT_UPDATE: &str = "mutation($id:String!,$input:DocumentUpdateInput!){ documentUpdate(id:$id,input:$input){success document{id}} }";
    /// Delete a document, so a copy that could not finish can take back what it created.
    pub const DOCUMENT_DELETE: &str = "mutation($id:String!){ documentDelete(id:$id){success} }";
    /// One page of an issue's comments, walked backwards.
    ///
    /// `last`/`before` rather than `first`/`after`, and `pageInfo{hasPreviousPage
    /// startCursor}` rather than its forward pair, because Linear lists a connection newest
    /// first and the contract owes the oldest first — see the ruling on comments in this
    /// crate's module documentation. `archivedAt` is selected for the reason every by-id read
    /// here selects it: a trashed issue is not an issue this source holds.
    pub const ISSUE_COMMENTS: &str = "query($id:String!,$last:Int,$before:String){ issue(id:$id){ archivedAt comments(last:$last,before:$before){ nodes{id body url createdAt updatedAt user{displayName}} pageInfo{hasPreviousPage startCursor} } } }";
    /// Place one comment: which issue it is on, if any.
    ///
    /// `$id` is a nullable `String` because that is what `Query.comment` declares — it also
    /// takes a `hash` instead — and a variable has to be exactly its argument's type.
    pub const COMMENT: &str = "query($id:String){ comment(id:$id){ id archivedAt issue{id} } }";
    /// Add a comment to an issue.
    pub const COMMENT_CREATE: &str = "mutation($input:CommentCreateInput!){ commentCreate(input:$input){success comment{id body url createdAt updatedAt user{displayName}}} }";
    /// Replace a comment's body.
    pub const COMMENT_UPDATE: &str = "mutation($id:String!,$input:CommentUpdateInput!){ commentUpdate(id:$id,input:$input){success comment{id body url createdAt updatedAt user{displayName}}} }";
    /// Remove a comment.
    pub const COMMENT_DELETE: &str = "mutation($id:String!){ commentDelete(id:$id){success} }";
}

use graphql::{
    DOCUMENT, DOCUMENTS, ISSUE, ISSUE_RELATIONS, ISSUES, LABELS, PROJECT, PROJECT_RELATIONS,
    PROJECTS, VIEWER,
};

/// One `linear` source's configuration.
///
/// It names the credential's environment variable, never its value. Serializable so the
/// schema it is published under carries each member's default, which is what a configuration
/// that leaves the member out means.
#[derive(Debug, Clone, Deserialize, serde::Serialize, schemars::JsonSchema)]
#[serde(default, deny_unknown_fields)]
pub struct LinearConfig {
    /// Environment variable resolved by the host.
    #[schemars(with = "String")]
    api_key_env: EnvName,
    /// Linear team key/id used to narrow reads and required for item writes.
    team: Option<Team>,
    /// GraphQL endpoint override, primarily for fixture servers.
    #[schemars(with = "String")]
    endpoint: Endpoint,
    /// Per-instance mapping from a status category to the exact name of one workflow state
    /// of the configured team, or `null` to disable that category.
    ///
    /// The keys are status categories: `draft`, `backlog`, `todo`, `queued`, `in-progress`,
    /// `done`, `cancelled` and `unknown`. A mapped category is written as the named state by
    /// every write — never as the first state of that state's type — and an issue at a state
    /// the mapping names reads as that category, under that state's name; any other state
    /// reads by its type, and `--status` returns exactly the issues that read as the
    /// categories it names. A category this does not mention keeps the
    /// behaviour of a source without the key: `backlog`, `todo`, `in-progress`, `done` and
    /// `cancelled` are written as the team's first state of the matching type, and `draft`,
    /// `queued` and `unknown` are disabled. A name the team lacks is refused before any
    /// write, and two categories mapped to one name are refused when this configuration is
    /// read.
    #[schemars(schema_with = "status_mapping_schema")]
    status_mapping: std::collections::HashMap<StatusCategory, Option<StateName>>,
    /// The id of one Linear project of the configured team, scoping this source to it.
    ///
    /// When set, every task read is narrowed to that project's issues, project and document
    /// reads return only that project and its documents, a task or a document written with
    /// no project is placed in it, and one naming another project is refused. Absent, the
    /// source reads and writes team-wide.
    project: Option<ProjectScope>,
}

/// The schema of `status_mapping`: an object whose values are workflow state names or `null`,
/// and whose keys are the contract's own `StatusCategory`.
///
/// Built here rather than derived, because the derived schema of a map says nothing of its
/// keys; the key schema is the contract's own, referenced rather than restated.
fn status_mapping_schema(generator: &mut schemars::SchemaGenerator) -> Schema {
    let mut schema = <std::collections::HashMap<StatusCategory, Option<StateName>> as schemars::JsonSchema>::json_schema(generator);
    schema.insert(
        "propertyNames".to_owned(),
        generator.subschema_for::<StatusCategory>().to_value(),
    );
    schema
}

/// The name of one workflow state of a Linear team.
///
/// Validated on the way in rather than checked later, so a blank name — which no workflow
/// state can have — is a state this type cannot hold.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize, serde::Serialize, schemars::JsonSchema)]
#[serde(try_from = "String", into = "String")]
#[schemars(rename = "LinearWorkflowStateName", extend("minLength" = 1))]
struct StateName(String);
impl From<StateName> for String {
    fn from(value: StateName) -> Self {
        value.0
    }
}
impl TryFrom<String> for StateName {
    type Error = String;
    fn try_from(value: String) -> Result<Self, Self::Error> {
        if value.trim().is_empty() {
            Err("a status_mapping workflow state name cannot be blank".into())
        } else {
            Ok(Self(value))
        }
    }
}

/// The id of one Linear project, which a scoped source holds alone.
#[derive(Debug, Clone, Deserialize, serde::Serialize, schemars::JsonSchema)]
#[serde(try_from = "String", into = "String")]
#[schemars(rename = "LinearProjectId", extend("minLength" = 1))]
struct ProjectScope(String);
impl From<ProjectScope> for String {
    fn from(value: ProjectScope) -> Self {
        value.0
    }
}
impl TryFrom<String> for ProjectScope {
    type Error = String;
    fn try_from(value: String) -> Result<Self, Self::Error> {
        if value.trim().is_empty() {
            Err("a project id cannot be blank".into())
        } else {
            Ok(Self(value))
        }
    }
}

#[derive(Debug, Clone, Deserialize, serde::Serialize)]
#[serde(try_from = "String", into = "String")]
struct EnvName(String);
impl From<EnvName> for String {
    fn from(value: EnvName) -> Self {
        value.0
    }
}
impl TryFrom<String> for EnvName {
    type Error = String;
    fn try_from(value: String) -> Result<Self, Self::Error> {
        let mut bytes = value.bytes();
        if bytes
            .next()
            .is_some_and(|byte| byte == b'_' || byte.is_ascii_uppercase())
            && bytes.all(|byte| byte == b'_' || byte.is_ascii_uppercase() || byte.is_ascii_digit())
        {
            Ok(Self(value))
        } else {
            Err("must be an uppercase environment-variable name".into())
        }
    }
}
/// A Linear team's key or id.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize, serde::Serialize, schemars::JsonSchema)]
#[serde(try_from = "String", into = "String")]
#[schemars(rename = "LinearTeam", extend("minLength" = 1))]
struct Team(String);
impl From<Team> for String {
    fn from(value: Team) -> Self {
        value.0
    }
}
impl TryFrom<String> for Team {
    type Error = String;
    fn try_from(value: String) -> Result<Self, Self::Error> {
        if value.trim().is_empty() {
            Err("must not be empty".into())
        } else {
            Ok(Self(value))
        }
    }
}
#[derive(Debug, Clone, Deserialize, serde::Serialize)]
#[serde(try_from = "String", into = "String")]
struct Endpoint(String);
impl From<Endpoint> for String {
    fn from(value: Endpoint) -> Self {
        value.0
    }
}
impl TryFrom<String> for Endpoint {
    type Error = String;
    fn try_from(value: String) -> Result<Self, Self::Error> {
        let url = reqwest::Url::parse(&value).map_err(|e| e.to_string())?;
        if matches!(url.scheme(), "http" | "https") {
            Ok(Self(value))
        } else {
            Err("must use http or https".into())
        }
    }
}

impl Default for LinearConfig {
    fn default() -> Self {
        Self {
            api_key_env: EnvName("LINEAR_API_KEY".into()),
            team: None,
            endpoint: Endpoint(DEFAULT_ENDPOINT.into()),
            status_mapping: std::collections::HashMap::new(),
            project: None,
        }
    }
}

/// Where `category` sits in the contract's own order — the order a mapping is reported in.
///
/// An exhaustive match, so a status category the contract adds fails to compile here rather
/// than going unordered, and the order is the contract's rather than a list restated beside it.
const fn category_position(category: StatusCategory) -> usize {
    match category {
        StatusCategory::Draft => 0,
        StatusCategory::Backlog => 1,
        StatusCategory::Todo => 2,
        StatusCategory::Queued => 3,
        StatusCategory::InProgress => 4,
        StatusCategory::Done => 5,
        StatusCategory::Cancelled => 6,
        StatusCategory::Unknown => 7,
    }
}

/// A `WorkflowState.type` a status category is written as when the mapping names no state for
/// it — the closed set this source writes, which Linear's open vocabulary of read types
/// (`triage`, `duplicate`, …) is wider than.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum WorkflowType {
    Backlog,
    Unstarted,
    Started,
    Completed,
    Canceled,
}

impl WorkflowType {
    /// Every type this source writes, which is every type that reads as a category other than
    /// `unknown`: a state of any other type — `triage`, `duplicate`, or one Linear adds — reads
    /// as `unknown`. Held complete by the assertion beside [`Self::position`].
    const ALL: [Self; 5] = [
        Self::Backlog,
        Self::Unstarted,
        Self::Started,
        Self::Completed,
        Self::Canceled,
    ];

    /// Where this type sits in [`Self::ALL`] — an exhaustive match, so a type added here and
    /// left out of that list fails to compile.
    const fn position(self) -> usize {
        match self {
            Self::Backlog => 0,
            Self::Unstarted => 1,
            Self::Started => 2,
            Self::Completed => 3,
            Self::Canceled => 4,
        }
    }

    /// The type a category reads back as and is written as, or `None` for one no type stands
    /// for — `draft`, `queued` and `unknown`.
    const fn of(category: StatusCategory) -> Option<Self> {
        match category {
            StatusCategory::Backlog => Some(Self::Backlog),
            StatusCategory::Todo => Some(Self::Unstarted),
            StatusCategory::InProgress => Some(Self::Started),
            StatusCategory::Done => Some(Self::Completed),
            StatusCategory::Cancelled => Some(Self::Canceled),
            StatusCategory::Draft | StatusCategory::Queued | StatusCategory::Unknown => None,
        }
    }

    /// The type as Linear spells it.
    const fn as_str(self) -> &'static str {
        match self {
            Self::Backlog => "backlog",
            Self::Unstarted => "unstarted",
            Self::Started => "started",
            Self::Completed => "completed",
            Self::Canceled => "canceled",
        }
    }
}

const _: () = {
    let mut index = 0;
    while index < WorkflowType::ALL.len() {
        assert!(WorkflowType::ALL[index].position() == index);
        index += 1;
    }
};

/// Where one category is written, as this instance's `status_mapping` resolves it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum StateTarget<'a> {
    /// The team's workflow state of exactly this name.
    Named(&'a StateName),
    /// The team's first workflow state of this type: a category the mapping does not mention,
    /// as a source without the key writes it.
    FirstOfType(WorkflowType),
    /// No state, because the mapping sets the category to `null`.
    DisabledByMapping,
    /// No state, because the mapping leaves out a category no workflow state type stands for.
    NoWorkflowType,
}

/// This instance's `status_mapping`, read in both directions.
///
/// One entry per category the configuration names, at most once each and in the contract's
/// category order, and no two of them naming one state — a state that read back as two
/// categories would answer a status filter for either with the other's rows.
#[derive(Debug, Clone, Default)]
struct StatusMapping {
    entries: Vec<(StatusCategory, Option<StateName>)>,
}

impl StatusMapping {
    fn resolve(
        configured: std::collections::HashMap<StatusCategory, Option<StateName>>,
        instance: &SourceName,
    ) -> Result<Self, SourceError> {
        let mut configured: Vec<_> = configured.into_iter().collect();
        configured.sort_by_key(|(category, _)| category_position(*category));
        let mut entries: Vec<(StatusCategory, Option<StateName>)> = Vec::new();
        for (category, name) in configured {
            if let Some(name) = &name
                && let Some((other, _)) = entries.iter().find(|(_, held)| {
                    held.as_ref()
                        .is_some_and(|held| held.0.eq_ignore_ascii_case(&name.0))
                })
            {
                return Err(SourceError::Config {
                    message: format!(
                        "source {instance}: status_mapping sends both {} and {} to the workflow \
                         state {:?}; one state cannot read back as two categories, so map one of \
                         them to another state or to null",
                        category_word(*other),
                        category_word(category),
                        name.0
                    ),
                });
            }
            entries.push((category, name));
        }
        Ok(Self { entries })
    }

    fn is_empty(&self) -> bool {
        self.entries.is_empty()
    }

    fn target(&self, category: StatusCategory) -> StateTarget<'_> {
        match self.entries.iter().find(|(held, _)| *held == category) {
            Some((_, Some(name))) => StateTarget::Named(name),
            Some((_, None)) => StateTarget::DisabledByMapping,
            None => WorkflowType::of(category)
                .map_or(StateTarget::NoWorkflowType, StateTarget::FirstOfType),
        }
    }

    /// The category a workflow state of this name reads as, when the mapping names it.
    fn category_of(&self, state: &str) -> Option<StatusCategory> {
        self.entries.iter().find_map(|(category, name)| {
            name.as_ref()
                .is_some_and(|name| name.0.eq_ignore_ascii_case(state))
                .then_some(*category)
        })
    }

    /// Every workflow state name the mapping names, with its category.
    fn named(&self) -> impl Iterator<Item = (StatusCategory, &StateName)> {
        self.entries
            .iter()
            .filter_map(|(category, name)| name.as_ref().map(|name| (*category, name)))
    }
}

/// The Linear plugin factory.
#[derive(Debug, Clone, Copy, Default)]
pub struct Plugin;

impl SourcePlugin for Plugin {
    fn kind(&self) -> &'static str {
        KIND
    }
    fn config_schema(&self) -> Schema {
        schema_for!(LinearConfig)
    }
    fn build(
        &self,
        name: &SourceName,
        config: &Value,
        secrets: &dyn SecretResolver,
    ) -> Result<Box<dyn TaskSource>, SourceError> {
        let config: LinearConfig =
            serde_json::from_value(config.clone()).map_err(|e| SourceError::Config {
                message: format!("source {name}: {e}"),
            })?;
        Ok(Box::new(LinearSource::new(name, config, secrets)?))
    }
}

/// What `onetaskgraph sources fields` reports for a `linear` source: each workflow state its
/// `status_mapping` names, and whether the configured team has it.
///
/// A plan and nothing else. Workflow states are settings of the team that the people who own
/// it decide, so this source never creates, renames or retypes one; a state reported missing
/// is added in Linear's own team settings.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, schemars::JsonSchema)]
pub struct WorkflowStatesReport {
    /// The configured source name.
    pub source: SourceName,
    /// The configured team, as `team` names it.
    team: Team,
    /// Every workflow state `status_mapping` names, in category order. Empty when the mapping
    /// names none.
    pub states: Vec<MappedWorkflowState>,
}

impl WorkflowStatesReport {
    /// The configured team, as `team` names it.
    #[must_use]
    pub fn team(&self) -> &str {
        &self.team.0
    }
}

/// One workflow state `status_mapping` names, and whether the configured team has it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MappedWorkflowState {
    category: StatusCategory,
    state: StateName,
    found: Found,
}

/// Whether the configured team has a workflow state of the name the mapping gives.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Found {
    /// The team has it, of this `WorkflowState.type` — `backlog`, `unstarted`, `started`,
    /// `completed`, `canceled`, `triage`, or another Linear names, reported verbatim because
    /// Linear adds types (`duplicate` among them) this report must not refuse.
    // llmlint: ignore[invalid_states_unrepresentable] Linear's own open `String!` vocabulary, reported verbatim; an enum here would refuse a type Linear adds, which a report must not.
    Present(String),
    /// The team has no state of that name.
    Missing,
}

impl MappedWorkflowState {
    /// The category the mapping sends to the state.
    #[must_use]
    pub fn category(&self) -> StatusCategory {
        self.category
    }

    /// The state's name, as the mapping spells it.
    #[must_use]
    pub fn state(&self) -> &str {
        &self.state.0
    }

    /// Whether the team has it, and of which type.
    #[must_use]
    pub fn found(&self) -> &Found {
        &self.found
    }
}

/// [`MappedWorkflowState`] as it is written: `present`, and the state's `type` where it is.
///
/// The wire shape of the report, spelled once for its serialization and its schema, so the
/// public type can hold only the combinations [`Found`] allows.
#[derive(serde::Serialize, schemars::JsonSchema)]
#[schemars(rename = "MappedWorkflowState")]
struct MappedWorkflowStateWire<'a> {
    /// The category the mapping sends to the state.
    category: StatusCategory,
    /// The state's name, as the mapping spells it.
    state: &'a StateName,
    /// Whether the configured team has a workflow state of that name.
    present: bool,
    /// The state's `WorkflowState.type` on the team — `backlog`, `unstarted`, `started`,
    /// `completed`, `canceled`, `triage` or another Linear names — absent when it is missing.
    #[serde(rename = "type", skip_serializing_if = "Option::is_none")]
    state_type: Option<&'a str>,
}

impl<'a> From<&'a MappedWorkflowState> for MappedWorkflowStateWire<'a> {
    fn from(mapped: &'a MappedWorkflowState) -> Self {
        let state_type = match &mapped.found {
            Found::Present(kind) => Some(kind.as_str()),
            Found::Missing => None,
        };
        Self {
            category: mapped.category,
            state: &mapped.state,
            present: state_type.is_some(),
            state_type,
        }
    }
}

impl serde::Serialize for MappedWorkflowState {
    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        MappedWorkflowStateWire::from(self).serialize(serializer)
    }
}

impl schemars::JsonSchema for MappedWorkflowState {
    fn schema_name() -> std::borrow::Cow<'static, str> {
        MappedWorkflowStateWire::schema_name()
    }

    fn json_schema(generator: &mut schemars::SchemaGenerator) -> Schema {
        MappedWorkflowStateWire::json_schema(generator)
    }
}

/// Report each workflow state one `linear` source's `status_mapping` names, as present on its
/// team or missing from it, with its type. It writes nothing.
///
/// # Errors
///
/// [`SourceError::Config`] or [`SourceError::Auth`] for a source that cannot be built, a
/// refusal for one with no `team`, and whatever else Linear could not answer.
pub async fn workflow_states(
    name: &SourceName,
    config: LinearConfig,
    secrets: &dyn SecretResolver,
) -> Result<WorkflowStatesReport, SourceError> {
    LinearSource::new(name, config, secrets)?
        .workflow_states()
        .await
}

struct LinearSource {
    client: reqwest::Client,
    endpoint: Endpoint,
    key: SecretString,
    team: Option<Team>,
    /// This source's configured name, kept for one comparison: a far end recorded as
    /// `<this name>:<native>` is a Linear item Linear itself relates, so the reserved key
    /// is refused for it exactly as a bare id of the same kind is.
    name: SourceName,
    /// Where each status category is written and how each workflow state reads.
    statuses: StatusMapping,
    /// The one Linear project this source is scoped to, when it is.
    project: Option<ProjectScope>,
}

impl LinearSource {
    fn new(
        name: &SourceName,
        config: LinearConfig,
        secrets: &dyn SecretResolver,
    ) -> Result<Self, SourceError> {
        let statuses = StatusMapping::resolve(config.status_mapping, name)?;
        let key = secrets
            .get(&config.api_key_env.0)
            .filter(|v| !v.expose_secret().trim().is_empty())
            .ok_or_else(|| SourceError::Auth {
                message: format!("set environment variable {}", config.api_key_env.0),
            })?;
        Ok(Self {
            client: reqwest::Client::new(),
            endpoint: config.endpoint,
            key,
            team: config.team,
            name: name.clone(),
            statuses,
            project: config.project,
        })
    }
}
#[derive(Clone, Copy)]
enum WriteKind {
    Task,
    Project,
}
enum Lookup<'a> {
    Team(&'a str),
    IssueState { name: &'a str, team: &'a NativeId },
    ProjectStatus(&'a str),
    IssueLabel(&'a str),
    ProjectLabel(&'a str),
}
impl Lookup<'_> {
    fn query(&self) -> &'static str {
        match self {
            Self::Team(_) => graphql::TEAM,
            Self::IssueState { .. } => graphql::ISSUE_STATE,
            Self::ProjectStatus(_) => graphql::PROJECT_STATUS,
            Self::IssueLabel(_) => graphql::ISSUE_LABEL,
            Self::ProjectLabel(_) => graphql::PROJECT_LABEL,
        }
    }
    fn connection(&self) -> &'static str {
        match self {
            Self::Team(_) => "teams",
            Self::IssueState { .. } => "workflowStates",
            Self::ProjectStatus(_) => "projectStatuses",
            Self::IssueLabel(_) => "issueLabels",
            Self::ProjectLabel(_) => "projectLabels",
        }
    }
    fn diagnostic(&self) -> String {
        match self {
            Self::Team(_) => "configured team".into(),
            Self::IssueState { name, .. } => format!("workflow state {name:?}"),
            Self::ProjectStatus(name) => format!("project status {name:?}"),
            Self::IssueLabel(name) | Self::ProjectLabel(name) => format!("label {name:?}"),
        }
    }
    fn variables(&self) -> Value {
        match self {
            Self::Team(key) => json!({"key":key}),
            Self::IssueState { name, team } => json!({"name":name,"team":team.0}),
            Self::IssueLabel(name) | Self::ProjectLabel(name) => json!({"name":name}),
            // `PROJECT_STATUS` names nothing, for the reason recorded on that document.
            Self::ProjectStatus(_) => json!({}),
        }
    }
    /// The display name `one_id` matches locally, for the one lookup whose connection
    /// Linear will not narrow server-side.
    fn local_name(&self) -> Option<&str> {
        match self {
            Self::ProjectStatus(name) => Some(name),
            _ => None,
        }
    }
}
#[derive(Clone, Copy)]
enum MutationRoot {
    IssueCreate,
    IssueUpdate,
    ProjectCreate,
    ProjectUpdate,
    IssueRelationCreate,
    ProjectRelationCreate,
    IssueRelationDelete,
    ProjectRelationDelete,
    IssueDelete,
    ProjectDelete,
    DocumentCreate,
    DocumentUpdate,
    DocumentDelete,
    CommentCreate,
    CommentUpdate,
    CommentDelete,
}
impl MutationRoot {
    fn as_str(self) -> &'static str {
        match self {
            Self::IssueCreate => "issueCreate",
            Self::IssueUpdate => "issueUpdate",
            Self::ProjectCreate => "projectCreate",
            Self::ProjectUpdate => "projectUpdate",
            Self::IssueRelationCreate => "issueRelationCreate",
            Self::ProjectRelationCreate => "projectRelationCreate",
            Self::IssueRelationDelete => "issueRelationDelete",
            Self::ProjectRelationDelete => "projectRelationDelete",
            Self::IssueDelete => "issueDelete",
            Self::ProjectDelete => "projectDelete",
            Self::DocumentCreate => "documentCreate",
            Self::DocumentUpdate => "documentUpdate",
            Self::DocumentDelete => "documentDelete",
            Self::CommentCreate => "commentCreate",
            Self::CommentUpdate => "commentUpdate",
            Self::CommentDelete => "commentDelete",
        }
    }
}

#[derive(Deserialize)]
struct Envelope {
    // llmlint: ignore[invalid_states_unrepresentable] One transport envelope carries eight distinct GraphQL data shapes; each operation immediately validates its own complete mapper into typed plugin-api values, so malformed external data cannot cross the plugin boundary and a union here would duplicate every query response solely inside transport code.
    data: Option<Value>,
    #[serde(default)]
    errors: Vec<GqlError>,
}
#[derive(Deserialize)]
struct GqlError {
    message: String,
    // Held raw rather than typed, for two reasons. Linear puts the whole of *why* it
    // refused in here — `message` is a category name like `Argument Validation Error`,
    // which named neither the field nor the value when the live project-relation write
    // was refused by it — so a refusal carries this verbatim and a reader diagnoses from
    // it. And a typed shape with a required `code` fails the whole envelope's
    // deserialization when Linear sends extensions without one, turning a refusal this
    // source could explain into an unexplained malformed response.
    extensions: Option<Value>,
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct GqlExtensions {
    code: GqlErrorCode,
    retry_after: Option<u64>,
}
impl GqlError {
    /// The rate-limit shape of [`Self::extensions`], when it has one.
    fn coded(&self) -> Option<GqlExtensions> {
        self.extensions
            .as_ref()
            .and_then(|value| serde_json::from_value(value.clone()).ok())
    }
    /// Everything Linear said about this refusal, on one line and cut to [`SAID_LIMIT`].
    ///
    /// Linear's own sentence comes first, then the raw envelope, because only the first
    /// of those two is short enough to survive [`SAID_LIMIT`] on its merits. `message` is
    /// a category name — `Argument Validation Error` — and the sentence naming the field
    /// and the values it would have taken is `extensions.userPresentableMessage`, one of
    /// several keys in an envelope whose `validationErrors` echoes the whole rejected
    /// input back. Observed against the real API on 2026-09-04, a `projectRelationCreate`
    /// refusal rendered past the cut, and the echo is what got cut.
    ///
    /// That the sentence itself did not was luck: this build of `serde_json` renders an
    /// object's keys sorted, and `userPresentableMessage` happens to sort ahead of
    /// `validationErrors`. Nobody chose that — Linear sends the echo first — and any key
    /// Linear adds sorting between the two would move the sentence behind an echo longer
    /// than the whole limit, as would turning `preserve_order` on. Leading with it makes
    /// what a reader diagnoses from independent of both.
    fn said(&self) -> String {
        let Some(extensions) = &self.extensions else {
            return elided(&self.message);
        };
        match extensions
            .get("userPresentableMessage")
            .and_then(Value::as_str)
            .filter(|sentence| !sentence.is_empty())
        {
            Some(sentence) => elided(&format!("{}: {sentence} {extensions}", self.message)),
            None => elided(&format!("{}: {extensions}", self.message)),
        }
    }
}
#[derive(Deserialize)]
enum GqlErrorCode {
    #[serde(rename = "RATELIMITED", alias = "RATE_LIMITED")]
    RateLimited,
    #[serde(other)]
    Other,
}

/// How much of a failed response's body a refusal carries.
///
/// Enough for Linear's own error envelope, which is one or two sentences naming the field
/// or argument it would not accept, and short enough that a proxy's HTML error page does
/// not become the whole message.
const SAID_LIMIT: usize = 400;

/// `said` made safe to put in a message: one line of printable text, cut to [`SAID_LIMIT`].
///
/// A failed response's body is whatever answered — Linear's error envelope, or an HTML
/// page from a proxy in front of it — and this message is written to a terminal. So every
/// control character goes, escape sequences with them, and each run of whitespace becomes
/// one space: a body cannot move the cursor, repaint the line or hide the rest of the
/// diagnostic behind itself. Cut by characters rather than bytes, because slicing UTF-8
/// mid-codepoint would panic inside the path that exists to explain a failure.
fn elided(said: &str) -> String {
    let mut printable = String::new();
    let mut spaced = true;
    for character in said.chars() {
        if character.is_control() || character.is_whitespace() {
            if !spaced {
                printable.push(' ');
                spaced = true;
            }
            continue;
        }
        printable.push(character);
        spaced = false;
    }
    let printable = printable.trim_end();
    if printable.chars().count() <= SAID_LIMIT {
        return printable.to_owned();
    }
    let kept: String = printable.chars().take(SAID_LIMIT).collect();
    format!("{kept}…")
}

impl LinearSource {
    // llmlint: ignore[invalid_states_unrepresentable] This private generic transport accepts only variables constructed immediately at typed TaskSource call sites, never untrusted input; per-operation response mappers validate every external field before returning public values.
    async fn send(&self, query: &str, variables: Value) -> Result<Value, SourceError> {
        let response = self
            .client
            .post(&self.endpoint.0)
            .header("Authorization", self.key.expose_secret())
            .json(&json!({"query": query, "variables": variables}))
            .send()
            .await
            .map_err(|e| SourceError::Unavailable {
                message: e.to_string(),
            })?;
        let status = response.status();
        let retry = response
            .headers()
            .get("retry-after")
            .and_then(|v| v.to_str().ok())
            .and_then(|v| v.parse().ok());
        if status.as_u16() == 429 {
            return Err(SourceError::RateLimited {
                retry_after_seconds: retry,
                // Linear has one rate limiter and the status is the whole of what it said,
                // so there is nothing to add beyond the kind — which is what an absent
                // message means.
                message: None,
            });
        }
        if status.as_u16() == 401 || status.as_u16() == 403 {
            return Err(SourceError::Auth {
                message: "Linear rejected the configured credential".into(),
            });
        }
        if !status.is_success() {
            // Linear puts its GraphQL error envelope in the *body* of a 400, so the status
            // alone names the whole call and nothing about what Linear objected to. The
            // body is Linear's answer to this request and holds no credential; it is cut
            // because a proxy in front of Linear can answer with a page.
            let said = elided(&response.text().await.unwrap_or_default());
            return Err(SourceError::Unavailable {
                message: if said.is_empty() {
                    format!("Linear returned HTTP {status}")
                } else {
                    format!("Linear returned HTTP {status}: {said}")
                },
            });
        }
        let body: Envelope = response.json().await.map_err(|e| SourceError::Malformed {
            message: e.to_string(),
        })?;
        if let Some(error) = body.errors.first() {
            if let Some(extensions) = error
                .coded()
                .filter(|extensions| matches!(extensions.code, GqlErrorCode::RateLimited))
            {
                return Err(SourceError::RateLimited {
                    retry_after_seconds: extensions.retry_after.or(retry),
                    message: None,
                });
            }
            return Err(SourceError::Refused {
                message: error.said(),
            });
        }
        body.data.ok_or_else(|| SourceError::Malformed {
            message: "GraphQL response has no data".into(),
        })
    }

    // llmlint: ignore-block[contracts_have_one_source_or_a_drift_gate] These operators follow the accepted 2026-08-24 Linear contract, but Linear exposes their authoritative definitions only through an authenticated unversioned explorer; the real-HTTP tests assert every serialized operator and the shared CLI journeys assert resulting rows without making credentials required.
    /// The label predicates, which really are spelled the same at both levels.
    ///
    /// `IssueFilter.labels` is an `IssueLabelCollectionFilter` and `ProjectFilter.labels`
    /// is a `ProjectLabelCollectionFilter` — two types — but `some`, `every` and a `name`
    /// of `StringComparator` are members of both, so one spelling satisfies each. That is
    /// the whole of what the two filters have in common, and everything else about them is
    /// built separately for the reason recorded on the two builders below.
    ///
    /// "At least one of these" is a disjunction of `eqIgnoreCase` rather than one
    /// case-insensitive list operator, because Linear has no such operator. This source
    /// sent `labels:{some:{name:{inIgnoreCase:[…]}}}` until Linear refused it outright,
    /// HTTP 400, on the first read of the live lane that ever reached a label filter:
    ///
    /// ```text
    /// Variable "$filter" got invalid value { inIgnoreCase: […] } at
    /// "filter.and[1].labels.some.name"; Field "inIgnoreCase" is not defined by
    /// type "StringComparator". Did you mean "eqIgnoreCase" or "neqIgnoreCase"?
    /// ```
    ///
    /// That refusal is also the evidence for the replacement: Linear named the two members
    /// of `StringComparator` closest to what it was sent, and `eqIgnoreCase` is one of
    /// them — the same operator `all_of` below has always sent and the live lane has always
    /// exercised. `in` exists there too and would need no `or`, but it is case-sensitive,
    /// so `any_of` would stop agreeing with `all_of` and `none_of` and with what the table
    /// at the top of this file says this source does.
    fn label_parts(labels: &onetaskgraph_plugin_api::LabelFilter) -> Vec<Value> {
        let mut parts = Vec::new();
        if !labels.any_of.is_empty() {
            parts.push(json!({"or": labels
                .any_of
                .iter()
                .map(|name| json!({"labels": {"some": {"name": {"eqIgnoreCase": name}}}}))
                .collect::<Vec<_>>()}));
        }
        for name in &labels.all_of {
            parts.push(json!({"labels": {"some": {"name": {"eqIgnoreCase": name}}}}));
        }
        for name in &labels.none_of {
            parts.push(json!({"labels": {"every": {"name": {"neqIgnoreCase": name}}}}));
        }
        parts
    }
    fn narrowed(mut parts: Vec<Value>) -> Value {
        if parts.len() == 1 {
            parts.pop().unwrap()
        } else {
            json!({"and": parts})
        }
    }
    /// The filter this source sends to `issues(filter:)`.
    ///
    /// **`IssueFilter` and `ProjectFilter` are different input types, and one builder for
    /// both is what put two wrong fields on the wire.** They read as though they were the
    /// same filter over different rows — the label member really is spelled alike, and the
    /// `and`/`or` are identical — and a single builder producing one object for both
    /// connections had shipped `team` and the issue's `state` shape into `projects(filter:)`
    /// since long before this branch. Linear refused the first outright:
    ///
    /// ```text
    /// Variable "$filter" got invalid value { team: { key: [Object] } };
    /// Field "team" is not defined by type "ProjectFilter". Did you mean "lead"?
    /// ```
    ///
    /// So there are two builders, and each names its own type's members. Adding a predicate
    /// means deciding twice, on purpose, rather than once by accident.
    fn issue_filter(&self, query: &TaskQuery) -> Result<Value, SourceError> {
        let mut parts = self.issue_scope();
        parts.extend(Self::label_parts(&query.labels));
        if !query.statuses.is_empty() {
            parts.push(self.status_narrowing(&query.statuses));
        }
        match &query.project {
            ProjectFilter::Orphans => parts.push(json!({"project": {"null": true}})),
            ProjectFilter::Is(id) => parts.push(json!({"project": {"id": {"eq": id.0}}})),
            ProjectFilter::Any => {}
        }
        // The narrowings below are each confirmed in process by `confirms` before a row is
        // returned: Linear's comparators are a candidate set, and the contract's predicate is
        // what decides. None of them can drop a row the predicate keeps — see the module
        // documentation's ruling on the follow-up searches for what each one rests on.
        if !query.priorities.is_empty() {
            parts.push(json!({"priority": {"in": query
                .priorities
                .iter()
                .map(|priority| linear_priority(*priority))
                .collect::<Vec<_>>()}}));
        }
        if let Some(since) = query.commented_since {
            let since = since.to_rfc3339_opts(chrono::SecondsFormat::AutoSi, true);
            parts.push(json!({"comments": {"some": {"or": [
                {"createdAt": {"gte": since}},
                {"updatedAt": {"gte": since}},
            ]}}}));
        }
        for wanted in &query.metadata {
            parts.extend(slot_phrase(wanted.value()));
        }
        if let Some(origin) = &query.origin {
            parts.extend(slot_phrase(origin));
        }
        if let Some(text) = &query.text {
            let title = json!({"title": {"containsIgnoreCase": text.terms}});
            let content = json!({"description": {"containsIgnoreCase": text.terms}});
            parts.push(match text.fields {
                TextFields::Title => title,
                TextFields::Content => content,
                TextFields::TitleOrContent => json!({"or": [title, content]}),
            });
        }
        Ok(Self::narrowed(parts))
    }

    /// What every issue read of this source is narrowed to before any predicate: the
    /// configured team, and the project this source is scoped to, when it is.
    fn issue_scope(&self) -> Vec<Value> {
        let mut parts = Vec::new();
        if let Some(team) = &self.team {
            parts.push(json!({"team": {"key": {"eqIgnoreCase": team.0}}}));
        }
        if let Some(project) = &self.project {
            parts.push(json!({"project": {"id": {"eq": project.0}}}));
        }
        parts
    }

    /// The issue-side status narrowing for `statuses`, through this instance's mapping.
    ///
    /// **Without a mapping it is exactly what it always was** — one `state:{type:{in:[…]}}`
    /// over the categories' workflow-state types. With one, each category asks for exactly
    /// what reads as it: the issues at its named state, by name, when it is mapped; and the
    /// issues of its type *at no state the mapping names*, because those read as the category
    /// they are mapped to — a `todo` filter returning an issue at `Queued` would be a row that
    /// reads back as another category. A category set to `null` is never written, and still
    /// asks for the issues of its type, which read as it.
    fn status_narrowing(&self, statuses: &[StatusCategory]) -> Value {
        if self.statuses.is_empty() {
            return json!({"state": {"type": {"in": statuses
                .iter()
                .flat_map(workflow_state_types)
                .collect::<Vec<_>>()}}});
        }
        let mut alternatives = Vec::new();
        for category in statuses {
            let target = self.statuses.target(*category);
            if let StateTarget::Named(name) = target {
                alternatives.push(json!({"state": {"name": {"eqIgnoreCase": name.0}}}));
            }
            // Every issue that reads as this category by its type — at a state of the type and
            // of no name the mapping claims — whether the category is mapped, left out or set
            // to `null`: with `backlog` mapped to `Proposed` an issue at `Backlog` still reads
            // as `backlog`, and with `cancelled` set to `null` an issue at `Canceled` still
            // reads as `cancelled`, which a write can no longer move it to but a read reports.
            let types = workflow_state_types(category);
            // `unknown` has no type of its own: a state reads as it when its type is none of
            // the five a category stands for — `Triage`'s `triage`, a `duplicate` state.
            let by_type = if *category == StatusCategory::Unknown {
                Some(
                    json!({"state": {"type": {"nin": WorkflowType::ALL.map(WorkflowType::as_str)}}}),
                )
            } else {
                (!types.is_empty()).then(|| json!({"state": {"type": {"in": types}}}))
            };
            if let Some(by_type) = by_type {
                let mut parts = vec![by_type];
                parts.extend(
                    self.statuses
                        .named()
                        .map(|(_, name)| json!({"state": {"name": {"neqIgnoreCase": name.0}}})),
                );
                alternatives.push(Self::narrowed(parts));
            }
        }
        match alternatives.len() {
            // Every category asked for is disabled: a type no state has, which matches nothing
            // and is refused by nothing — what a disabled category always narrowed to.
            0 => json!({"state": {"type": {"in": Vec::<&str>::new()}}}),
            1 => alternatives.pop().expect("one alternative"),
            _ => json!({ "or": alternatives }),
        }
    }

    /// Whether one task this source read satisfies every predicate of `query` that a
    /// narrowing above only approximates — the metadata matches, the origin, the text and the
    /// priorities — by the contract's own statement of each.
    fn confirms(query: &TaskQuery, task: &Task) -> bool {
        query.metadata_matches(&task.metadata)
            && query.origin_matches(&task.metadata)
            && (query.priorities.is_empty() || query.priorities.contains(&task.priority))
            && query
                .text
                .as_ref()
                .is_none_or(|text| text_holds(&task.title, task.content.as_deref(), text))
    }

    /// The filter this source sends to `projects(filter:)`.
    ///
    /// Two members differ from [`Self::issue_filter`] and both are Linear's doing; see that
    /// builder for why they are written out twice rather than shared.
    ///
    /// **A project has no `team`.** It has the teams it is accessible from, and
    /// `ProjectFilter.accessibleTeams` is a `TeamCollectionFilter`, so the same team key
    /// reaches it under `some:`. `leadTeam` is the other team-shaped member and is a
    /// different set — one designated team rather than every team the project is in — so
    /// narrowing by it would drop projects the configured team really does hold.
    ///
    /// **A project's status is not an issue's state, and they do not even share a
    /// vocabulary.** An issue's is `WorkflowState`, reached through `IssueFilter.state`,
    /// and its `type` is `backlog`, `unstarted`, `started`, `completed`, `canceled` or
    /// `triage`. A project's is `ProjectStatus`, reached through `ProjectFilter.status` —
    /// `ProjectFilter.state` exists and is *not* it: that member is a bare
    /// `StringComparator` over a different thing — and its `type` is the `ProjectStatusType`
    /// enum, `backlog`, `planned`, `started`, `paused`, `completed`, `canceled`. So the
    /// nearest thing to an issue's `unstarted` is a project's `planned`, and `paused` has no
    /// issue counterpart at all. [`project_status_types`] is that vocabulary and
    /// [`workflow_state_types`] is the other; sending either one's words to the other's
    /// connection matches nothing while refusing nothing, which is the worst way to be
    /// wrong.
    fn project_filter(
        &self,
        labels: &onetaskgraph_plugin_api::LabelFilter,
        statuses: &[StatusCategory],
    ) -> Value {
        let mut parts = self.project_scope();
        parts.extend(Self::label_parts(labels));
        if !statuses.is_empty() {
            parts.push(json!({"status": {"type": {"in": statuses.iter().flat_map(project_status_types).collect::<Vec<_>>()}}}));
        }
        Self::narrowed(parts)
    }

    /// What every project read of this source is narrowed to: the projects the configured
    /// team can reach, and the one project this source is scoped to, when it is.
    fn project_scope(&self) -> Vec<Value> {
        let mut parts = Vec::new();
        if let Some(team) = &self.team {
            parts.push(json!({"accessibleTeams": {"some": {"key": {"eqIgnoreCase": team.0}}}}));
        }
        if let Some(project) = &self.project {
            parts.push(json!({"id": {"eq": project.0}}));
        }
        parts
    }

    /// Whether an item filed under `project` is one this source holds: always, unless it is
    /// scoped to one project and this is not filed under it.
    fn in_scope(&self, project: Option<&NativeId>) -> bool {
        self.project
            .as_ref()
            .is_none_or(|scope| project.is_some_and(|project| project.0 == scope.0))
    }

    /// The project a task or a document written with `project` is filed under: that one, or
    /// the scope when it names none — and a refusal naming both when it names another.
    fn filed_in(
        &self,
        project: Option<&NativeId>,
        what: &str,
    ) -> Result<Option<String>, SourceError> {
        match (&self.project, project) {
            (None, project) => Ok(project.map(|id| id.0.clone())),
            (Some(scope), None) => Ok(Some(scope.0.clone())),
            (Some(scope), Some(project)) if project.0 == scope.0 => Ok(Some(scope.0.clone())),
            (Some(scope), Some(project)) => Err(SourceError::Refused {
                message: format!(
                    "source {} is scoped to the Linear project {} and cannot hold a {what} in                      the project {}; next: write it with no project, or with {}, or to a                      source scoped to {}",
                    self.name, scope.0, project.0, scope.0, project.0
                ),
            }),
        }
    }
    // llmlint: ignore-end[contracts_have_one_source_or_a_drift_gate]

    async fn one_id(&self, lookup: Lookup<'_>) -> Result<NativeId, SourceError> {
        let data = self.send(lookup.query(), lookup.variables()).await?;
        let connection = lookup.connection();
        let nodes = data
            .get(connection)
            .and_then(|v| v.get("nodes"))
            .and_then(Value::as_array)
            .ok_or_else(|| SourceError::Malformed {
                message: format!("missing {connection}.nodes"),
            })?;
        // A node this comparison cannot read is malformed rather than a nonmatch: dropping
        // it would turn Linear having answered nonsense into this source reporting no such
        // status, which is a different thing and reads as the caller's mistake.
        let matched = match lookup.local_name() {
            Some(name) => {
                let mut matched = Vec::new();
                for node in nodes {
                    if str_at(node, "name")?.eq_ignore_ascii_case(name) {
                        matched.push(node);
                    }
                }
                matched
            }
            None => nodes.iter().collect::<Vec<_>>(),
        };
        match matched.as_slice() {
            [] => Err(SourceError::Refused {
                message: format!(
                    "source {} cannot resolve {}: found 0 matches",
                    self.name,
                    lookup.diagnostic()
                ),
            }),
            [node] => Ok(NativeId(backend_id(node, "id")?.to_owned())),
            nodes => {
                let ids = nodes
                    .iter()
                    .map(|node| backend_id(node, "id"))
                    .collect::<Result<Vec<_>, _>>()?;
                Err(SourceError::Refused {
                    message: format!(
                        "source {} cannot resolve {}: found {} matches with ids {ids:?}",
                        self.name,
                        lookup.diagnostic(),
                        nodes.len()
                    ),
                })
            }
        }
    }
    async fn team_id(&self) -> Result<NativeId, SourceError> {
        let team = self.team.as_ref().ok_or_else(|| SourceError::Refused {
            message: format!(
                "source {} needs config.team before it can create Linear items",
                self.name
            ),
        })?;
        self.one_id(Lookup::Team(&team.0)).await
    }
    async fn label_ids(
        &self,
        labels: &[Label],
        kind: WriteKind,
    ) -> Result<Vec<NativeId>, SourceError> {
        let mut ids = Vec::with_capacity(labels.len());
        for label in labels {
            ids.push(
                self.one_id(if matches!(kind, WriteKind::Project) {
                    Lookup::ProjectLabel(&label.name)
                } else {
                    Lookup::IssueLabel(&label.name)
                })
                .await?,
            );
        }
        Ok(ids)
    }
    fn write_description(
        &self,
        content: Option<&str>,
        metadata: &std::collections::BTreeMap<String, Value>,
        repositories: &[Repository],
        edges: &[DependencyEdge],
        kind: WriteKind,
    ) -> Result<Option<String>, SourceError> {
        Self::long_form(
            content,
            metadata,
            repositories,
            self.recorded_ends(edges, kind),
        )
    }

    /// The far ends of `edges` no relation of this workspace can name — another level, or
    /// another source — as the reserved key records them.
    fn recorded_ends(&self, edges: &[DependencyEdge], kind: WriteKind) -> Vec<Value> {
        edges
            .iter()
            .filter(|edge| {
                edge.to.kind
                    != match kind {
                        WriteKind::Task => ItemKind::Task,
                        WriteKind::Project => ItemKind::Project,
                    }
                    || edge
                        .to
                        .id()
                        .split_once(':')
                        .is_some_and(|(source, _)| source != self.name.as_str())
            })
            .map(|edge| json!({"id":edge.to.id(),"kind":edge.to.kind}))
            .collect()
    }

    /// Every forward edge `id` holds, relations and recorded far ends alike, walked to
    /// exhaustion.
    async fn forward_edges(&self, id: &NativeId) -> Result<Vec<DependencyEdge>, SourceError> {
        let mut edges = Vec::new();
        let mut cursor = None;
        loop {
            let page = self
                .dependencies(
                    ISSUE_RELATIONS,
                    DependencyRoot::Issue,
                    id,
                    Direction::DependsOn,
                    &PageRequest {
                        cursor,
                        limit: MAX_PAGE_SIZE,
                    },
                )
                .await?;
            edges.extend(page.items);
            match page.next {
                Some(next) => cursor = Some(next),
                None => return Ok(edges),
            }
        }
    }

    /// Apply one targeted update to one issue; see [`TaskSource::update_task`].
    ///
    /// One read of the issue, then one `issueUpdate` carrying only the members that differ
    /// from it — `title`, `description` (content and metadata slot together), `stateId`,
    /// `priority` — and, when the named edges differ from the ones the issue holds, its
    /// relations replaced. Nothing is sent for a field already holding the requested value,
    /// and nothing at all when nothing differs. A status is written by its category, exactly
    /// as `set_task_status` writes one: an issue already in that category keeps the state it
    /// is in, and one that is not moves to the state `status_mapping` names for it, or to the
    /// team's first state of that type where the mapping names none. `delivers` lands in the
    /// metadata slot under its reserved key, as a copy writes it.
    ///
    /// The task answered is read back after the write, so its status is Linear's.
    async fn targeted_update(
        &self,
        id: &NativeId,
        update: &TaskUpdate,
    ) -> Result<Option<TaskUpdateOutcome>, SourceError> {
        update.consistent()?;
        if let Some(delivers) = &update.delivers {
            TaskRef::listed(
                TaskRef::DELIVERS_KEY,
                id,
                Some(&self.name),
                delivers.clone(),
            )
            .map_err(|message| SourceError::Refused { message })?;
        }
        if let Some(status) = &update.status {
            self.writable(status.category)?;
        }
        let Some((before, description)) = self.issue_held(id).await? else {
            return Ok(None);
        };
        let (visible, held) = metadata_description(description)?;
        let mut slot = held.clone();
        for (key, value) in &update.metadata_set {
            slot.insert(key.as_str().to_owned(), value.clone());
        }
        for key in &update.metadata_remove {
            slot.remove(key.as_str());
        }
        if let Some(delivers) = &update.delivers {
            set_task_list(&mut slot, TaskRef::DELIVERS_KEY, delivers);
        }
        let mut relations = None;
        if let Some(wanted) = &update.depends_on {
            let current = self.forward_edges(&before.id).await?;
            let ends = |edges: &[DependencyEdge]| {
                let mut ends: Vec<(String, String)> = edges
                    .iter()
                    .map(|edge| {
                        (
                            edge.to.id().to_owned(),
                            format!("{:?}{:?}", edge.to.kind, edge.kind),
                        )
                    })
                    .collect();
                ends.sort();
                ends
            };
            if ends(&current) != ends(wanted) {
                let prepared = self.prepare_edges(wanted, WriteKind::Task).await?;
                let recorded = self.recorded_ends(&prepared, WriteKind::Task);
                if recorded.is_empty() {
                    slot.remove(DependencyEdge::RECORDED_KEY);
                } else {
                    slot.insert(DependencyEdge::RECORDED_KEY.into(), Value::Array(recorded));
                }
                relations = Some(prepared);
            }
        }
        let mut input = serde_json::Map::new();
        if let Some(title) = update
            .title
            .as_ref()
            .filter(|title| **title != before.title)
        {
            input.insert("title".into(), json!(title));
        }
        let content = update.content.as_deref().or(visible.as_deref());
        if content != visible.as_deref() || slot != held {
            let written = Self::described(content, &slot)?;
            // Checked before anything is sent: content ending in what this source reads as
            // its own slot would read back as metadata rather than as the content it was.
            let (reads, read) = metadata_description(written.clone())?;
            if reads.as_deref().unwrap_or_default() != content.unwrap_or_default() || read != slot {
                return Err(SourceError::Refused {
                    message: format!(
                        "this content would read back from source {} as something other than \
                         itself, or ends in what it reads as its own metadata slot; next: change \
                         how the content ends",
                        self.name
                    ),
                });
            }
            input.insert("description".into(), json!(written));
        }
        if let Some(status) = update
            .status
            .as_ref()
            .filter(|status| status.category != before.status.category)
        {
            let (state, _) = self.resolve_state(status.category, &before.id).await?;
            input.insert("stateId".into(), json!(state.0));
        }
        if let Some(priority) = update
            .priority
            .filter(|priority| *priority != before.priority)
        {
            input.insert("priority".into(), json!(linear_priority(priority)));
        }
        let sent = !input.is_empty();
        if sent {
            let data = self
                .send(
                    graphql::ISSUE_UPDATE,
                    json!({"id":before.id.0,"input":Value::Object(input)}),
                )
                .await?;
            let issue = mutation_payload(&data, MutationRoot::IssueUpdate)?
                .get("issue")
                .filter(|issue| !issue.is_null())
                .ok_or_else(|| SourceError::Malformed {
                    message: "missing issueUpdate.issue".into(),
                })?;
            written_is(issue, &before.id)?;
        }
        if let Some(prepared) = &relations {
            self.write_relations(&before.id, prepared, WriteKind::Task)
                .await?;
        }
        let task = if sent || relations.is_some() {
            self.get_task(&before.id)
                .await?
                .ok_or_else(|| SourceError::Malformed {
                    message: format!("task {id} was updated and then could not be read back"),
                })?
        } else {
            before.clone()
        };
        let mut written = update.changed(&before, &task);
        if relations.is_some() {
            written.insert(UpdatedField::DependsOn);
        }
        Ok(Some(TaskUpdateOutcome {
            task,
            written,
            delivers_before: before.delivers,
        }))
    }

    /// The one long-form field a Linear item has, with this source's own slot at the end.
    ///
    /// Shared by every kind this source writes rather than reimplemented per kind: a
    /// document keeps caller metadata in exactly the slot an issue and a project do, which
    /// is what lets the same read side take it back out.
    fn long_form(
        content: Option<&str>,
        metadata: &std::collections::BTreeMap<String, Value>,
        repositories: &[Repository],
        recorded: Vec<Value>,
    ) -> Result<Option<String>, SourceError> {
        let mut metadata = metadata.clone();
        if repositories.is_empty() {
            metadata.remove(Repository::METADATA_KEY);
        } else {
            metadata.insert(Repository::METADATA_KEY.into(), json!(repositories));
        }
        if recorded.is_empty() {
            metadata.remove(DependencyEdge::RECORDED_KEY);
        } else {
            metadata.insert(DependencyEdge::RECORDED_KEY.into(), Value::Array(recorded));
        }
        Self::described(content, &metadata)
    }

    /// The long-form field holding `content` and a slot of exactly `metadata`, in the one
    /// encoding [`long_form`](Self::long_form) writes: the content alone when there is no
    /// metadata, and otherwise the slot after one blank line.
    fn described(
        content: Option<&str>,
        metadata: &std::collections::BTreeMap<String, Value>,
    ) -> Result<Option<String>, SourceError> {
        let visible = content.unwrap_or_default();
        if metadata.is_empty() {
            return Ok((!visible.is_empty()).then(|| visible.to_owned()));
        }
        let slot = slot_text(metadata)?;
        Ok(Some(if visible.is_empty() {
            slot
        } else {
            format!("{visible}\n\n{slot}")
        }))
    }
    /// What this source says when asked for a project edge carrying no ordering.
    ///
    /// Linear's project relations have exactly one type and it is an ordering. Asked on
    /// 2026-09-04 to create one typed `related` — and separately `blocks` and `dependsOn`
    /// — the real API refused each with `Argument Validation Error` and
    /// `constraints: {"isEnum": "type must be one of the following values: dependency"}`.
    /// That is Linear's own enumeration of the field, from the validator behind GraphQL
    /// where introspection cannot reach it, and it has one member. An issue relation is a
    /// different relation with a different set, which does include `related`, so this
    /// reaches projects alone.
    fn unordered_project_relation(&self, near: &NativeId, far: &str) -> SourceError {
        SourceError::Refused {
            message: format!(
                "source {} cannot carry an unordered dependency between projects, because \
                 Linear types every project relation `dependency` and that is an ordering; \
                 record {near} to {far} as a dependency, or between tasks",
                self.name,
                near = near.0,
            ),
        }
    }
    /// The one edge [`Self::unordered_project_relation`] refuses, if there is one here.
    fn unordered_project_edge(edges: &[DependencyEdge]) -> Option<&DependencyEdge> {
        edges
            .iter()
            .find(|edge| edge.to.kind == ItemKind::Project && edge.kind == DependencyKind::Related)
    }
    async fn write_relations(
        &self,
        near: &NativeId,
        edges: &[DependencyEdge],
        kind: WriteKind,
    ) -> Result<(), SourceError> {
        let mut cursor: Option<Cursor> = None;
        loop {
            let data = self
                .send(
                    if matches!(kind, WriteKind::Project) {
                        PROJECT_RELATIONS
                    } else {
                        ISSUE_RELATIONS
                    },
                    json!({"id":near.0,"first":MAX_PAGE_SIZE,"after":cursor.as_ref().map(|cursor|&cursor.0)}),
                )
                .await?;
            let root = data
                .get(if matches!(kind, WriteKind::Project) {
                    "project"
                } else {
                    "issue"
                })
                .ok_or_else(|| SourceError::Malformed {
                    message: "missing relation item".into(),
                })?;
            let relations = root
                .get("relations")
                .ok_or_else(|| SourceError::Malformed {
                    message: "missing relations".into(),
                })?;
            for relation in relations
                .get("nodes")
                .and_then(Value::as_array)
                .ok_or_else(|| SourceError::Malformed {
                    message: "missing relations.nodes".into(),
                })?
            {
                let id = backend_id(relation, "id")?;
                let (query, mutation) = if matches!(kind, WriteKind::Project) {
                    (
                        graphql::PROJECT_RELATION_DELETE,
                        MutationRoot::ProjectRelationDelete,
                    )
                } else {
                    (
                        graphql::ISSUE_RELATION_DELETE,
                        MutationRoot::IssueRelationDelete,
                    )
                };
                let deleted = self.send(query, json!({"id":id})).await?;
                mutation_payload(&deleted, mutation)?;
            }
            let Some(next) = page_next(relations)? else {
                break;
            };
            cursor = Some(next);
        }
        // Linear requires an anchor at each end of a project relation and validates both
        // against an enum GraphQL cannot see: `ProjectRelationCreateInput` declares them
        // `String!` and enumerates nothing, and the field descriptions read as a choice
        // between the project and a milestone, which is not what they are. Linear's own
        // refusal enumerates them — sent `project` in both, it answered `anchorType must
        // be one of the following values: start, end, milestone` — and `milestone` needs
        // an id this source never sends, so the two whole-project anchors are the whole of
        // what it can send.
        //
        // **Which of them goes where carries the direction, and the two id slots do not.**
        // Linear stores whatever pair it is given and reads a backwards dependency as
        // readily as the right one, so acceptance settles nothing; what does is Linear's
        // own reading of a stored relation, published as the computed `ProjectFilter`
        // members `hasBlockingRelations` ("projects which are blocking") and
        // `hasBlockedByRelations` ("projects which are blocked"). Three relations between
        // two scratch projects, read back through them on 2026-09-04:
        //
        // | `projectId` | `anchorType` | `relatedProjectId` | `relatedAnchorType` | blocked | blocking |
        // | ----------- | ------------ | ------------------ | ------------------- | ------- | -------- |
        // | A           | `start`      | B                  | `end`               | A       | B        |
        // | A           | `end`        | B                  | `start`             | B       | A        |
        // | B           | `end`        | A                  | `start`             | A       | B        |
        //
        // Rows one and three exchange the ids and the anchors together and read alike;
        // rows one and two exchange only the anchors and the reading flips. So the project
        // anchored `start` is the one that waits, whichever slot it sits in, and row one is
        // what this source sends — `near`, the item that depends, in `projectId`. Linear's
        // own callers put the blocker there instead, so copying their `end`/`start` pair
        // across by position would state every dependency backwards in the workspace, and
        // nothing would refuse it.
        const NEAR_ANCHOR: &str = "start";
        const FAR_ANCHOR: &str = "end";
        for edge in edges {
            if edge.to.kind
                != match kind {
                    WriteKind::Task => ItemKind::Task,
                    WriteKind::Project => ItemKind::Project,
                }
            {
                continue;
            }
            let far = match edge.to.id().split_once(':') {
                Some((source, native)) if source == self.name.as_str() => native,
                Some(_) => continue,
                None => edge.to.id(),
            };
            // A project relation is not spelled the way an issue relation is, and this is
            // the whole of what a project's `type` may say.
            //
            // `blocks` there is what the live journey's project write was refused for
            // once the two anchors above stopped being missing: Linear answered HTTP 200
            // with `Argument Validation Error`, the message class its input validator
            // raises for a value outside an accepted set, having already accepted every
            // field of the same input by name — which is what tells that refusal apart
            // from the missing-field one before it, and what says the anchors were not the
            // cause.
            //
            // Which field, and what it takes, was measured against the real API on
            // 2026-09-04 rather than inferred. Each of `blocks`, `dependsOn`, `related`
            // and `DEPENDENCY` was refused with `property: "type"` and
            // `constraints: {"isEnum": "type must be one of the following values:
            // dependency"}`; `dependency` was accepted. That enumeration, like the
            // anchors' above, reaches this source through the validator's `extensions`;
            // see `GqlError::said`.
            //
            // A `Related` project edge is refused at the top of this function by that same
            // enumeration: it has one member and it is an ordering. An issue relation is a
            // different relation with a different set, which does include `related`.
            let relation_type = match (kind, edge.kind) {
                (WriteKind::Project, DependencyKind::Blocks) => "dependency",
                (WriteKind::Task, DependencyKind::Blocks) => "blocks",
                (WriteKind::Task, DependencyKind::Related) => "related",
                // Unreachable past `write_project`'s guard, and an error rather than a
                // skip so it stays that way: an edge dropped here would be a copy
                // reporting success for a dependency the destination does not hold.
                (WriteKind::Project, DependencyKind::Related) => {
                    return Err(self.unordered_project_relation(near, edge.to.id()));
                }
            };
            let (query, input) = if matches!(kind, WriteKind::Project) {
                (
                    graphql::PROJECT_RELATION_CREATE,
                    json!({"projectId":near.0,"relatedProjectId":far,"type":relation_type,"anchorType":NEAR_ANCHOR,"relatedAnchorType":FAR_ANCHOR}),
                )
            } else {
                (
                    graphql::ISSUE_RELATION_CREATE,
                    json!({"issueId":near.0,"relatedIssueId":far,"type":relation_type}),
                )
            };
            let data = self.send(query, json!({"input":input})).await?;
            let mutation = if matches!(kind, WriteKind::Project) {
                MutationRoot::ProjectRelationCreate
            } else {
                MutationRoot::IssueRelationCreate
            };
            let payload = mutation_payload(&data, mutation)?;
            let relation = payload
                .get(if matches!(kind, WriteKind::Project) {
                    "projectRelation"
                } else {
                    "issueRelation"
                })
                .ok_or_else(|| SourceError::Malformed {
                    message: format!("missing {} relation", mutation.as_str()),
                })?;
            backend_id(relation, "id")?;
        }
        Ok(())
    }

    async fn prepare_edges(
        &self,
        edges: &[DependencyEdge],
        kind: WriteKind,
    ) -> Result<Vec<DependencyEdge>, SourceError> {
        let mut prepared = Vec::with_capacity(edges.len());
        for edge in edges {
            let mut edge = edge.clone();
            if edge.to.kind
                == match kind {
                    WriteKind::Task => ItemKind::Task,
                    WriteKind::Project => ItemKind::Project,
                }
                && edge
                    .to
                    .id()
                    .split_once(':')
                    .is_some_and(|(source, _)| source != self.name.as_str())
            {
                // Narrowed to what this source holds — its team, and its project when it is
                // scoped to one — and, for an issue, to the ones whose description carries
                // the far end as its origin: a copy of the far end is an item of this source,
                // and an item of another team or project carrying the same origin is not one
                // this source could relate. The origin is confirmed over the parsed slot below.
                let filter = match kind {
                    WriteKind::Project => Self::narrowed(self.project_scope()),
                    WriteKind::Task => {
                        let mut parts = self.issue_scope();
                        parts.extend(slot_phrase(edge.to.id()));
                        Self::narrowed(parts)
                    }
                };
                let mut cursor: Option<Cursor> = None;
                loop {
                    let data = self.send(if matches!(kind, WriteKind::Project) { PROJECTS } else { ISSUES }, json!({"first":MAX_PAGE_SIZE,"after":cursor.as_ref().map(|cursor|&cursor.0),"filter":filter})).await?;
                    let (items, next) = if matches!(kind, WriteKind::Project) {
                        let page = connection(&data, "projects", map_project)?;
                        (
                            page.items
                                .into_iter()
                                .map(|item| (item.id, item.metadata))
                                .collect::<Vec<_>>(),
                            page.next,
                        )
                    } else {
                        let page = connection(&data, "issues", |v| {
                            map_task(v, &self.name, &self.statuses)
                        })?;
                        (
                            page.items
                                .into_iter()
                                .map(|item| (item.id, item.metadata))
                                .collect::<Vec<_>>(),
                            page.next,
                        )
                    };
                    if let Some((id, _)) = items.into_iter().find(|(_, metadata)| {
                        metadata.get("onetaskgraph.origin").and_then(Value::as_str)
                            == Some(edge.to.id())
                    }) {
                        edge.to = DependencyEndpoint::from_native(id, edge.to.kind);
                        break;
                    }
                    let Some(next) = next else { break };
                    cursor = Some(next);
                }
            }
            prepared.push(edge);
        }
        Ok(prepared)
    }
}

#[async_trait::async_trait]
impl TaskSource for LinearSource {
    fn kind(&self) -> &'static str {
        KIND
    }
    fn capabilities(&self) -> Capabilities {
        Capabilities {
            projects: Support::Native,
            documents: Support::Native,
            comments: Support::Native,
            priority: Support::Native,
            filter_by_priority: Support::Native,
            filter_by_comment_activity: Support::Native,
            filter_by_metadata: Support::Native,
            filter_by_origin: Support::Native,
            orphan_tasks: Support::Native,
            filter_by_label: Support::Native,
            filter_by_status: Support::Native,
            search_title: Support::Native,
            search_content: Support::Native,
            task_dependencies: DependencySupport::BothDirections,
            project_dependencies: DependencySupport::BothDirections,
            max_page_size: MAX_PAGE_SIZE,
        }
    }
    fn writes(&self) -> WriteSupport {
        WriteSupport::Supported
    }
    async fn health(&self) -> Result<Health, SourceError> {
        let data = self.send(VIEWER, json!({})).await?;
        str_at(
            data.get("viewer").ok_or_else(|| SourceError::Malformed {
                message: "missing viewer".into(),
            })?,
            "id",
        )?;
        Ok(Health {
            reachable: true,
            detail: None,
        })
    }
    async fn get_task(&self, id: &NativeId) -> Result<Option<Task>, SourceError> {
        Ok(self.issue_held(id).await?.map(|(task, _)| task))
    }
    async fn get_project(&self, id: &NativeId) -> Result<Option<Project>, SourceError> {
        Ok(self.project_held(id).await?.map(|(project, _)| project))
    }
    async fn query_tasks(
        &self,
        query: &TaskQuery,
        page: &PageRequest,
    ) -> Result<Page<Task>, SourceError> {
        let d=self.send(ISSUES,json!({"first":page.limit.min(MAX_PAGE_SIZE),"after":page.cursor.as_ref().map(|c|&c.0),"filter":self.issue_filter(query)?})).await?;
        let page = connection(&d, "issues", |v| map_task(v, &self.name, &self.statuses))?;
        Ok(Page {
            items: page
                .items
                .into_iter()
                .filter(|task| Self::confirms(query, task))
                .collect(),
            next: page.next,
        })
    }
    async fn query_projects(
        &self,
        query: &ProjectQuery,
        page: &PageRequest,
    ) -> Result<Page<Project>, SourceError> {
        // llmlint: ignore[changed_behavior_has_e2e] The shared CLI journey `every_complete_dataset_source_filters_projects_by_label_status_and_text` asserts that Linear status filtering returns only P-2 and reports native pushdown; this lower-level HTTP test separately asserts the serialized `started` predicate.
        let d=self.send(PROJECTS,json!({"first":page.limit.min(MAX_PAGE_SIZE),"after":page.cursor.as_ref().map(|c|&c.0),"filter":self.project_filter(&query.labels,&query.statuses)})).await?;
        let page = connection(&d, "projects", map_project)?;
        // A project's text is applied here, over the page Linear answered: `search_title` and
        // `search_content` are declared for every level, and `ProjectFilter` is not asked for
        // a description match — so the rule decides, over every row of the page.
        Ok(Page {
            items: page
                .items
                .into_iter()
                .filter(|project| {
                    query.text.as_ref().is_none_or(|text| {
                        text_holds(&project.title, project.content.as_deref(), text)
                    })
                })
                .collect(),
            next: page.next,
        })
    }
    async fn labels(&self, page: &PageRequest) -> Result<Page<Label>, SourceError> {
        let d = self
            .send(
                LABELS,
                json!({"first":page.limit.min(MAX_PAGE_SIZE),"after":page.cursor.as_ref().map(|c|&c.0)}),
            )
            .await?;
        connection(&d, "issueLabels", map_label)
    }
    async fn task_dependencies(
        &self,
        id: &NativeId,
        direction: Direction,
        page: &PageRequest,
    ) -> Result<Page<DependencyEdge>, SourceError> {
        self.dependencies(ISSUE_RELATIONS, DependencyRoot::Issue, id, direction, page)
            .await
    }
    async fn project_dependencies(
        &self,
        id: &NativeId,
        direction: Direction,
        page: &PageRequest,
    ) -> Result<Page<DependencyEdge>, SourceError> {
        self.dependencies(
            PROJECT_RELATIONS,
            DependencyRoot::Project,
            id,
            direction,
            page,
        )
        .await
    }
    async fn write_task(&self, write: &ItemWrite<Task>) -> Result<NativeId, SourceError> {
        // Before anything is read or written, because nothing Linear could answer changes
        // either: neither list may name the task itself or name one task twice, a category
        // this source has disabled is refused in the words a status write is refused with,
        // and a project other than the one this source is scoped to is refused naming both.
        let near = write.target.as_ref().unwrap_or(&write.item.id);
        for (key, entries) in [
            (TaskRef::DELIVERS_KEY, &write.item.delivers),
            (TaskRef::DELIVERED_BY_KEY, &write.item.delivered_by),
        ] {
            TaskRef::listed(key, near, Some(&self.name), entries.clone())
                .map_err(|message| SourceError::Refused { message })?;
        }
        if self.statuses.target(write.item.status.category) == StateTarget::DisabledByMapping {
            return Err(self.disabled(write.item.status.category, true));
        }
        let project = self.filed_in(write.item.project.as_ref(), "task")?;
        let edges = self
            .prepare_edges(&write.depends_on, WriteKind::Task)
            .await?;
        let team = self.team_id().await?;
        // A mapped category is written as the state the mapping names, whatever the status
        // was called where it came from; one the mapping leaves out keeps the lookup a source
        // without the key has always made, by the status's own name.
        let state = match self.statuses.target(write.item.status.category) {
            StateTarget::Named(name) => {
                self.named_state(&name.0, write.item.status.category)
                    .await?
                    .0
            }
            StateTarget::FirstOfType(_)
            | StateTarget::DisabledByMapping
            | StateTarget::NoWorkflowType => {
                self.one_id(Lookup::IssueState {
                    name: &write.item.status.name,
                    team: &team,
                })
                .await?
            }
        };
        let labels = self.label_ids(&write.item.labels, WriteKind::Task).await?;
        // The typed lists are what land, whatever the caller's own metadata held under their
        // keys: a key of either name travelling beside the field would be a second answer to
        // the same question, and the field is the one the contract names.
        let mut metadata = write.item.metadata.clone();
        set_task_list(&mut metadata, TaskRef::DELIVERS_KEY, &write.item.delivers);
        set_task_list(
            &mut metadata,
            TaskRef::DELIVERED_BY_KEY,
            &write.item.delivered_by,
        );
        let description = self.write_description(
            write.item.content.as_deref(),
            &metadata,
            &write.item.repositories,
            &edges,
            WriteKind::Task,
        )?;
        // `priority` on a create and on an update alike, `0` included: an update that left
        // it out would keep whatever the destination held, so a copy moving an issue back to
        // no priority would report success for a priority the destination still carries.
        let input = json!({"title":write.item.title,"description":description,"stateId":state,"priority":linear_priority(write.item.priority),"labelIds":labels,"projectId":project});
        let (query, variables, root) = match &write.target {
            Some(id) => (
                graphql::ISSUE_UPDATE,
                json!({"id":id.0,"input":input}),
                MutationRoot::IssueUpdate,
            ),
            None => (
                graphql::ISSUE_CREATE,
                {
                    let mut input = input;
                    input["teamId"] = Value::String(team.0);
                    json!({"input":input})
                },
                MutationRoot::IssueCreate,
            ),
        };
        let data = self.send(query, variables).await?;
        let issue =
            mutation_payload(&data, root)?
                .get("issue")
                .ok_or_else(|| SourceError::Malformed {
                    message: format!("missing {}.issue", root.as_str()),
                })?;
        let id = NativeId(backend_id(issue, "id")?.into());
        self.write_relations(&id, &edges, WriteKind::Task).await?;
        Ok(id)
    }
    async fn write_project(&self, write: &ItemWrite<Project>) -> Result<NativeId, SourceError> {
        // Before anything is read or written, and before the item's own description
        // records these edges: an edge Linear will never accept has to refuse the whole
        // write, or a copy would create the project and then fail relating it, leaving the
        // undo to clean up a write that could have been refused without a call at all.
        if let Some(edge) = Self::unordered_project_edge(&write.depends_on) {
            return Err(self.unordered_project_relation(&write.item.id, edge.to.id()));
        }
        if let Some(key) = delivery_key_in(&write.item.metadata) {
            return Err(self.undeliverable(key, "project"));
        }
        // A source scoped to one project holds that project and no other, so it writes no
        // other: a project it created would be one none of its reads could find.
        if let Some(scope) = &self.project
            && write
                .target
                .as_ref()
                .is_none_or(|target| target.0 != scope.0)
        {
            return Err(SourceError::Refused {
                message: format!(
                    "source {} is scoped to the Linear project {} and holds no other project, \
                     so it cannot write {}; next: copy the project to a source with no \
                     `project`, or copy its tasks here",
                    self.name,
                    scope.0,
                    write.target.as_ref().map_or_else(
                        || "a new one".to_owned(),
                        |target| format!("the project {}", target.0)
                    ),
                ),
            });
        }
        let edges = self
            .prepare_edges(&write.depends_on, WriteKind::Project)
            .await?;
        let team = self.team_id().await?;
        let status = self
            .one_id(Lookup::ProjectStatus(&write.item.status.name))
            .await?;
        let labels = self
            .label_ids(&write.item.labels, WriteKind::Project)
            .await?;
        let description = self.write_description(
            write.item.content.as_deref(),
            &write.item.metadata,
            &write.item.repositories,
            &edges,
            WriteKind::Project,
        )?;
        let input = json!({"name":write.item.title,"description":description,"statusId":status,"labelIds":labels});
        let (query, variables, root) = match &write.target {
            Some(id) => (
                graphql::PROJECT_UPDATE,
                json!({"id":id.0,"input":input}),
                MutationRoot::ProjectUpdate,
            ),
            None => (
                graphql::PROJECT_CREATE,
                {
                    let mut input = input;
                    input["teamIds"] = json!([team]);
                    json!({"input":input})
                },
                MutationRoot::ProjectCreate,
            ),
        };
        let data = self.send(query, variables).await?;
        let project = mutation_payload(&data, root)?
            .get("project")
            .ok_or_else(|| SourceError::Malformed {
                message: format!("missing {}.project", root.as_str()),
            })?;
        let id = NativeId(backend_id(project, "id")?.into());
        self.write_relations(&id, &edges, WriteKind::Project)
            .await?;
        Ok(id)
    }
    async fn delete_task(&self, id: &NativeId) -> Result<(), SourceError> {
        // An id naming nothing is the state this asks for, not an error — Linear reports
        // an unknown issue as an errored response rather than an unsuccessful payload, and
        // `get_task` answering `None` is what says the item is already gone.
        if self.get_task(id).await?.is_none() {
            return Ok(());
        }
        let data = self.send(graphql::ISSUE_DELETE, json!({"id":id.0})).await?;
        mutation_payload(&data, MutationRoot::IssueDelete)?;
        Ok(())
    }
    async fn delete_project(&self, id: &NativeId) -> Result<(), SourceError> {
        // An id naming nothing is the state this asks for, on exactly the terms
        // `delete_task` reads it on.
        if self.get_project(id).await?.is_none() {
            return Ok(());
        }
        let data = self
            .send(graphql::PROJECT_DELETE, json!({"id":id.0}))
            .await?;
        mutation_payload(&data, MutationRoot::ProjectDelete)?;
        Ok(())
    }
    async fn get_document(&self, id: &NativeId) -> Result<Option<Document>, SourceError> {
        // Read as an optional although the pinned `document(id:)` returns `Document!`, for
        // the reason `delete_task` records: Linear answers an id naming nothing with an
        // errored response rather than a null, and reading the null defensively is what
        // keeps a responder that does answer one from being a malformed-response failure.
        Ok(self.document_held(id).await?.map(|(document, _)| document))
    }
    async fn query_documents(
        &self,
        query: &DocumentQuery,
        page: &PageRequest,
    ) -> Result<Page<Document>, SourceError> {
        // A document's text is applied over each fetched page with the labels and the
        // orphans, by the contract's own rule: both searches are declared for every level.
        let want = page.limit.min(MAX_PAGE_SIZE) as usize;
        let mut filter = serde_json::Map::new();
        if let ProjectFilter::Is(id) = &query.project {
            if !self.in_scope(Some(id)) {
                return Ok(Page::last(Vec::new()));
            }
            filter.insert("project".into(), json!({"id": {"eq": id.0}}));
        } else if let Some(scope) = &self.project {
            filter.insert("project".into(), json!({"id": {"eq": scope.0}}));
        }
        let filter = Value::Object(filter);
        let mut items = Vec::new();
        let mut cursor = page.cursor.clone();
        loop {
            // Only what is still owed, so the predicates applied here can never make this
            // return more than the caller asked for, and never drop what it fetched.
            let first = want.saturating_sub(items.len()).max(1);
            let d = self
                .send(
                    DOCUMENTS,
                    json!({"first":first,"after":cursor.as_ref().map(|cursor|&cursor.0),"filter":filter}),
                )
                .await?;
            let fetched = connection(&d, "documents", map_document)?;
            items.extend(fetched.items.into_iter().filter(|document| {
                document_matches(document, &query.project, &query.labels)
                    && self.in_scope(document.project.as_ref())
                    && query.text.as_ref().is_none_or(|text| {
                        text_holds(&document.title, document.content.as_deref(), text)
                    })
            }));
            cursor = fetched.next;
            if cursor.is_none() || items.len() >= want {
                return Ok(Page {
                    items,
                    next: cursor,
                });
            }
        }
    }
    async fn write_document(&self, write: &ItemWrite<Document>) -> Result<NativeId, SourceError> {
        // Two refusals by name rather than two silent drops. Linear's own document type
        // has no labels and a document is not work, so neither a label nor a dependency
        // has anywhere here to land — and a copy that dropped one would report success for
        // an item the destination does not hold.
        if !write.item.labels.is_empty() {
            let named = write
                .item
                .labels
                .iter()
                .map(|label| label.name.as_str())
                .collect::<Vec<_>>()
                .join(", ");
            return Err(SourceError::Refused {
                message: format!(
                    "source {} cannot carry a document's labels, because Linear's own \
                     document type has none: {named}",
                    self.name
                ),
            });
        }
        if !write.depends_on.is_empty()
            || write
                .item
                .metadata
                .contains_key(DependencyEdge::RECORDED_KEY)
        {
            return Err(SourceError::Refused {
                message: format!(
                    "source {} cannot carry {} on a document, because a document is not \
                     work and nothing may depend on one",
                    self.name,
                    DependencyEdge::RECORDED_KEY
                ),
            });
        }
        if let Some(key) = delivery_key_in(&write.item.metadata) {
            return Err(self.undeliverable(key, "document"));
        }
        let content = Self::long_form(
            write.item.content.as_deref(),
            &write.item.metadata,
            &write.item.repositories,
            Vec::new(),
        )?;
        let project = self.filed_in(write.item.project.as_ref(), "document")?;
        let (query, variables, root) = match &write.target {
            Some(id) => {
                // A target this workspace does not hold is refused rather than created:
                // the engine established that id before asking, so an absent one is a race
                // this destination must not paper over by writing a second document.
                if self.get_document(id).await?.is_none() {
                    return Err(SourceError::Refused {
                        message: format!("source {} holds no document {}", self.name, id.0),
                    });
                }
                (
                    graphql::DOCUMENT_UPDATE,
                    json!({"id":id.0,"input":{"title":write.item.title,"content":content,"projectId":project}}),
                    MutationRoot::DocumentUpdate,
                )
            }
            None => {
                let mut input = json!({"title":write.item.title,"content":content});
                // A Linear document lives in a project, an initiative, an issue or a team.
                // One filed under no project needs the configured team to be its home, and
                // one filed under a project already has one — so the team is asked for
                // only where it is the answer, rather than made a condition of every write.
                //
                // **`projectId` is left out rather than sent as null, and that is Linear's
                // rule rather than tidiness.** `documentCreate` refuses an input that names
                // more than one home — `Exactly one of initiativeId, teamId, issueId,
                // releaseId, cycleId or projectId must be defined.` — and it counts a
                // *present* key, observed on 2026-09-04: `{projectId: null, teamId: …}` is
                // refused where `{teamId: …}` is accepted. So a document filed under no
                // project must carry no `projectId` at all. `documentUpdate` is the
                // opposite and keeps its explicit null, because there the null is the
                // instruction — it is how a document is moved out of a project, and
                // omitting the key would leave it where it was.
                match &project {
                    Some(project) => input["projectId"] = Value::String(project.clone()),
                    None => input["teamId"] = Value::String(self.team_id().await?.0),
                }
                (
                    graphql::DOCUMENT_CREATE,
                    json!({ "input": input }),
                    MutationRoot::DocumentCreate,
                )
            }
        };
        let data = self.send(query, variables).await?;
        let document = mutation_payload(&data, root)?
            .get("document")
            .ok_or_else(|| SourceError::Malformed {
                message: format!("missing {}.document", root.as_str()),
            })?;
        Ok(NativeId(backend_id(document, "id")?.into()))
    }
    async fn delete_document(&self, id: &NativeId) -> Result<(), SourceError> {
        // An id naming nothing is the state this asks for, on exactly the terms
        // `delete_task` reads it on.
        if self.get_document(id).await?.is_none() {
            return Ok(());
        }
        let data = self
            .send(graphql::DOCUMENT_DELETE, json!({"id":id.0}))
            .await?;
        mutation_payload(&data, MutationRoot::DocumentDelete)?;
        Ok(())
    }
    async fn task_comments(
        &self,
        task: &NativeId,
        page: &PageRequest,
    ) -> Result<Option<Page<Comment>>, SourceError> {
        // A page of no rows is not a page: refused here rather than sent as `last: 0`, which
        // would answer an empty page that reads as a task with no comments.
        if page.limit == 0 {
            return Err(SourceError::Config {
                message: "a page limit of 0 is not a page; ask for at least 1 comment".to_owned(),
            });
        }
        // One request rather than a task lookup and then a read: the issue the comments
        // hang off answers "no such task" by itself, on exactly the terms `get_task` reads
        // it — null, or trashed.
        let d = self
            .send(
                graphql::ISSUE_COMMENTS,
                json!({"id":task.0,"last":page.limit.min(MAX_PAGE_SIZE),"before":page.cursor.as_ref().map(|c|&c.0)}),
            )
            .await?;
        optional(&d, "issue", comment_page)
    }
    async fn add_comment(
        &self,
        task: &NativeId,
        comment: &NewComment,
    ) -> Result<Option<Comment>, SourceError> {
        // Before anything is sent, because nothing Linear could answer changes it: see the
        // ruling on the author in this crate's module documentation.
        if let Some(author) = &comment.author {
            return Err(SourceError::Refused {
                message: format!(
                    "source {} cannot post a comment as {author:?}, because Linear records the \
                     user whose API key makes the request as the author of every comment; \
                     leave --author out to post as that user",
                    self.name
                ),
            });
        }
        let Some(issue) = self.commented_issue(task).await? else {
            return Ok(None);
        };
        let data = self
            .send(
                graphql::COMMENT_CREATE,
                json!({"input":{"issueId":issue.0,"body":comment.body.as_str()}}),
            )
            .await?;
        written_comment(&data, MutationRoot::CommentCreate).map(Some)
    }
    async fn edit_comment(
        &self,
        task: &NativeId,
        comment: &NativeId,
        body: &CommentBody,
    ) -> Result<Option<Comment>, SourceError> {
        if !self.comment_is_on(task, comment).await? {
            return Ok(None);
        }
        // `body` alone: the id, the author and the time it was written are the comment's
        // own, so nothing else is sent that Linear could move.
        let data = self
            .send(
                graphql::COMMENT_UPDATE,
                json!({"id":comment.0,"input":{"body":body.as_str()}}),
            )
            .await?;
        written_comment(&data, MutationRoot::CommentUpdate).map(Some)
    }
    async fn delete_comment(
        &self,
        task: &NativeId,
        comment: &NativeId,
    ) -> Result<Option<NativeId>, SourceError> {
        if !self.comment_is_on(task, comment).await? {
            return Ok(None);
        }
        let data = self
            .send(graphql::COMMENT_DELETE, json!({"id":comment.0}))
            .await?;
        mutation_payload(&data, MutationRoot::CommentDelete)?;
        Ok(Some(comment.clone()))
    }
    async fn set_task_status(
        &self,
        id: &NativeId,
        category: StatusCategory,
    ) -> Result<Option<Status>, SourceError> {
        // Before any request: a category no workflow state has is not one Linear could
        // answer differently for another issue.
        self.writable(category)?;
        let Some(task) = self.get_task(id).await? else {
            return Ok(None);
        };
        // Already in the category asked for: its own state is left where it is. A team can
        // hold several states of one type — `In Progress` and `In Review` are both `started`
        // — and moving an issue from one to the other is a change nobody asked for.
        if task.status.category == category {
            return Ok(Some(task.status));
        }
        let (state_id, name) = self.resolve_state(category, id).await?;
        // `stateId` alone, so nothing else about the issue can move: Linear's
        // `IssueUpdateInput` makes every member optional and leaves an absent one as it was.
        let data = self
            .send(
                graphql::ISSUE_UPDATE,
                json!({"id":task.id.0,"input":{"stateId":state_id.0}}),
            )
            .await?;
        let issue = mutation_payload(&data, MutationRoot::IssueUpdate)?
            .get("issue")
            .ok_or_else(|| SourceError::Malformed {
                message: "missing issueUpdate.issue".into(),
            })?;
        backend_id(issue, "id")?;
        Ok(Some(Status { category, name }))
    }
    async fn set_task_priority(
        &self,
        id: &NativeId,
        priority: Priority,
    ) -> Result<Option<Priority>, SourceError> {
        // Read first, on exactly the terms `set_task_status` reads: Linear answers an
        // `issueUpdate` naming no issue with an errored response rather than a null, so the
        // read is what tells "no such task" from a refusal — and a trashed issue is not one
        // this source holds, so it is never written to.
        let Some(task) = self.get_task(id).await? else {
            return Ok(None);
        };
        // `priority` alone: every member of `IssueUpdateInput` is optional and Linear leaves
        // an absent one as the issue holds it, so nothing else about the issue can move.
        let data = self
            .send(
                graphql::ISSUE_PRIORITY_UPDATE,
                json!({"id":task.id.0,"input":{"priority":linear_priority(priority)}}),
            )
            .await?;
        let issue = mutation_payload(&data, MutationRoot::IssueUpdate)?
            .get("issue")
            .filter(|issue| !issue.is_null())
            .ok_or_else(|| SourceError::Malformed {
                message: "missing issueUpdate.issue".into(),
            })?;
        written_is(issue, &task.id)?;
        issue_priority(issue).map(Some)
    }
    async fn set_task_content(
        &self,
        id: &NativeId,
        content: &str,
    ) -> Result<Option<()>, SourceError> {
        // The raw description, because the metadata slot lives in that same field and has to
        // go back byte for byte: re-encoding it would be a metadata write nobody asked for.
        // The whole issue is read as `get_task` reads it first, so an issue this source could
        // not read is refused before anything is written rather than overwritten blind.
        let Some((task, description)) = self.issue_held(id).await? else {
            return Ok(None);
        };
        let issue = task.id;
        let slot = match description.as_deref() {
            Some(description) => metadata_slot(description)?.map(str::to_owned),
            None => None,
        };
        let description = match &slot {
            None => content.to_owned(),
            Some(slot) if content.is_empty() => slot.clone(),
            // The separator `long_form` writes, so a content write and a copy leave one shape.
            Some(slot) => format!("{content}\n\n{slot}"),
        };
        // Checked before anything is sent: content ending in what this source reads as its own
        // metadata slot would read back as metadata rather than as the content it was.
        if metadata_slot(&description)? != slot.as_deref() {
            return Err(SourceError::Refused {
                message: format!(
                    "this content ends in what source {} reads as its own metadata slot, so part \
                     of it would read back as metadata rather than as content; next: remove that \
                     trailing block from the content",
                    self.name
                ),
            });
        }
        // And what a read will report is exactly what was asked for, or nothing is sent.
        let (reads, _) = metadata_description(Some(description.clone()))?;
        if reads.as_deref().unwrap_or_default() != content {
            return Err(SourceError::Refused {
                message: format!(
                    "this content would read back from source {} as {:?} rather than as itself; \
                     next: change how the content ends",
                    self.name,
                    reads.as_deref().unwrap_or_default()
                ),
            });
        }
        // `description` alone, for the reason `set_task_priority` sends `priority` alone.
        let data = self
            .send(
                graphql::ISSUE_UPDATE,
                json!({"id":issue.0,"input":{"description":description}}),
            )
            .await?;
        let written = mutation_payload(&data, MutationRoot::IssueUpdate)?
            .get("issue")
            .filter(|issue| !issue.is_null())
            .ok_or_else(|| SourceError::Malformed {
                message: "missing issueUpdate.issue".into(),
            })?;
        written_is(written, &issue)?;
        Ok(Some(()))
    }
    async fn set_delivered_by(
        &self,
        id: &NativeId,
        delivered_by: &[TaskRef],
    ) -> Result<Option<()>, SourceError> {
        let entries = TaskRef::listed(
            TaskRef::DELIVERED_BY_KEY,
            id,
            Some(&self.name),
            delivered_by.to_vec(),
        )
        .map_err(|message| SourceError::Refused { message })?;
        let Some((task, description)) = self.issue_held(id).await? else {
            return Ok(None);
        };
        let (_, held) = metadata_description(description.clone())?;
        let mut slot = held.clone();
        set_task_list(&mut slot, TaskRef::DELIVERED_BY_KEY, &entries);
        // Compared as JSON rather than as the field's bytes, so a slot already holding the
        // list is not rewritten for its spelling alone.
        if slot != held {
            let rewritten = reslotted(description.as_deref(), &slot)?;
            self.write_description_alone(&task.id, rewritten.as_deref())
                .await?;
        }
        Ok(Some(()))
    }
    /// One read of the issue and, unless the key already holds the value, one `issueUpdate`
    /// carrying the description alone, whose metadata slot is the only part that moved.
    async fn set_task_metadata(
        &self,
        id: &NativeId,
        key: &MetadataKey,
        value: &Value,
    ) -> Result<Option<Task>, SourceError> {
        let Some((task, description)) = self.issue_held(id).await? else {
            return Ok(None);
        };
        let (_, mut slot) = metadata_description(description.clone())?;
        if slot.get(key.as_str()) == Some(value) {
            return Ok(Some(task));
        }
        slot.insert(key.as_str().to_owned(), value.clone());
        let rewritten = reslotted(description.as_deref(), &slot)?;
        self.write_description_alone(&task.id, rewritten.as_deref())
            .await?;
        self.get_task(&task.id)
            .await?
            .map(Some)
            .ok_or_else(|| SourceError::Malformed {
                message: format!(
                    "task {} was written and then could not be read back",
                    task.id
                ),
            })
    }
    /// One metadata key of a project, on the terms of `set_task_metadata`: one read, and one
    /// `projectUpdate` carrying the description alone.
    async fn set_project_metadata(
        &self,
        id: &NativeId,
        key: &MetadataKey,
        value: &Value,
    ) -> Result<Option<Project>, SourceError> {
        let Some((project, description)) = self.project_held(id).await? else {
            return Ok(None);
        };
        let (_, mut slot) = metadata_description(description.clone())?;
        if slot.get(key.as_str()) == Some(value) {
            return Ok(Some(project));
        }
        slot.insert(key.as_str().to_owned(), value.clone());
        let rewritten = reslotted(description.as_deref(), &slot)?;
        self.write_project_description(&project.id, rewritten.as_deref())
            .await?;
        self.get_project(&project.id)
            .await?
            .map(Some)
            .ok_or_else(|| SourceError::Malformed {
                message: format!(
                    "project {} was written and then could not be read back",
                    project.id
                ),
            })
    }
    /// One metadata key of a document, on the terms of `set_task_metadata`: one read, and one
    /// `documentUpdate` carrying the content alone.
    async fn set_document_metadata(
        &self,
        id: &NativeId,
        key: &MetadataKey,
        value: &Value,
    ) -> Result<Option<Document>, SourceError> {
        let Some((document, content)) = self.document_held(id).await? else {
            return Ok(None);
        };
        let (_, mut slot) = metadata_description(content.clone())?;
        if slot.get(key.as_str()) == Some(value) {
            return Ok(Some(document));
        }
        slot.insert(key.as_str().to_owned(), value.clone());
        let rewritten = reslotted(content.as_deref(), &slot)?;
        self.write_document_content(&document.id, rewritten.as_deref())
            .await?;
        self.get_document(&document.id)
            .await?
            .map(Some)
            .ok_or_else(|| SourceError::Malformed {
                message: format!(
                    "document {} was written and then could not be read back",
                    document.id
                ),
            })
    }
    /// One read of the issue and one `issueUpdate` carrying the description alone: the new
    /// content, and a slot whose `onetaskgraph.template` entry is `provenance` and whose every
    /// other entry is as it was. This source keeps no template answers — an issue has no room
    /// beside its description that is not the description, and answers written there would
    /// repeat what the content already says — so `answers` reaches nothing here.
    async fn set_task_rendering(
        &self,
        id: &NativeId,
        content: &str,
        provenance: &Value,
        _answers: &std::collections::BTreeMap<String, Value>,
    ) -> Result<Option<()>, SourceError> {
        let Some((task, description)) = self.issue_held(id).await? else {
            return Ok(None);
        };
        let (_, mut slot) = metadata_description(description.clone())?;
        slot.insert(MetadataKey::TEMPLATE_KEY.to_owned(), provenance.clone());
        // No read-back check, unlike a content write: this slot is never empty, it is written
        // after the content, and a read takes the last slot off first, so the content reads
        // back as itself whatever it ends in.
        let rewritten = Self::described(Some(content), &slot)?;
        if rewritten != description {
            self.write_description_alone(&task.id, rewritten.as_deref())
                .await?;
        }
        Ok(Some(()))
    }
    /// One document's rendering, on the terms of `set_task_rendering`, through one
    /// `documentUpdate` carrying the content alone.
    async fn set_document_rendering(
        &self,
        id: &NativeId,
        content: &str,
        provenance: &Value,
        _answers: &std::collections::BTreeMap<String, Value>,
    ) -> Result<Option<()>, SourceError> {
        let Some((document, held)) = self.document_held(id).await? else {
            return Ok(None);
        };
        let (_, mut slot) = metadata_description(held.clone())?;
        slot.insert(MetadataKey::TEMPLATE_KEY.to_owned(), provenance.clone());
        let rewritten = Self::described(Some(content), &slot)?;
        if rewritten != held {
            self.write_document_content(&document.id, rewritten.as_deref())
                .await?;
        }
        Ok(Some(()))
    }
    /// One read of the issue and one `issueUpdate` carrying only what differs; see
    /// `targeted_update`.
    async fn update_task(
        &self,
        id: &NativeId,
        update: &TaskUpdate,
    ) -> Result<Option<TaskUpdateOutcome>, SourceError> {
        self.targeted_update(id, update).await
    }
}

/// Refuse a narrow project or document write whose payload names another item than the one it
/// was sent for, on the terms [`written_is`] refuses an issue's.
fn acknowledged(item: &Value, asked: &NativeId, root: MutationRoot) -> Result<(), SourceError> {
    let written = backend_id(item, "id")?;
    if written == asked.0 {
        return Ok(());
    }
    Err(SourceError::Malformed {
        message: format!(
            "{} for {asked} answered with the item {written}",
            root.as_str()
        ),
    })
}

/// Refuse a narrow write's payload naming an issue other than the one it was sent for.
///
/// An `issueUpdate` answering with another issue is not this write landing, so it is reported
/// as the malformed answer it is rather than as the task having been written.
fn written_is(issue: &Value, asked: &NativeId) -> Result<(), SourceError> {
    let written = backend_id(issue, "id")?;
    if written == asked.0 {
        return Ok(());
    }
    Err(SourceError::Malformed {
        message: format!("issueUpdate for {asked} answered with the issue {written}"),
    })
}

/// Why a project and a document carry neither [`Task::delivers`] nor [`Task::delivered_by`].
///
/// Only a task delivers or is delivered, so the two reserved keys name nothing a project or a
/// document has. A task keeps both in its description's metadata slot; anything else naming
/// one is refused by name rather than written.
const NO_DELIVERY: &str = "only a task delivers or is delivered, so neither list has a place \
                           on anything else";

/// The reserved delivery key `metadata` carries, if it carries one.
fn delivery_key_in(metadata: &std::collections::BTreeMap<String, Value>) -> Option<&'static str> {
    [TaskRef::DELIVERS_KEY, TaskRef::DELIVERED_BY_KEY]
        .into_iter()
        .find(|key| metadata.contains_key(*key))
}

/// A category as the wire spells it — `in-progress`, `queued` — for a message.
fn category_word(category: StatusCategory) -> String {
    serde_json::to_value(category)
        .ok()
        .and_then(|value| value.as_str().map(str::to_owned))
        .unwrap_or_else(|| format!("{category:?}"))
}

impl LinearSource {
    /// The refusal of a category this source writes no workflow state for — before any request,
    /// because nothing Linear could answer changes it.
    fn writable(&self, category: StatusCategory) -> Result<(), SourceError> {
        match self.statuses.target(category) {
            StateTarget::DisabledByMapping => Err(self.disabled(category, true)),
            StateTarget::NoWorkflowType => Err(self.disabled(category, false)),
            StateTarget::Named(_) | StateTarget::FirstOfType(_) => Ok(()),
        }
    }

    /// The refusal of a disabled category, saying which of the two it is.
    fn disabled(&self, category: StatusCategory, configured: bool) -> SourceError {
        let word = category_word(category);
        SourceError::Refused {
            message: if configured {
                format!(
                    "source {} cannot set a task's status to {word}: that category is disabled \
                     for this source, because its status_mapping sets {word} to null; map \
                     status_mapping.{word} to one of the team's workflow states to write it",
                    self.name
                )
            } else {
                format!(
                    "source {} cannot set a task's status to {word}: that category is disabled \
                     for this source, because Linear has no workflow state of that kind — its \
                     workflow states are triage, backlog, unstarted, started, completed and \
                     canceled; choose backlog, todo, in-progress, done or cancelled, or map \
                     status_mapping.{word} to one of the team's workflow states",
                    self.name
                )
            },
        }
    }

    /// The team's workflow state a status of `category` is written as — the first Linear lists
    /// of that type, and deliberately no choice beyond that: every state of this type reads
    /// back as the category asked for, which is the whole of what a status write owes, and
    /// nothing a category carries says which of several the caller meant — and its name.
    async fn state_of(
        &self,
        category: StatusCategory,
        state_type: &str,
        task: &NativeId,
    ) -> Result<(NativeId, String), SourceError> {
        let team = self.team_id().await?;
        let data = self
            .send(
                graphql::ISSUE_STATE_OF_TYPE,
                json!({"type":state_type,"team":team.0}),
            )
            .await?;
        let nodes = data
            .get("workflowStates")
            .and_then(|v| v.get("nodes"))
            .and_then(Value::as_array)
            .ok_or_else(|| SourceError::Malformed {
                message: "missing workflowStates.nodes".into(),
            })?;
        let Some(state) = nodes.first() else {
            return Err(SourceError::Refused {
                message: format!(
                    "source {} cannot set task {} to {}: its configured team has no workflow \
                     state of type {state_type}; add one to the team in Linear",
                    self.name,
                    task.0,
                    category_word(category)
                ),
            });
        };
        Ok((
            NativeId(backend_id(state, "id")?.into()),
            str_at(state, "name")?.to_owned(),
        ))
    }

    /// The refusal a write naming `named` — a field or a reserved key — on a `what` gets.
    fn undeliverable(&self, named: &str, what: &str) -> SourceError {
        SourceError::Refused {
            message: format!(
                "source {} cannot carry {named} on a {what}: {NO_DELIVERY}; write the {what} \
                 without it",
                self.name
            ),
        }
    }

    /// The backend id of the issue `task` names, or `None` when this source holds no such
    /// task — resolved by `get_task` itself, so a comment call and a task read cannot
    /// disagree about whether a task is there.
    ///
    /// The id Linear answers with rather than the one asked for, because `issue(id:)` also
    /// takes an identifier such as `ENG-1`, and the comment's own `issue{id}` is compared
    /// against — and a comment is created on — the backend id.
    async fn commented_issue(&self, task: &NativeId) -> Result<Option<NativeId>, SourceError> {
        Ok(self.get_task(task).await?.map(|task| task.id))
    }

    /// Whether `comment` is a comment on the issue `task` names.
    ///
    /// Asked before any edit or removal, so an id belonging to another issue — or to no
    /// issue, or to nothing — is answered as no such comment without a mutation reaching
    /// Linear. `commentUpdate` and `commentDelete` address a comment by its id alone, so
    /// without this a task named in error would edit or remove somebody else's comment.
    async fn comment_is_on(
        &self,
        task: &NativeId,
        comment: &NativeId,
    ) -> Result<bool, SourceError> {
        let Some(issue) = self.commented_issue(task).await? else {
            return Ok(false);
        };
        let data = self.send(graphql::COMMENT, json!({"id":comment.0})).await?;
        Ok(optional(&data, "comment", comment_issue)?.flatten() == Some(issue))
    }
}

impl LinearSource {
    /// One issue as this source reads it, beside its raw `description` — or `None` for an
    /// issue Linear does not hold, has trashed, or that is outside the project this source is
    /// scoped to.
    ///
    /// Every read of one issue goes through here, so a status, a content, a metadata and a
    /// rendering write all answer "no such task" on exactly the terms `get_task` does.
    async fn issue_held(
        &self,
        id: &NativeId,
    ) -> Result<Option<(Task, Option<String>)>, SourceError> {
        let data = self.send(ISSUE, json!({"id":id.0})).await?;
        Ok(optional(&data, "issue", |v| {
            Ok((
                map_task(v, &self.name, &self.statuses)?,
                optional_string(v, "description")?,
            ))
        })?
        .filter(|(task, _)| self.in_scope(task.project.as_ref())))
    }

    /// One project and its raw `description`, on the terms of [`Self::issue_held`]: a source
    /// scoped to one project holds that one alone.
    async fn project_held(
        &self,
        id: &NativeId,
    ) -> Result<Option<(Project, Option<String>)>, SourceError> {
        let data = self.send(PROJECT, json!({"id":id.0})).await?;
        Ok(optional(&data, "project", |v| {
            Ok((map_project(v)?, optional_string(v, "description")?))
        })?
        .filter(|(project, _)| self.in_scope(Some(&project.id))))
    }

    /// One document and its raw `content`, on the terms of [`Self::issue_held`].
    async fn document_held(
        &self,
        id: &NativeId,
    ) -> Result<Option<(Document, Option<String>)>, SourceError> {
        // Read as an optional although the pinned `document(id:)` returns `Document!`, for
        // the reason `delete_task` records: Linear answers an id naming nothing with an
        // errored response rather than a null, and reading the null defensively is what
        // keeps a responder that does answer one from being a malformed-response failure.
        let data = self.send(DOCUMENT, json!({"id":id.0})).await?;
        Ok(optional(&data, "document", |v| {
            Ok((map_document(v)?, optional_string(v, "content")?))
        })?
        .filter(|(document, _)| self.in_scope(document.project.as_ref())))
    }

    /// Send one issue's new `description` alone, and nothing else about it.
    async fn write_description_alone(
        &self,
        id: &NativeId,
        description: Option<&str>,
    ) -> Result<(), SourceError> {
        let data = self
            .send(
                graphql::ISSUE_UPDATE,
                json!({"id":id.0,"input":{"description":description}}),
            )
            .await?;
        let issue = mutation_payload(&data, MutationRoot::IssueUpdate)?
            .get("issue")
            .filter(|issue| !issue.is_null())
            .ok_or_else(|| SourceError::Malformed {
                message: "missing issueUpdate.issue".into(),
            })?;
        written_is(issue, id)
    }

    /// Send one project's new `description` alone.
    async fn write_project_description(
        &self,
        id: &NativeId,
        description: Option<&str>,
    ) -> Result<(), SourceError> {
        let data = self
            .send(
                graphql::PROJECT_UPDATE,
                json!({"id":id.0,"input":{"description":description}}),
            )
            .await?;
        let project = mutation_payload(&data, MutationRoot::ProjectUpdate)?
            .get("project")
            .filter(|project| !project.is_null())
            .ok_or_else(|| SourceError::Malformed {
                message: "missing projectUpdate.project".into(),
            })?;
        acknowledged(project, id, MutationRoot::ProjectUpdate)
    }

    /// Send one document's new `content` alone.
    async fn write_document_content(
        &self,
        id: &NativeId,
        content: Option<&str>,
    ) -> Result<(), SourceError> {
        let data = self
            .send(
                graphql::DOCUMENT_UPDATE,
                json!({"id":id.0,"input":{"content":content}}),
            )
            .await?;
        let document = mutation_payload(&data, MutationRoot::DocumentUpdate)?
            .get("document")
            .filter(|document| !document.is_null())
            .ok_or_else(|| SourceError::Malformed {
                message: "missing documentUpdate.document".into(),
            })?;
        acknowledged(document, id, MutationRoot::DocumentUpdate)
    }

    /// The configured team's workflow states, every one with its type, in Linear's order.
    async fn team_states(&self) -> Result<Vec<TeamState>, SourceError> {
        let team = self.team_id().await?;
        let data = self
            .send(graphql::TEAM_WORKFLOW_STATES, json!({"team":team.0}))
            .await?;
        data.get("workflowStates")
            .and_then(|v| v.get("nodes"))
            .and_then(Value::as_array)
            .ok_or_else(|| SourceError::Malformed {
                message: "missing workflowStates.nodes".into(),
            })?
            .iter()
            .map(|node| {
                Ok(TeamState {
                    id: NativeId(backend_id(node, "id")?.into()),
                    name: StateName::try_from(str_at(node, "name")?.to_owned())
                        .map_err(|message| SourceError::Malformed { message })?,
                    kind: str_at(node, "type")?.to_owned(),
                })
            })
            .collect()
    }

    /// The team's workflow state `status_mapping` sends `category` to, by its name — or the
    /// refusal naming the state, the team and the category when the team has none so named.
    async fn named_state(
        &self,
        name: &str,
        category: StatusCategory,
    ) -> Result<(NativeId, String), SourceError> {
        self.team_states()
            .await?
            .into_iter()
            .find(|state| state.name.0.eq_ignore_ascii_case(name))
            .map(|state| (state.id, state.name.0))
            .ok_or_else(|| SourceError::Refused {
                message: format!(
                    "source {} maps status {} to the workflow state {name:?}, which team {} \
                     does not have; next: add {name:?} to that team in Linear's team settings, \
                     or point status_mapping.{} of this source at a state the team has",
                    self.name,
                    category_word(category),
                    self.team.as_ref().map_or("(none)", |team| team.0.as_str()),
                    category_word(category)
                ),
            })
    }

    /// The workflow state a status of `category` is written as — the state the mapping names,
    /// or the team's first of the category's type — and its name.
    async fn resolve_state(
        &self,
        category: StatusCategory,
        task: &NativeId,
    ) -> Result<(NativeId, String), SourceError> {
        match self.statuses.target(category) {
            StateTarget::Named(name) => self.named_state(&name.0, category).await,
            StateTarget::FirstOfType(kind) => self.state_of(category, kind.as_str(), task).await,
            StateTarget::DisabledByMapping => Err(self.disabled(category, true)),
            StateTarget::NoWorkflowType => Err(self.disabled(category, false)),
        }
    }

    /// What `sources fields` reports: each state the mapping names, present or missing on
    /// the team, with its type.
    async fn workflow_states(&self) -> Result<WorkflowStatesReport, SourceError> {
        let held = self.team_states().await?;
        let states = self
            .statuses
            .named()
            .map(|(category, name)| {
                let found = held
                    .iter()
                    .find(|state| state.name.0.eq_ignore_ascii_case(&name.0));
                MappedWorkflowState {
                    category,
                    state: name.clone(),
                    found: found.map_or(Found::Missing, |state| Found::Present(state.kind.clone())),
                }
            })
            .collect();
        // `team_states` has just resolved the configured team, so there is one.
        let team = self.team.clone().ok_or_else(|| SourceError::Refused {
            message: format!(
                "source {} needs config.team to report its states",
                self.name
            ),
        })?;
        Ok(WorkflowStatesReport {
            source: self.name.clone(),
            team,
            states,
        })
    }
}

/// One workflow state of the configured team.
struct TeamState {
    id: NativeId,
    name: StateName,
    kind: String,
}

/// `held` with its trailing metadata slot replaced by one holding exactly `slot`, and every
/// byte above the slot exactly as it was — or with a slot appended after one blank line where
/// it had none, and the slot taken off, with the one blank line that set it off, where `slot`
/// is empty.
///
/// The one place a narrow metadata write composes a long-form field, so a metadata set, a
/// `delivered_by` write and a copy-link record move nothing a person wrote.
fn reslotted(
    held: Option<&str>,
    slot: &std::collections::BTreeMap<String, Value>,
) -> Result<Option<String>, SourceError> {
    let held = held.unwrap_or_default();
    let Some((start, _, _)) = slot_bounds(held)? else {
        return LinearSource::described((!held.is_empty()).then_some(held), slot);
    };
    let above = &held[..start];
    if slot.is_empty() {
        let visible = above
            .strip_suffix("\n\n")
            .or_else(|| above.strip_suffix('\n'))
            .unwrap_or(above);
        return Ok((!visible.is_empty()).then(|| visible.to_owned()));
    }
    Ok(Some(format!("{above}{}", slot_text(slot)?)))
}

/// Hold `entries` under `key` in one slot's metadata, or no such key when there are none.
fn set_task_list(
    metadata: &mut std::collections::BTreeMap<String, Value>,
    key: &str,
    entries: &[TaskRef],
) {
    if entries.is_empty() {
        metadata.remove(key);
    } else {
        metadata.insert(
            key.to_owned(),
            Value::Array(
                entries
                    .iter()
                    .map(|entry| Value::String(entry.as_str().to_owned()))
                    .collect(),
            ),
        );
    }
}

/// One page of an issue's comments, oldest first.
///
/// Linear answered newest first, walking backwards from `before`, so the page is reversed
/// and the next cursor is the one *behind* it; see the ruling on comments in this crate's
/// module documentation for why the walk runs that way.
fn comment_page(v: &Value) -> Result<Page<Comment>, SourceError> {
    let c = v.get("comments").ok_or_else(|| SourceError::Malformed {
        message: "missing comments connection".into(),
    })?;
    let mut items = c
        .get("nodes")
        .and_then(Value::as_array)
        .ok_or_else(|| SourceError::Malformed {
            message: "missing comment nodes".into(),
        })?
        .iter()
        .map(map_comment)
        .collect::<Result<Vec<_>, _>>()?;
    items.reverse();
    let info = c.get("pageInfo").ok_or_else(|| SourceError::Malformed {
        message: "missing pageInfo".into(),
    })?;
    let older = info
        .get("hasPreviousPage")
        .and_then(Value::as_bool)
        .ok_or_else(|| SourceError::Malformed {
            message: "missing boolean pageInfo.hasPreviousPage".into(),
        })?;
    let next = if older {
        Some(Cursor(str_at(info, "startCursor")?.into()))
    } else {
        None
    };
    Ok(Page { items, next })
}

fn map_comment(v: &Value) -> Result<Comment, SourceError> {
    let author = match v.get("user") {
        None => {
            return Err(SourceError::Malformed {
                message: "missing comment user field".into(),
            });
        }
        // An integration or a bot: Linear names no user, and this source invents none.
        Some(Value::Null) => None,
        Some(user) => Some(str_at(user, "displayName")?.to_owned()),
    };
    Ok(Comment {
        id: NativeId(backend_id(v, "id")?.into()),
        author,
        created_at: time(v, "createdAt")?,
        updated_at: time(v, "updatedAt")?,
        body: str_at(v, "body")?.into(),
        url: optional_string(v, "url")?,
    })
}

/// The comment a `commentCreate` or `commentUpdate` answered with, as Linear now holds it.
fn written_comment(data: &Value, root: MutationRoot) -> Result<Comment, SourceError> {
    let comment = mutation_payload(data, root)?
        .get("comment")
        .ok_or_else(|| SourceError::Malformed {
            message: format!("missing {}.comment", root.as_str()),
        })?;
    map_comment(comment)
}

/// The issue a comment is on, or `None` for a comment on something else — a project, a
/// document, an update — which is a comment no task of this source has.
fn comment_issue(v: &Value) -> Result<Option<NativeId>, SourceError> {
    match v.get("issue") {
        None => Err(SourceError::Malformed {
            message: "missing comment issue field".into(),
        }),
        Some(Value::Null) => Ok(None),
        Some(issue) => Ok(Some(NativeId(backend_id(issue, "id")?.into()))),
    }
}

/// Linear relates one Linear item to another and nothing else, so an edge whose far end
/// is in a different source is the one edge no `relations` entry can hold. Those edges
/// are read from the near item's own [`DependencyEdge::RECORDED_KEY`] metadata, and they
/// are served *after* the native relations are spent: a page under this cursor is the
/// recorded tail of the same walk, which keeps the native pages exactly what they were.
const RECORDED_CURSOR: &str = "onetaskgraph.depends_on:";

impl LinearSource {
    async fn dependencies(
        &self,
        query: &str,
        root: DependencyRoot,
        id: &NativeId,
        direction: Direction,
        page: &PageRequest,
    ) -> Result<Page<DependencyEdge>, SourceError> {
        let limit = page.limit.min(MAX_PAGE_SIZE);
        let cursor = page.cursor.as_ref().map(|c| c.0.as_str());
        if let Some(offset) = cursor.and_then(|c| c.strip_prefix(RECORDED_CURSOR)) {
            // This cursor resumes the *forward* tail and only a forward walk ever issues
            // one, so a reverse read carrying it is resuming a walk it did not come from.
            // Serving it would answer a reverse read with forward edges, which is the one
            // thing a recorded edge must never do — its reverse is derived from the far
            // end and is never written down here.
            if direction != Direction::DependsOn {
                return Err(SourceError::Malformed {
                    message: format!(
                        "{RECORDED_CURSOR}{offset} resumes recorded forward edges, which a                          reverse dependency read never issues; resume it in the direction                          that reported it"
                    ),
                });
            }
            let offset: usize = offset.parse().map_err(|_| SourceError::Malformed {
                message: format!("{RECORDED_CURSOR}{offset} is not a recorded-edge cursor"),
            })?;
            let d = self
                .send(query, json!({"id":id.0,"first":1,"after":null}))
                .await?;
            return Ok(recorded_page(
                recorded(&d, root, id, &self.name)?,
                offset,
                limit as usize,
            ));
        }
        let d = self
            .send(query, json!({"id":id.0,"first":limit,"after":cursor}))
            .await?;
        let mut answered = relation_page(&d, root, id, direction)?;
        // Only forwards: the reverse of a recorded edge is derived from the far end, never
        // written down on the near item.
        if answered.next.is_none()
            && direction == Direction::DependsOn
            && !recorded(&d, root, id, &self.name)?.is_empty()
        {
            answered.next = Some(Cursor(format!("{RECORDED_CURSOR}0")));
        }
        Ok(answered)
    }
}

fn recorded(
    d: &Value,
    root: DependencyRoot,
    id: &NativeId,
    name: &SourceName,
) -> Result<Vec<DependencyEdge>, SourceError> {
    let item = d.get(root.as_str()).ok_or_else(|| SourceError::Malformed {
        message: format!("missing {}", root.as_str()),
    })?;
    let (_, metadata) = metadata_description(optional_string(item, "description")?)?;
    // `relations` on an issue holds issues and on a project holds projects, both of this
    // workspace — so a same-kind far end in this same source is one Linear itself was
    // supposed to hold, and the key is refused rather than quietly read, whether the entry
    // left the source out or spelled this one.
    DependencyEdge::recorded(
        &metadata,
        id,
        root.item_kind(),
        name,
        Some(root.item_kind()),
    )
    .map_err(|message| SourceError::Malformed { message })
}

fn recorded_page(edges: Vec<DependencyEdge>, offset: usize, limit: usize) -> Page<DependencyEdge> {
    let total = edges.len();
    let items: Vec<DependencyEdge> = edges.into_iter().skip(offset).take(limit.max(1)).collect();
    let end = offset.saturating_add(items.len());
    Page {
        items,
        next: (end < total).then(|| Cursor(format!("{RECORDED_CURSOR}{end}"))),
    }
}

// llmlint: ignore-block[contracts_have_one_source_or_a_drift_gate] Linear's workflow-state strings follow the accepted 2026-08-24 contract; its authoritative enum is exposed only through an authenticated unversioned explorer, while real-HTTP tests cover every serialized and parsed value.
/// A category as `WorkflowState.type` spells it — the vocabulary an **issue**'s state has.
///
/// Linear's workflow states are triage, backlog, unstarted, started, completed and
/// canceled. None of them is a draft, so `Draft` narrows to nothing exactly as `Unknown`
/// does rather than filtering on a state Linear does not have.
///
/// Linear has no type for work that is claimed and not yet started: `unstarted` is `todo` and
/// `started` is `in-progress`, and a Linear issue reads back as one of those. So `queued`
/// narrows to nothing by type, exactly as `draft` does — mapping it onto either neighbour would
/// have a `queued` filter return an item that reads back as `todo` or `in-progress`, which is
/// capability rule 1 broken. A state `status_mapping` names is what reads as either.
fn workflow_state_types(s: &StatusCategory) -> Vec<&'static str> {
    WorkflowType::of(*s)
        .map(WorkflowType::as_str)
        .into_iter()
        .collect()
}
/// A category as `ProjectStatus.type` spells it — a **different** vocabulary, and a
/// different enum: Linear declares that field `ProjectStatusType!`, whose members are
/// backlog, planned, started, paused, completed and canceled.
///
/// Two of them have no issue counterpart and are why this cannot be the function above.
/// `planned` is where `unstarted` would be, so it is what `Todo` narrows to; a project
/// filtered with `unstarted` matches nothing and is refused by nothing, which is how this
/// went unnoticed. And `paused` is a project that has started and is neither finished nor
/// cancelled, so it reads as in progress — the same reading [`status`] gives it, which is
/// what keeps this narrowing and that mapping the same claim rather than two.
fn project_status_types(s: &StatusCategory) -> Vec<&'static str> {
    match s {
        StatusCategory::Draft => vec![],
        StatusCategory::Backlog => vec!["backlog"],
        StatusCategory::Todo => vec!["planned"],
        // No `ProjectStatusType` is claimed-and-not-started either, so `queued` narrows to
        // nothing here for the reason it does for an issue above.
        StatusCategory::Queued => vec![],
        StatusCategory::InProgress => vec!["started", "paused"],
        StatusCategory::Done => vec!["completed"],
        StatusCategory::Cancelled => vec!["canceled"],
        StatusCategory::Unknown => vec![],
    }
}
/// The category a Linear status name and type normalise to, at either level.
///
/// One mapper for both vocabularies, because the two are disjoint where they differ: no
/// issue is ever `planned` or `paused`, and no project is ever `unstarted` or `triage`. It
/// is the inverse of [`workflow_state_types`] and [`project_status_types`] together, and
/// has to stay so: a category this reports and that filter cannot ask for is capability
/// rule 1 broken, and the row would go missing rather than be refused.
///
/// **It never answers `Queued` or `Draft`**, and that is the other half of the same claim:
/// both filters narrow those two to nothing, because no Linear state or project status means
/// either, so a row this reported as one would be a row no filter for it could return. A type
/// Linear does not document — even one spelled `queued` — is `Unknown`, never a guess. An
/// issue reads as `queued` or `draft` only through [`issue_status`], at a state
/// `status_mapping` names, and the status filter then narrows that category by the same name.
fn status(v: &Value) -> Result<Status, SourceError> {
    let name = str_at(v, "name")?.into();
    let category = match str_at(v, "type")? {
        "backlog" => StatusCategory::Backlog,
        "unstarted" | "planned" => StatusCategory::Todo,
        "started" | "paused" => StatusCategory::InProgress,
        "completed" => StatusCategory::Done,
        "canceled" => StatusCategory::Cancelled,
        _ => StatusCategory::Unknown,
    };
    Ok(Status { category, name })
}
// llmlint: ignore-end[contracts_have_one_source_or_a_drift_gate]
/// The status an issue's workflow state reads as, through this instance's `status_mapping`.
///
/// A state whose name the mapping names reads as that category, under the state's own name —
/// which is what lets two states of one type, `Todo` and `Queued`, read as two categories. Every
/// other state reads by its type exactly as [`status`] reads it, so a team's review states, its
/// triage state and every state the mapping leaves out read as they did without one.
fn issue_status(v: &Value, statuses: &StatusMapping) -> Result<Status, SourceError> {
    let name = str_at(v, "name")?;
    match statuses.category_of(name) {
        Some(category) => Ok(Status {
            category,
            name: name.to_owned(),
        }),
        None => status(v),
    }
}
fn str_at<'a>(v: &'a Value, k: &str) -> Result<&'a str, SourceError> {
    v.get(k)
        .and_then(Value::as_str)
        .ok_or_else(|| SourceError::Malformed {
            message: format!("missing string field {k}"),
        })
}
fn map_label(v: &Value) -> Result<Label, SourceError> {
    Ok(Label {
        id: NativeId(str_at(v, "id")?.into()),
        name: str_at(v, "name")?.into(),
        color: optional_string(v, "color")?,
    })
}
fn labels_of(v: &Value) -> Result<Vec<Label>, SourceError> {
    v.get("nodes")
        .and_then(Value::as_array)
        .ok_or_else(|| SourceError::Malformed {
            message: "missing label nodes".into(),
        })?
        .iter()
        .map(map_label)
        .collect()
}
fn time(v: &Value, k: &str) -> Result<Option<DateTime<Utc>>, SourceError> {
    optional_str(v, k)?
        .map(|s| {
            s.parse().map_err(|e| SourceError::Malformed {
                message: format!("invalid {k}: {e}"),
            })
        })
        .transpose()
}
/// One issue as a task, `source` being this source's configured name.
///
/// The name is what lets [`TaskRef::listed`] tell `work:I-1` on the issue `I-1` of the
/// source `work` apart as that issue itself, rather than recognising only the bare spelling.
fn map_task(v: &Value, source: &SourceName, statuses: &StatusMapping) -> Result<Task, SourceError> {
    let (content, mut metadata) = metadata_description(optional_string(v, "description")?)?;
    let repositories = Repository::from_metadata(&metadata)
        .map_err(|message| SourceError::Malformed { message })?;
    let url = optional_string(v, "url")?;
    let id = NativeId(str_at(v, "id")?.into());
    // Taken out of the caller's metadata as they are read: a reserved key is this product's,
    // and reporting it there as well would hand a consumer two spellings of one list.
    let delivers = delivery_list(&mut metadata, TaskRef::DELIVERS_KEY, &id, source)?;
    let delivered_by = delivery_list(&mut metadata, TaskRef::DELIVERED_BY_KEY, &id, source)?;
    Ok(Task {
        id,
        // `Issue.identifier` is `String!` and every read of an issue selects it, so a
        // response without one is a response this source cannot read rather than an issue
        // with no handle — Linear gives every issue one.
        key: Some(str_at(v, "identifier")?.into()),
        title: str_at(v, "title")?.into(),
        content,
        status: issue_status(
            v.get("state").ok_or_else(|| SourceError::Malformed {
                message: "missing state".into(),
            })?,
            statuses,
        )?,
        priority: issue_priority(v)?,
        labels: labels_of(v.get("labels").ok_or_else(|| SourceError::Malformed {
            message: "missing labels".into(),
        })?)?,
        project: filed_under(v)?,
        location: web_address(url.as_deref()),
        url,
        created_at: time(v, "createdAt")?,
        updated_at: time(v, "updatedAt")?,
        metadata,
        repositories,
        delivers,
        delivered_by,
    })
}
/// A priority as Linear's `Issue.priority` and its two input members spell it.
///
/// Linear's own scale, as its published schema describes the field: `0` is no priority,
/// `1` urgent, `2` high, `3` normal and `4` low. Normal is this contract's `medium`.
const fn linear_priority(priority: Priority) -> u8 {
    match priority {
        Priority::None => 0,
        Priority::Urgent => 1,
        Priority::High => 2,
        Priority::Medium => 3,
        Priority::Low => 4,
    }
}

/// The priority an issue carries, read from `Issue.priority`.
///
/// Linear declares that field `Float!` while its inputs take an `Int`, so `2` and `2.0` are
/// the same answer. Anything else — absent, null, fractional, or outside `0` to `4` — is a
/// response this source cannot read, never a guess at the nearest level: a priority reported
/// that a filter for it could not find is capability rule 1 broken.
fn issue_priority(v: &Value) -> Result<Priority, SourceError> {
    let raw = v.get("priority").ok_or_else(|| SourceError::Malformed {
        message: "missing number field priority".into(),
    })?;
    let level = raw.as_f64().filter(|level| level.fract() == 0.0);
    Priority::ALL
        .into_iter()
        .find(|priority| level == Some(f64::from(linear_priority(*priority))))
        .ok_or_else(|| SourceError::Malformed {
            message: format!(
                "field priority is {raw}, which is none of Linear's priorities 0 (none), \
                 1 (urgent), 2 (high), 3 (normal) and 4 (low)"
            ),
        })
}

/// One delivery list read out of an issue's metadata slot, and removed from it.
///
/// An entry that is not a task id, that names the issue itself, or that repeats is a
/// malformed response naming the task and the entry, never a list quietly shortened.
fn delivery_list(
    metadata: &mut std::collections::BTreeMap<String, Value>,
    key: &str,
    task: &NativeId,
    source: &SourceName,
) -> Result<Vec<TaskRef>, SourceError> {
    let held = metadata.remove(key);
    TaskRef::from_value(key, task, Some(source), held.as_ref())
        .map_err(|message| SourceError::Malformed { message })
}
/// Remove the two delivery keys from a project's or a document's metadata.
///
/// Neither is work that delivers anything, so a key there names nothing this contract has,
/// and it is not the caller's free metadata either: it is this product's reserved spelling.
fn strip_delivery_keys(metadata: &mut std::collections::BTreeMap<String, Value>) {
    metadata.remove(TaskRef::DELIVERS_KEY);
    metadata.remove(TaskRef::DELIVERED_BY_KEY);
}
fn map_project(v: &Value) -> Result<Project, SourceError> {
    let (content, mut metadata) = metadata_description(optional_string(v, "description")?)?;
    strip_delivery_keys(&mut metadata);
    let repositories = Repository::from_metadata(&metadata)
        .map_err(|message| SourceError::Malformed { message })?;
    let url = optional_string(v, "url")?;
    Ok(Project {
        id: NativeId(str_at(v, "id")?.into()),
        title: str_at(v, "name")?.into(),
        content,
        status: status(v.get("status").ok_or_else(|| SourceError::Malformed {
            message: "missing status".into(),
        })?)?,
        labels: labels_of(v.get("labels").ok_or_else(|| SourceError::Malformed {
            message: "missing project labels".into(),
        })?)?,
        location: web_address(url.as_deref()),
        url,
        created_at: time(v, "createdAt")?,
        updated_at: time(v, "updatedAt")?,
        metadata,
        repositories,
    })
}

/// Where a Linear entity is: the web address Linear itself reports for it, as a link.
///
/// Every issue, project and document of a Linear workspace has a page a person can open,
/// so this source says so for all three — the counterpart of a folder of Markdown
/// reporting the path of the file behind an item. A source that reported nothing here is
/// what leaves a reader holding an opaque id, and `None` is reserved for the case Linear
/// really did not say, which is not the same as saying the entity is nowhere.
fn web_address(url: Option<&str>) -> Option<Location> {
    url.map(|url| Location::Url(url.to_owned()))
}

/// The project a Linear item is filed under, or `None` for one filed under nothing.
///
/// One reader for issues and documents alike, because the field is the same field: an
/// absent `project` key is a malformed response, a null one is an orphan.
fn filed_under(v: &Value) -> Result<Option<NativeId>, SourceError> {
    match v.get("project") {
        None => Err(SourceError::Malformed {
            message: "missing project field".into(),
        }),
        Some(Value::Null) => Ok(None),
        Some(project) => Ok(Some(NativeId(str_at(project, "id")?.into()))),
    }
}

fn map_document(v: &Value) -> Result<Document, SourceError> {
    let (content, mut metadata) = metadata_description(optional_string(v, "content")?)?;
    strip_delivery_keys(&mut metadata);
    let repositories = Repository::from_metadata(&metadata)
        .map_err(|message| SourceError::Malformed { message })?;
    let url = optional_string(v, "url")?;
    Ok(Document {
        id: NativeId(str_at(v, "id")?.into()),
        title: str_at(v, "title")?.into(),
        content,
        project: filed_under(v)?,
        // Linear's `Document` carries no labels, and that is the published schema rather
        // than a gap here: the types of it that carry `labels` are `Issue`, `Project`,
        // `Team`, `Initiative` and `Organization`. Reporting none is what a source with no
        // native slot owes; standing one up beside a first-class type is what this source
        // exists not to do, and `write_document` refuses a label by name for the same
        // reason rather than dropping it.
        labels: Vec::new(),
        location: web_address(url.as_deref()),
        url,
        created_at: time(v, "createdAt")?,
        updated_at: time(v, "updatedAt")?,
        metadata,
        repositories,
    })
}

/// Whether this document satisfies the predicates this source applies to a fetched page.
///
/// Two of them reach a page rather than the `documents(filter:)` variables, and each for a
/// reason of Linear's own. `DocumentFilter.project` is a `ProjectFilter` where
/// `IssueFilter.project` is a `NullableProjectFilter`, so only the issue side can be asked
/// for the items belonging to no project. And a Linear document carries no label at all,
/// so a query demanding one keeps nothing and a query excluding one keeps everything —
/// which is this source *applying* the predicate it declares native, over the labels the
/// document really has, rather than ignoring it.
fn document_matches(document: &Document, project: &ProjectFilter, labels: &LabelFilter) -> bool {
    let carries = |name: &String| {
        document
            .labels
            .iter()
            .any(|label| label.name.eq_ignore_ascii_case(name))
    };
    let filed = match project {
        ProjectFilter::Any => true,
        ProjectFilter::Orphans => document.project.is_none(),
        ProjectFilter::Is(id) => document.project.as_ref() == Some(id),
    };
    filed
        && (labels.any_of.is_empty() || labels.any_of.iter().any(&carries))
        && labels.all_of.iter().all(&carries)
        && !labels.none_of.iter().any(&carries)
}

fn optional<T>(
    d: &Value,
    k: &str,
    f: impl Fn(&Value) -> Result<T, SourceError>,
) -> Result<Option<T>, SourceError> {
    match d.get(k) {
        None => Err(SourceError::Malformed {
            message: format!("missing {k}"),
        }),
        Some(Value::Null) => Ok(None),
        // An item Linear no longer shows is not an item this source holds, and Linear says
        // so with `archivedAt` rather than by answering null.
        //
        // **None of Linear's three `delete` verbs removes anything.** `issueDelete`,
        // `projectDelete` and `documentDelete` move the item to the trash: observed on
        // 2026-09-04, each answered `success: true` and the item still read back by id,
        // carrying `archivedAt` and `trashed: true`. Its separate *archive* verb is a third
        // state — `archivedAt` set, `trashed` null — and Linear excludes both from every
        // connection, so `issues`, `projects` and `documents` had already stopped returning
        // them while a read by id still did.
        //
        // `archivedAt` rather than `trashed` for exactly that reason: it is the marker both
        // states share, so a read by id answers what a listing answers, and a delete means
        // what a copy's undo needs it to mean — the item this run created is gone.
        Some(value) if !matches!(value.get("archivedAt"), None | Some(Value::Null)) => Ok(None),
        Some(value) => f(value).map(Some),
    }
}
fn connection<T>(
    d: &Value,
    k: &str,
    f: impl Fn(&Value) -> Result<T, SourceError>,
) -> Result<Page<T>, SourceError> {
    let c = d.get(k).ok_or_else(|| SourceError::Malformed {
        message: format!("missing {k} connection"),
    })?;
    let items = c
        .get("nodes")
        .and_then(Value::as_array)
        .ok_or_else(|| SourceError::Malformed {
            message: "missing nodes".into(),
        })?
        .iter()
        .map(f)
        .collect::<Result<_, _>>()?;
    let next = page_next(c)?;
    Ok(Page { items, next })
}
#[derive(Clone, Copy)]
enum DependencyRoot {
    Issue,
    Project,
}
impl DependencyRoot {
    const fn item_kind(self) -> ItemKind {
        match self {
            Self::Issue => ItemKind::Task,
            Self::Project => ItemKind::Project,
        }
    }
    const fn as_str(self) -> &'static str {
        match self {
            Self::Issue => "issue",
            Self::Project => "project",
        }
    }
}
fn relation_page(
    d: &Value,
    root: DependencyRoot,
    id: &NativeId,
    direction: Direction,
) -> Result<Page<DependencyEdge>, SourceError> {
    let key = if direction == Direction::DependsOn {
        "relations"
    } else {
        "inverseRelations"
    };
    let c = d
        .get(root.as_str())
        .and_then(|v| v.get(key))
        .ok_or_else(|| SourceError::Malformed {
            message: format!("missing {key}"),
        })?;
    let nodes = c
        .get("nodes")
        .and_then(Value::as_array)
        .ok_or_else(|| SourceError::Malformed {
            message: "missing relation nodes".into(),
        })?;
    let mut items = Vec::new();
    for n in nodes {
        let other = n
            .get(if direction == Direction::DependsOn {
                "relatedIssue"
            } else {
                "issue"
            })
            .or_else(|| {
                n.get(if direction == Direction::DependsOn {
                    "relatedProject"
                } else {
                    "project"
                })
            })
            .and_then(|v| v.get("id"))
            .and_then(Value::as_str)
            .ok_or_else(|| SourceError::Malformed {
                message: "missing related id".into(),
            })?;
        let (from, to) = if direction == Direction::DependsOn {
            (id.clone(), NativeId(other.into()))
        } else {
            (NativeId(other.into()), id.clone())
        };
        // llmlint: ignore-block[contracts_have_one_source_or_a_drift_gate] Linear publishes relation type as a string in the accepted 2026-08-24 schema; this boundary deliberately rejects every undocumented value, and real-HTTP tests prove both accepted values and rejection.
        let relation_type =
            n.get("type")
                .and_then(Value::as_str)
                .ok_or_else(|| SourceError::Malformed {
                    message: "missing relation type".into(),
                })?;
        // An issue relation and a project relation do not share a vocabulary. Linear
        // spells a project dependency `dependency`, where an issue's is `blocks`; the
        // write side sends exactly that pair and says why. So each root reads only its
        // own, and a value the other root would have accepted is refused here rather than
        // read as an edge this source could not have written.
        //
        // `related` is one of those values, and only an issue relation has it. Linear's
        // validator enumerates a project relation's `type` as `dependency` alone — see
        // the write side, which had `related` refused by the real API on 2026-09-04 — so
        // a project relation typed `related` is not a relation this workspace can hold.
        let kind = match (root, relation_type) {
            (DependencyRoot::Issue, "blocks") | (DependencyRoot::Project, "dependency") => {
                DependencyKind::Blocks
            }
            (DependencyRoot::Issue, "related") => DependencyKind::Related,
            _ => {
                return Err(SourceError::Malformed {
                    message: format!(
                        "invalid relation type: {relation_type} on a {} relation",
                        root.as_str()
                    ),
                });
            }
        };
        // llmlint: ignore-end[contracts_have_one_source_or_a_drift_gate]
        let item_kind = root.item_kind();
        items.push(DependencyEdge {
            from: DependencyEndpoint::from_native(from, item_kind),
            to: DependencyEndpoint::from_native(to, item_kind),
            kind,
        });
    }
    let next = page_next(c)?;
    Ok(Page { items, next })
}

fn optional_str<'a>(v: &'a Value, k: &str) -> Result<Option<&'a str>, SourceError> {
    match v.get(k) {
        None => Err(SourceError::Malformed {
            message: format!("missing field {k}"),
        }),
        Some(Value::Null) => Ok(None),
        Some(value) => value
            .as_str()
            .map(Some)
            .ok_or_else(|| SourceError::Malformed {
                message: format!("field {k} is not a string"),
            }),
    }
}

/// Linear has no caller-defined fields. The source owns an unobtrusive Markdown comment at the
/// end of the long-form field, and writes it on one line with the canonical JSON inside a code
/// span: ``<!-- onetaskgraph.metadata `{…}` -->``. The code span is the one spelling Linear keeps
/// byte for byte in a description or a document — see the module documentation's ruling on what
/// Linear does to an HTML comment — and [`slot_json`] escapes the three characters that could
/// end it early.
const METADATA_PREFIX: &str = "<!-- onetaskgraph.metadata";
/// The opening of the slot this source writes.
const METADATA_OPEN_SPAN: &str = "<!-- onetaskgraph.metadata `";
/// The close of the slot this source writes.
const METADATA_CLOSE_SPAN: &str = "` -->";
/// The opening of the multi-line slot this source wrote before the code span, still read: an
/// item written then keeps its metadata. Linear normalized its JSON, so a key or a value
/// Linear rewrote reads back as Linear left it — or, for an array Linear escaped, as a
/// malformed slot naming itself — and the next write of that item writes the code span.
const METADATA_OPEN: &str = "<!-- onetaskgraph.metadata\n";
const METADATA_CLOSE: &str = "\n-->";
/// The same close as Linear hands that multi-line slot back: it escapes a line opening
/// `-->`, so the slot reads back with a backslash before its close (observed from the real API
/// on 2026-09-14 for a document and on 2026-10-02 for an issue's description too).
const METADATA_CLOSE_ESCAPED: &str = "\n\\-->";

/// One JSON value in the encoding the slot holds it in: compact canonical JSON with `<`, `>`
/// and `` ` `` escaped as `\u003c`, `\u003e` and `\u0060`.
///
/// All three only ever occur inside a JSON string, where the escape means the same character,
/// so the value parses back exactly; and with them escaped no value can close the code span or
/// the HTML comment around it. A search phrase is not built with this: `slot_phrase` sends only
/// a value none of whose characters any encoder escapes, which this leaves as written.
fn slot_json(value: &impl serde::Serialize) -> Result<String, SourceError> {
    let encoded = serde_json::to_string(value).map_err(|error| SourceError::Malformed {
        message: error.to_string(),
    })?;
    Ok(encoded
        .replace('<', "\\u003c")
        .replace('>', "\\u003e")
        .replace('`', "\\u0060"))
}

/// The slot holding exactly `metadata`, as this source writes it.
fn slot_text(metadata: &std::collections::BTreeMap<String, Value>) -> Result<String, SourceError> {
    Ok(format!(
        "{METADATA_OPEN_SPAN}{}{METADATA_CLOSE_SPAN}",
        slot_json(metadata)?
    ))
}

/// Where the trailing metadata slot of `description` is: the byte its opening marker starts
/// at, and the span of the encoded JSON inside it — or `None` when it ends in no slot.
///
/// The one place the slot is recognised, in either spelling, so what [`metadata_description`]
/// reads out and what [`metadata_slot`] keeps for a content write are the same bytes.
fn slot_bounds(description: &str) -> Result<Option<(usize, usize, usize)>, SourceError> {
    let Some(start) = description.rfind(METADATA_PREFIX) else {
        return Ok(None);
    };
    let rest = &description[start..];
    let (encoded_start, close) = if rest.starts_with(METADATA_OPEN_SPAN) {
        let encoded_start = start + METADATA_OPEN_SPAN.len();
        (
            encoded_start,
            description[encoded_start..]
                .rfind(METADATA_CLOSE_SPAN)
                .map(|at| (at, METADATA_CLOSE_SPAN.len())),
        )
    } else if rest.starts_with(METADATA_OPEN) {
        let encoded_start = start + METADATA_OPEN.len();
        (
            encoded_start,
            [METADATA_CLOSE, METADATA_CLOSE_ESCAPED]
                .into_iter()
                .filter_map(|close| {
                    description[encoded_start..]
                        .find(close)
                        .map(|at| (at, close.len()))
                })
                .min(),
        )
    } else {
        return Ok(None);
    };
    let Some((relative_end, close_len)) = close else {
        return Err(SourceError::Malformed {
            message: "unterminated onetaskgraph metadata slot in Linear description".into(),
        });
    };
    let encoded_end = encoded_start + relative_end;
    if !description[encoded_end + close_len..].trim().is_empty() {
        return Ok(None);
    }
    Ok(Some((start, encoded_start, encoded_end)))
}

/// The metadata slot `description` ends in, exactly as it is stored, or `None`.
fn metadata_slot(description: &str) -> Result<Option<&str>, SourceError> {
    Ok(slot_bounds(description)?.map(|(start, _, _)| &description[start..]))
}

fn metadata_description(
    description: Option<String>,
) -> Result<(Option<String>, std::collections::BTreeMap<String, Value>), SourceError> {
    let Some(description) = description else {
        return Ok((None, Default::default()));
    };
    let Some((start, encoded_start, encoded_end)) = slot_bounds(&description)? else {
        return Ok((Some(description), Default::default()));
    };
    let metadata =
        serde_json::from_str(&description[encoded_start..encoded_end]).map_err(|error| {
            SourceError::Malformed {
                message: format!(
                    "invalid canonical JSON in Linear onetaskgraph metadata slot: {error}"
                ),
            }
        })?;
    // Exactly the text above the slot less the one blank line `long_form` sets it off by, so
    // content whose own end is whitespace reads back as itself. A description edited in Linear
    // down to a single line break before the slot loses just that one.
    let above = &description[..start];
    let visible = above
        .strip_suffix("\n\n")
        .or_else(|| above.strip_suffix('\n'))
        .unwrap_or(above);
    Ok(((!visible.is_empty()).then(|| visible.to_owned()), metadata))
}

/// The narrowing that asks Linear for the issues whose description holds `"<value>"` — a
/// string `value` as any JSON encoder writes it, quotes included — or `None` when `value` holds
/// a character an encoder may write another way, and only the confirmation decides.
///
/// A candidate set rather than the answer: the phrase can sit in the visible prose, or under
/// another key, and both are kept out by the confirmation over the parsed slot that follows
/// every read. What it cannot do is miss an issue whose slot holds the value — in the code span
/// this source writes, in the multi-line slot it wrote before, or in one a person spaced or
/// re-encoded by hand — which is what makes sending it sound. So it names the value alone, never
/// the key beside it, whose spacing a slot is free to vary; and only a value of printable ASCII
/// none of whose characters any encoder escapes — not `"`, `\`, `/`, `<`, `>`, `&`, `'` or a
/// backtick — because one that is escaped would be spelled in a stored slot otherwise than here.
fn slot_phrase(value: &str) -> Option<Value> {
    let verbatim = value.chars().all(|character| {
        (character.is_ascii_graphic() && !"\"\\/<>&'`".contains(character)) || character == ' '
    });
    verbatim.then(|| json!({"description": {"contains": format!("\"{value}\"")}}))
}

/// Whether `title`/`content` satisfies `query`, case-insensitively — the contract's own rule,
/// which the engine applies for a source that does not search, so the two cannot answer one
/// workspace differently.
fn text_holds(title: &str, content: Option<&str>, query: &TextQuery) -> bool {
    let terms = query.terms.to_lowercase();
    let in_title = title.to_lowercase().contains(&terms);
    let in_content = content.is_some_and(|body| body.to_lowercase().contains(&terms));
    match query.fields {
        TextFields::Title => in_title,
        TextFields::Content => in_content,
        TextFields::TitleOrContent => in_title || in_content,
    }
}

fn optional_string(v: &Value, k: &str) -> Result<Option<String>, SourceError> {
    Ok(optional_str(v, k)?.map(Into::into))
}
fn backend_id<'a>(value: &'a Value, field: &str) -> Result<&'a str, SourceError> {
    let id = str_at(value, field)?;
    (!id.is_empty())
        .then_some(id)
        .ok_or_else(|| SourceError::Malformed {
            message: format!("field {field} is an empty backend id"),
        })
}
fn mutation_payload(data: &Value, root: MutationRoot) -> Result<&Value, SourceError> {
    let root = root.as_str();
    let payload = data.get(root).ok_or_else(|| SourceError::Malformed {
        message: format!("missing {root}"),
    })?;
    match payload.get("success").and_then(Value::as_bool) {
        Some(true) => Ok(payload),
        Some(false) => Err(SourceError::Refused {
            message: format!("Linear reported {root} was unsuccessful"),
        }),
        None => Err(SourceError::Malformed {
            message: format!("missing boolean {root}.success"),
        }),
    }
}
fn page_next(c: &Value) -> Result<Option<Cursor>, SourceError> {
    let info = c.get("pageInfo").ok_or_else(|| SourceError::Malformed {
        message: "missing pageInfo".into(),
    })?;
    let more = info
        .get("hasNextPage")
        .and_then(Value::as_bool)
        .ok_or_else(|| SourceError::Malformed {
            message: "missing boolean pageInfo.hasNextPage".into(),
        })?;
    if !more {
        return Ok(None);
    }
    let cursor = str_at(info, "endCursor")?;
    Ok(Some(Cursor(cursor.into())))
}