From b8a25e884ac7b65f50ed931125cff3517ff8fe4d Mon Sep 17 00:00:00 2001 From: seal Date: Tue, 18 Aug 2026 22:47:05 -0400 Subject: [PATCH 1/2] feat(forge): Linear provider (read + write, issues half) (RIG-2170) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add go/internal/forge/linear.go: a hand-rolled net/http GraphQL client for Linear, a co-equal forge write target beside GitHub, per the frozen write-path design §5. Stdlib-only (no go-github/GraphQL library), mirroring github.go's posture and sharing its seams: a TokenSource for LINEAR_FORGE_TOKEN (DL-052), a mu-guarded fail-fast rate gate, and an injectable clock. Linear is issues-only (DL-051): CreateIssue/CommentOnIssue/GetIssue/ListIssues implement the issues half; the PR/review family (CreatePullRequest, CommentOnPullRequest, SubmitReview, GetPullRequest, Checks) returns ErrUnsupported. repo is the Linear team key, resolved to a team UUID once and cached. Attribution (design §5, OQ-5/OQ-8): writes set Linear's createAsUser/displayIconUrl to one constant shared Compass app identity, gated by a one-time actor-capability probe (viewer.app); a non-actor token degrades to stamp-only and emits the named log line. The fine-grained per-agent owner truth rides the Service's StampOwner header unchanged. Rate limits map a 429 or a GraphQL RATELIMITED code to ErrBudgetExhausted with the reset instant from Retry-After / X-RateLimit-Requests-Reset; a GraphQL AUTHENTICATION_ERROR or HTTP 401 invalidates the token. 17 hermetic tests over a stubbed RoundTripper cover request goldens, team-cache, read mapping/filter/pagination, ErrUnsupported, the rate/auth/degrade paths. Refs RIG-2209. Co-authored-by: Matt Wilkinson --- go/internal/forge/linear.go | 775 +++++++++++++++++++++++++++++++ go/internal/forge/linear_test.go | 526 +++++++++++++++++++++ 2 files changed, 1301 insertions(+) create mode 100644 go/internal/forge/linear.go create mode 100644 go/internal/forge/linear_test.go diff --git a/go/internal/forge/linear.go b/go/internal/forge/linear.go new file mode 100644 index 00000000..593b832f --- /dev/null +++ b/go/internal/forge/linear.go @@ -0,0 +1,775 @@ +package forge + +// Linear is a hand-rolled net/http GraphQL client for the Linear issue tracker +// (design.md §5), a co-equal forge write target beside GitHub. It mirrors +// github.go's no-dependency posture (stdlib only, no go-github / GraphQL +// library) and shares its seams: a TokenSource for the credential, a +// mu-guarded fail-fast rate gate (a write burst respects the same reserve as +// the poll driver so it cannot starve it), and an injectable clock. +// +// Linear is ISSUES-ONLY (DL-051): the PR/review half of Provider returns +// ErrUnsupported. `repo` is the Linear TEAM KEY (e.g. "SEA"), not owner/name; +// the client resolves key -> team id once and caches it (mu-guarded). +// +// Attribution (design.md §5, OQ-5/OQ-8): writes set Linear's createAsUser + +// displayIconUrl to ONE constant shared Compass app identity, so native Linear +// display shows a single "via Application" identity for every agent while the +// fine-grained per-agent owner truth rides the Service's StampOwner header. +// Both channels are Server-chosen (DL-050 unforgeability). Whether the client +// may set createAsUser at all is governed by a one-time actor-capability probe +// (A4, a stated design INTENT, not an asserted API behavior): a token that is +// not an OAuth actor=app token degrades to stamp-only. +// +// Body handling matches the Provider contract: a Create/Comment body is +// PRE-stamped by the Service and sent verbatim; a read returns the body RAW +// (the Service strips/parses on read). + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http" + "slices" + "strconv" + "strings" + "sync" + "time" +) + +const ( + // linearDefaultEndpoint is the public Linear GraphQL endpoint; LinearConfig.Host + // overrides it (the whole endpoint URL, not just a hostname). + linearDefaultEndpoint = "https://api.linear.app/graphql" + + // linearBodyLimit is the max issue/comment body size (BYTES) the Service + // enforces before a Linear write. Linear does not publish a single pinned + // GraphQL body cap, so this is a CONSERVATIVE constant: 65536 bytes matches + // GitHub.BodyLimit, keeping the Service's cross-provider ceiling uniform and + // comfortably under any Linear description limit observed in practice. See + // the T6 summary — value is a choice, not a documented Linear cap. + linearBodyLimit = 65536 + + // attributionUser and attributionIconURL are the ONE shared Compass app + // identity every Linear write is attributed to via createAsUser / + // displayIconUrl (design.md §5). They are deliberately coarse (not + // per-agent); the per-agent owner truth lives in the StampOwner header. + attributionUser = "Compass" + attributionIconURL = "https://compass.sealedsecurity.com/assets/compass-app.png" + + // forge state truths mapped from Linear workflow-state types. + stateOpen = "open" + stateClosed = "closed" + + // GraphQL variable keys reused across queries. + varKey = "key" + varTeam = "team" + varFilter = "filter" +) + +// linearClosedStateTypes are the Linear workflow-state `type` values that map +// to the forge's "closed" truth. Every other type maps to "open". Verified +// against Linear SDL WorkflowState.type: "triage", "backlog", "unstarted", +// "started", "completed", "canceled", "duplicate". +var linearClosedStateTypes = []string{"completed", "canceled"} + +// LinearConfig configures a Linear client. +type LinearConfig struct { + Host string // GraphQL endpoint URL; "" -> linearDefaultEndpoint + Token TokenSource // required (its own LINEAR_FORGE_TOKEN, DL-052) + Client *http.Client // nil -> a default client with a sane timeout + Log *slog.Logger // nil -> slog.Default(); carries the degrade log line +} + +// Linear is a stdlib GraphQL client for a Linear forge. It is stateless about +// cursors (the caller owns durable poll state); the only in-memory state is the +// mu-guarded rate gate, team-id cache, and one-time actor-probe result. +type Linear struct { + host string + token TokenSource + client *http.Client + log *slog.Logger + + // mu guards resetAt, teamIDs, and the actor-probe fields. The client may be + // shared between the poll driver and write-RPC goroutines (OQ-6), so all + // three are concurrent read-modify-write; mu is held only around the fast + // state touches, never across an HTTP round-trip. + mu sync.Mutex + + // resetAt is the rate-budget gate (see GitHub.resetAt). Non-zero and before + // now() -> the next call fails fast with ErrBudgetExhausted. Zero -> open. + resetAt time.Time + + // teamIDs caches Linear team key -> team UUID; a key is resolved once via a + // teams query and reused for every subsequent CreateIssue. + teamIDs map[string]string + + // probeDone/actorCapable cache the one-time actor-capability probe (A4). + // Once probeDone, actorCapable governs whether writes set createAsUser. + probeDone bool + actorCapable bool + + // now is the clock seam (defaults to time.Now); tests override it to drive + // the reset-time gate deterministically. + now func() time.Time +} + +// NewLinear returns a Linear client. A nil cfg.Client gets a default client +// with a sane timeout; a nil cfg.Log falls back to slog.Default(). cfg.Token is +// required (the caller wires it — DL-052). +func NewLinear(cfg LinearConfig) *Linear { + client := cfg.Client + if client == nil { + client = &http.Client{Timeout: 30 * time.Second} + } + log := cfg.Log + if log == nil { + log = slog.Default() + } + return &Linear{ + host: cfg.Host, + token: cfg.Token, + client: client, + log: log, + teamIDs: make(map[string]string), + now: time.Now, + } +} + +// Compile-time proof that Linear satisfies the Provider interface. +var _ Provider = (*Linear)(nil) + +// --- Provider: exported methods ---------------------------------------------- + +// Name identifies this provider. +func (l *Linear) Name() string { return "linear" } + +// CreateIssue creates an issue on the team keyed by repo. in.Body is PRE-stamped +// by the Service; it becomes the Linear description verbatim. createAsUser / +// displayIconUrl are set to the shared Compass identity when the actor probe +// passes. Labels are NOT sent: Linear's IssueCreateInput.labelIds takes UUIDs, +// not names, and name->UUID resolution is out of this slice (see T6 summary). +func (l *Linear) CreateIssue(ctx context.Context, repo string, in CreateIssue) (Issue, error) { + teamID, err := l.resolveTeamID(ctx, repo) + if err != nil { + return Issue{}, fmt.Errorf("forge: linear create issue %q: %w", repo, err) + } + input := map[string]any{"teamId": teamID, "title": in.Title, "description": in.Body} + l.applyAttribution(ctx, input) + + const query = `mutation CompassIssueCreate($input: IssueCreateInput!) { + issueCreate(input: $input) { + issue { ...CompassIssueFields } + } +}` + issueFieldsFragment + var out struct { + IssueCreate struct { + Issue linearIssue `json:"issue"` + } `json:"issueCreate"` + } + if err := l.doGraphQL(ctx, query, map[string]any{"input": input}, &out); err != nil { + return Issue{}, fmt.Errorf("forge: linear create issue %q: %w", repo, err) + } + return out.IssueCreate.Issue.toIssue(), nil +} + +// CommentOnIssue posts a comment on issue number in the team keyed by repo. body +// is PRE-stamped. It resolves the issue UUID from (team, number), then runs +// commentCreate with createAsUser gated on the actor probe. +func (l *Linear) CommentOnIssue(ctx context.Context, repo string, number uint64, body string) (Comment, error) { + issueID, err := l.resolveIssueID(ctx, repo, number) + if err != nil { + return Comment{}, fmt.Errorf("forge: linear comment on issue %q#%d: %w", repo, number, err) + } + input := map[string]any{"issueId": issueID, "body": body} + l.applyAttribution(ctx, input) + + const query = `mutation CompassCommentCreate($input: CommentCreateInput!) { + commentCreate(input: $input) { + comment { id url body user { displayName } } + } +}` + var out struct { + CommentCreate struct { + Comment linearComment `json:"comment"` + } `json:"commentCreate"` + } + if err := l.doGraphQL(ctx, query, map[string]any{"input": input}, &out); err != nil { + return Comment{}, fmt.Errorf("forge: linear comment on issue %q#%d: %w", repo, number, err) + } + return out.CommentCreate.Comment.toComment(), nil +} + +// GetIssue fetches one issue by its per-team number in the team keyed by repo. +// Body is returned RAW. Resolved via the issues query (team key + number), +// which accepts the human coordinate without a prior UUID lookup. +func (l *Linear) GetIssue(ctx context.Context, repo string, number uint64) (Issue, error) { + const query = `query CompassIssueGet($filter: IssueFilter!) { + issues(filter: $filter, first: 1) { + nodes { ...CompassIssueFields } + } +}` + issueFieldsFragment + filter := map[string]any{ + varTeam: map[string]any{varKey: map[string]any{"eq": repo}}, + "number": map[string]any{"eq": float64(number)}, + } + var out struct { + Issues struct { + Nodes []linearIssue `json:"nodes"` + } `json:"issues"` + } + if err := l.doGraphQL(ctx, query, map[string]any{varFilter: filter}, &out); err != nil { + return Issue{}, fmt.Errorf("forge: linear get issue %q#%d: %w", repo, number, err) + } + if len(out.Issues.Nodes) == 0 { + return Issue{}, &StatusError{Status: http.StatusNotFound, Message: fmt.Sprintf("no issue %s-%d", repo, number)} + } + return out.Issues.Nodes[0].toIssue(), nil +} + +// ListIssues walks every issue in the team keyed by repo, narrowed by f, across +// all pages (Linear paginates at 50; the loop follows pageInfo). Bodies are RAW. +func (l *Linear) ListIssues(ctx context.Context, repo string, f IssueFilter) ([]Issue, error) { + const query = `query CompassIssueList($filter: IssueFilter!, $after: String) { + issues(filter: $filter, first: 50, after: $after) { + nodes { ...CompassIssueFields } + pageInfo { hasNextPage endCursor } + } +}` + issueFieldsFragment + filter := teamIssueFilter(repo, f) + + var all []Issue + var after string + for { + vars := map[string]any{varFilter: filter} + if after != "" { + vars["after"] = after + } + var out struct { + Issues struct { + Nodes []linearIssue `json:"nodes"` + PageInfo struct { + HasNextPage bool `json:"hasNextPage"` + EndCursor string `json:"endCursor"` + } `json:"pageInfo"` + } `json:"issues"` + } + if err := l.doGraphQL(ctx, query, vars, &out); err != nil { + return nil, fmt.Errorf("forge: linear list issues %q: %w", repo, err) + } + for _, n := range out.Issues.Nodes { + all = append(all, n.toIssue()) + } + if !out.Issues.PageInfo.HasNextPage { + break + } + after = out.Issues.PageInfo.EndCursor + } + return all, nil +} + +// CreatePullRequest is unsupported: Linear has no pull-request concept +// (design.md §5). The Service maps ErrUnsupported to the in-band `unimplemented`. +func (l *Linear) CreatePullRequest(ctx context.Context, repo string, in CreatePR) (PullRequest, error) { + return PullRequest{}, ErrUnsupported +} + +// CommentOnPullRequest is unsupported (Linear has no PRs). +func (l *Linear) CommentOnPullRequest(ctx context.Context, repo string, number uint64, body string) (Comment, error) { + return Comment{}, ErrUnsupported +} + +// SubmitReview is unsupported (Linear has no review concept). +func (l *Linear) SubmitReview(ctx context.Context, repo string, number uint64, in SubmitReview) (SubmittedReview, error) { + return SubmittedReview{}, ErrUnsupported +} + +// GetPullRequest is unsupported (Linear has no PRs); the canonical PullRequest +// surface is never fabricated on a Linear coordinate. +func (l *Linear) GetPullRequest(ctx context.Context, repo string, number uint64) (PullRequest, error) { + return PullRequest{}, ErrUnsupported +} + +// Checks is unsupported (Linear has no PR head checks). +func (l *Linear) Checks(ctx context.Context, repo string, number uint64) (Checks, error) { + return Checks{}, ErrUnsupported +} + +// BodyLimit is the max body size (BYTES) the Service enforces before a write. +// See linearBodyLimit for the conservative-constant rationale. +func (l *Linear) BodyLimit() int { return linearBodyLimit } + +// --- Provider: unexported plumbing ------------------------------------------- + +// doGraphQL carries the write- and read-path plumbing once: the resetAt +// fail-fast gate, token auth, the JSON POST of {query, variables}, and response +// classification. It decodes the `data` object into out on success. +func (l *Linear) doGraphQL(ctx context.Context, query string, variables map[string]any, out any) error { + if l.gateBlocked() { + return fmt.Errorf("linear graphql: %w", ErrBudgetExhausted) + } + + token, err := l.token.Token(ctx) + if err != nil { + return fmt.Errorf("resolve token: %w", err) + } + + payload, err := json.Marshal(map[string]any{"query": query, "variables": variables}) + if err != nil { + return fmt.Errorf("marshal request body: %w", err) + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, l.endpoint(), bytes.NewReader(payload)) + if err != nil { + return fmt.Errorf("build request: %w", err) + } + // Linear OAuth tokens (the actor=app token this provider uses, DL-052) are + // bearer tokens. See the T6 summary: the "Bearer " scheme is the grounded + // choice for an OAuth token; a raw personal API key would omit it. + req.Header.Set("Authorization", "Bearer "+token) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json") + + resp, err := l.client.Do(req) + if err != nil { + return fmt.Errorf("do request: %w", err) + } + defer func() { _ = resp.Body.Close() }() // body fully read below; a close error on a drained read body is not actionable + + body, err := io.ReadAll(resp.Body) + if err != nil { + return fmt.Errorf("read body: %w", err) + } + return l.handleResponse(resp, body, out) +} + +// handleResponse classifies a Linear response and, on success, decodes data +// into out. Precedence: rate limit (HTTP 429 or a RATELIMITED code) -> auth +// (HTTP 401 or AUTHENTICATION_ERROR code, which invalidates the token) -> any +// other GraphQL errors (even on HTTP 200) -> a non-2xx with no usable envelope +// -> success. +func (l *Linear) handleResponse(resp *http.Response, body []byte, out any) error { + var gr graphQLResponse + decodeErr := json.Unmarshal(body, &gr) + status := resp.StatusCode + + // Rate limit: a 429, or Linear's GraphQL-level complexity/rate rejection + // (HTTP 400 carrying extensions.code == "RATELIMITED"). Arms the gate; no + // token re-resolve. + if status == http.StatusTooManyRequests || (decodeErr == nil && hasErrorCode(gr.Errors, "RATELIMITED")) { + l.mu.Lock() + l.armGate(l.rateLimitReset(resp)) + l.mu.Unlock() + return fmt.Errorf("linear graphql http %d: %w", status, ErrBudgetExhausted) + } + + // Auth failure: a 401, or a GraphQL AUTHENTICATION_ERROR. Drop the cached + // token so the next batch re-resolves. + if status == http.StatusUnauthorized || (decodeErr == nil && hasErrorCode(gr.Errors, "AUTHENTICATION_ERROR")) { + l.token.Invalidate() + return &StatusError{Status: statusOr(status, http.StatusUnauthorized), Message: joinErrors(gr.Errors)} + } + + // Any other GraphQL top-level errors — Linear returns these on HTTP 200. + if decodeErr == nil && len(gr.Errors) > 0 { + return &StatusError{Status: statusOr(status, http.StatusOK), Message: joinErrors(gr.Errors)} + } + + // A non-2xx with no parseable GraphQL error envelope. + if status < 200 || status >= 300 { + return &StatusError{Status: status, Message: strings.TrimSpace(string(body))} + } + + // Success. A 2xx body that would not parse as the envelope is a decode failure. + if decodeErr != nil { + return fmt.Errorf("decode response: %w", decodeErr) + } + l.recordBudget(resp) + if out != nil { + if err := json.Unmarshal(gr.Data, out); err != nil { + return fmt.Errorf("decode data: %w", err) + } + } + return nil +} + +// endpoint is the configured GraphQL endpoint, defaulting to Linear's public one. +func (l *Linear) endpoint() string { + if l.host == "" { + return linearDefaultEndpoint + } + return l.host +} + +// gateBlocked reports whether the fail-fast budget gate is armed, clearing a +// gate whose reset instant has passed as a side effect. Guarded by mu. +func (l *Linear) gateBlocked() bool { + l.mu.Lock() + defer l.mu.Unlock() + if l.resetAt.IsZero() { + return false + } + if l.now().Before(l.resetAt) { + return true + } + l.resetAt = time.Time{} + return false +} + +// armGate sets the reset-time gate; a zero at falls back to a bounded skip so +// the gate self-clears rather than wedging. Caller MUST hold l.mu. +func (l *Linear) armGate(at time.Time) { + if at.IsZero() { + at = l.now().Add(defaultSkip) + } + l.resetAt = at +} + +// recordBudget updates the fail-fast gate from Linear's X-RateLimit-Requests-* +// headers on a successful response. remaining <= reserve arms the gate until +// the reset instant; absent/malformed headers leave it open (never wedge). +func (l *Linear) recordBudget(resp *http.Response) { + l.mu.Lock() + defer l.mu.Unlock() + raw := resp.Header.Get("X-Ratelimit-Requests-Remaining") + if raw == "" { + l.resetAt = time.Time{} + return + } + remaining, err := strconv.Atoi(raw) + if err != nil { + l.resetAt = time.Time{} + return + } + if remaining > reserve { + l.resetAt = time.Time{} + return + } + l.armGate(linearResetFromHeader(resp.Header.Get("X-Ratelimit-Requests-Reset"))) +} + +// rateLimitReset derives a reset instant from a rate-limited response: a +// Retry-After (delta-seconds or HTTP-date) wins, else X-RateLimit-Requests-Reset +// (UTC epoch MILLISECONDS per Linear's docs). A zero time lets armGate fall back +// to the bounded default skip. +func (l *Linear) rateLimitReset(resp *http.Response) time.Time { + if ra := resp.Header.Get("Retry-After"); ra != "" { + if secs, err := strconv.Atoi(ra); err == nil { + return l.now().Add(time.Duration(secs) * time.Second) + } + if at, err := http.ParseTime(ra); err == nil { + return at + } + } + return linearResetFromHeader(resp.Header.Get("X-Ratelimit-Requests-Reset")) +} + +// resolveTeamID maps a Linear team key to its UUID, caching the result. The +// cache is checked and stored under mu without holding it across the network +// call; a concurrent miss issues a redundant (idempotent) lookup at worst. +func (l *Linear) resolveTeamID(ctx context.Context, key string) (string, error) { + l.mu.Lock() + if id, ok := l.teamIDs[key]; ok { + l.mu.Unlock() + return id, nil + } + l.mu.Unlock() + + const query = `query CompassTeamByKey($key: String!) { + teams(filter: {key: {eq: $key}}, first: 1) { + nodes { id } + } +}` + var out struct { + Teams struct { + Nodes []struct { + ID string `json:"id"` + } `json:"nodes"` + } `json:"teams"` + } + if err := l.doGraphQL(ctx, query, map[string]any{varKey: key}, &out); err != nil { + return "", err + } + if len(out.Teams.Nodes) == 0 { + return "", &StatusError{Status: http.StatusNotFound, Message: fmt.Sprintf("no team with key %q", key)} + } + id := out.Teams.Nodes[0].ID + + l.mu.Lock() + l.teamIDs[key] = id + l.mu.Unlock() + return id, nil +} + +// resolveIssueID maps a (team key, per-team number) pair to a Linear issue UUID +// via the issues query, for CommentOnIssue's issueId. Not cached — an issue +// number is written to at most a handful of times per session. +func (l *Linear) resolveIssueID(ctx context.Context, repo string, number uint64) (string, error) { + const query = `query CompassIssueIDByNumber($filter: IssueFilter!) { + issues(filter: $filter, first: 1) { + nodes { id } + } +}` + filter := map[string]any{ + varTeam: map[string]any{varKey: map[string]any{"eq": repo}}, + "number": map[string]any{"eq": float64(number)}, + } + var out struct { + Issues struct { + Nodes []struct { + ID string `json:"id"` + } `json:"nodes"` + } `json:"issues"` + } + if err := l.doGraphQL(ctx, query, map[string]any{varFilter: filter}, &out); err != nil { + return "", err + } + if len(out.Issues.Nodes) == 0 { + return "", &StatusError{Status: http.StatusNotFound, Message: fmt.Sprintf("no issue %s-%d", repo, number)} + } + return out.Issues.Nodes[0].ID, nil +} + +// actorAttribution reports whether writes may set createAsUser, running the +// one-time capability probe on first call and caching the result. The probe +// queries `viewer { app }`: an actor=app OAuth token authenticates AS the app, +// so viewer.app is true; a plain user/API-key token reports false. A probe that +// errors is treated as not-capable (degrade, never block the write). On the +// not-capable transition it emits the named degrade log line EXACTLY once. +func (l *Linear) actorAttribution(ctx context.Context) bool { + l.mu.Lock() + if l.probeDone { + capable := l.actorCapable + l.mu.Unlock() + return capable + } + l.mu.Unlock() + + const query = `query CompassActorProbe { + viewer { app } +}` + var out struct { + Viewer struct { + App bool `json:"app"` + } `json:"viewer"` + } + probeErr := l.doGraphQL(ctx, query, nil, &out) + capable := probeErr == nil && out.Viewer.App + + l.mu.Lock() + defer l.mu.Unlock() + if l.probeDone { + // A concurrent caller finished the probe first; honor its result. + return l.actorCapable + } + l.probeDone = true + l.actorCapable = capable + if !capable { + if probeErr != nil { + l.log.Warn("linear: actor attribution unavailable; degrading to stamp-only", "probe_error", probeErr) + } else { + l.log.Warn("linear: actor attribution unavailable; degrading to stamp-only") + } + } + return capable +} + +// applyAttribution sets createAsUser/displayIconUrl on a mutation input when the +// actor probe reports the token is capable; otherwise it is a no-op (stamp-only +// degradation). Both values are the constant shared Compass app identity. +func (l *Linear) applyAttribution(ctx context.Context, input map[string]any) { + if l.actorAttribution(ctx) { + input["createAsUser"] = attributionUser + input["displayIconUrl"] = attributionIconURL + } +} + +// --- GraphQL wire envelope + helpers ----------------------------------------- + +// graphQLResponse is the standard GraphQL response envelope. Linear returns +// HTTP 200 with a top-level `errors` array for most failures (and HTTP 400 for +// GraphQL-level rate limits, still carrying the errors array), so both fields +// are always decoded. +type graphQLResponse struct { + Data json.RawMessage `json:"data"` + Errors []graphQLError `json:"errors"` +} + +// graphQLError is one entry of the GraphQL `errors` array. The extensions.code +// discriminates a rate-limit ("RATELIMITED") or auth ("AUTHENTICATION_ERROR") +// failure from an ordinary one. +type graphQLError struct { + Message string `json:"message"` + Extensions struct { + Code string `json:"code"` + } `json:"extensions"` +} + +func hasErrorCode(errs []graphQLError, code string) bool { + for _, e := range errs { + if e.Extensions.Code == code { + return true + } + } + return false +} + +func joinErrors(errs []graphQLError) string { + msgs := make([]string, 0, len(errs)) + for _, e := range errs { + msgs = append(msgs, e.Message) + } + return strings.Join(msgs, "; ") +} + +// statusOr returns status when it is a real HTTP error status (>=400), else the +// fallback. A GraphQL-level auth/error on an HTTP 200 thus surfaces a meaningful +// StatusError status (401 for auth, 200 for a plain query error) to the Service. +func statusOr(status, fallback int) int { + if status >= 400 { + return status + } + return fallback +} + +// linearResetFromHeader parses an X-RateLimit-Requests-Reset value (UTC epoch +// MILLISECONDS) into an absolute instant; absent/malformed yields the zero time. +func linearResetFromHeader(raw string) time.Time { + if raw == "" { + return time.Time{} + } + ms, err := strconv.ParseInt(raw, 10, 64) + if err != nil { + return time.Time{} + } + return time.UnixMilli(ms) +} + +// --- issue/comment wire types + mapping -------------------------------------- + +// issueFieldsFragment is the shared selection set for reading a Linear issue +// into forge.Issue. Body is Description (returned RAW; the Service strips). +const issueFieldsFragment = ` +fragment CompassIssueFields on Issue { + number + title + description + url + state { name type } + labels { nodes { name } } + creator { displayName } + updatedAt +}` + +// linearIssue is the wire shape of a Linear issue (only the forge.Issue fields +// are decoded). number is a GraphQL Float; creator is null for app/bot-created +// issues. +type linearIssue struct { + Number float64 `json:"number"` + Title string `json:"title"` + Description string `json:"description"` + URL string `json:"url"` + State struct { + Name string `json:"name"` + Type string `json:"type"` + } `json:"state"` + Labels struct { + Nodes []struct { + Name string `json:"name"` + } `json:"nodes"` + } `json:"labels"` + Creator *struct { + DisplayName string `json:"displayName"` + } `json:"creator"` + UpdatedAt string `json:"updatedAt"` +} + +// toIssue maps a decoded Linear issue to forge.Issue. Body is RAW; State is +// mapped to the forge's open/closed truth; UpdatedAt parses RFC-3339 (an +// unparseable value leaves the zero time). +func (r linearIssue) toIssue() Issue { + labels := make([]string, 0, len(r.Labels.Nodes)) + for _, n := range r.Labels.Nodes { + labels = append(labels, n.Name) + } + var updated time.Time + if r.UpdatedAt != "" { + if t, err := time.Parse(time.RFC3339, r.UpdatedAt); err == nil { + updated = t + } + } + account := "" + if r.Creator != nil { + account = r.Creator.DisplayName + } + return Issue{ + Number: uint64(r.Number), + Title: r.Title, + Body: r.Description, + State: mapLinearState(r.State.Type), + URL: r.URL, + ForgeAccount: account, + Labels: labels, + UpdatedAt: updated, + } +} + +// mapLinearState maps a Linear workflow-state type to the forge's open/closed +// truth (see linearClosedStateTypes). +func mapLinearState(stateType string) string { + if slices.Contains(linearClosedStateTypes, stateType) { + return stateClosed + } + return stateOpen +} + +// linearComment is the wire shape of a Linear comment. Its id is a UUID, which +// forge.Comment.ID (uint64) cannot carry — see toComment. +type linearComment struct { + ID string `json:"id"` + URL string `json:"url"` + Body string `json:"body"` + User *struct { + DisplayName string `json:"displayName"` + } `json:"user"` +} + +// toComment maps a decoded Linear comment to forge.Comment. Linear comment IDs +// are UUIDs, so ID stays zero and identity travels via URL (see the T6 summary +// flag: forge.Comment.ID is uint64 and cannot hold a Linear UUID). +func (r linearComment) toComment() Comment { + account := "" + if r.User != nil { + account = r.User.DisplayName + } + return Comment{URL: r.URL, Body: r.Body, ForgeAccount: account} +} + +// teamIssueFilter builds the Linear IssueFilter for a team's issues, narrowed by +// the forge IssueFilter. State maps to the workflow-state type ("open" -> +// type nin closed, "closed" -> type in closed; "all"/"" -> unfiltered). Labels +// require ALL given names (GitHub's AND semantics): an `and` of per-label +// `some` sub-filters. +func teamIssueFilter(key string, f IssueFilter) map[string]any { + filter := map[string]any{ + varTeam: map[string]any{varKey: map[string]any{"eq": key}}, + } + switch f.State { + case stateOpen: + filter["state"] = map[string]any{"type": map[string]any{"nin": linearClosedStateTypes}} + case stateClosed: + filter["state"] = map[string]any{"type": map[string]any{"in": linearClosedStateTypes}} + } + if len(f.Labels) > 0 { + ands := make([]any, 0, len(f.Labels)) + for _, name := range f.Labels { + ands = append(ands, map[string]any{ + "labels": map[string]any{"some": map[string]any{"name": map[string]any{"eq": name}}}, + }) + } + filter["and"] = ands + } + return filter +} diff --git a/go/internal/forge/linear_test.go b/go/internal/forge/linear_test.go new file mode 100644 index 00000000..ed22ddfd --- /dev/null +++ b/go/internal/forge/linear_test.go @@ -0,0 +1,526 @@ +package forge + +// Unit tests for the hand-rolled net/http Linear GraphQL client, driven by a +// stubbed http.RoundTripper (no network). Covers the T6 test cycle: issueCreate +// and commentCreate request goldens (query + variables, incl. teamId and +// createAsUser when the actor probe passes), team-key->id resolve-once-and-cache, +// GetIssue / ListIssues read mapping incl. IssueFilter state, ErrUnsupported for +// all five PR/review ops, the 429 -> resource_exhausted mapping, a GraphQL +// errors-on-200 -> *StatusError, the actor-probe-fails degrade path (no +// createAsUser + the exact log line), and 401 -> TokenSource.Invalidate. +// context.Background() here is the test root — the sanctioned F-ttsr exemption. + +import ( + "context" + "encoding/json" + "errors" + "log/slog" + "net/http" + "slices" + "strconv" + "strings" + "sync" + "testing" + "time" +) + +// --- stubs ------------------------------------------------------------------- + +// newTestLinear wires a Linear client over a scripted transport, a fake token +// source, and a captured logger (so the degrade line is assertable). +func newTestLinear(rt *scriptedRoundTripper, ts *fakeTokenSource, log *slog.Logger) *Linear { + return NewLinear(LinearConfig{ + Token: ts, + Client: &http.Client{Transport: rt}, + Log: log, + }) +} + +// capturingHandler records every log record for message assertions. +type capturingHandler struct { + mu sync.Mutex + msgs []string +} + +func (h *capturingHandler) Enabled(context.Context, slog.Level) bool { return true } +func (h *capturingHandler) Handle(_ context.Context, r slog.Record) error { + h.mu.Lock() + defer h.mu.Unlock() + h.msgs = append(h.msgs, r.Message) + return nil +} +func (h *capturingHandler) WithAttrs([]slog.Attr) slog.Handler { return h } +func (h *capturingHandler) WithGroup(string) slog.Handler { return h } +func (h *capturingHandler) has(msg string) bool { + h.mu.Lock() + defer h.mu.Unlock() + return slices.Contains(h.msgs, msg) +} +func (h *capturingHandler) count(msg string) int { + h.mu.Lock() + defer h.mu.Unlock() + n := 0 + for _, m := range h.msgs { + if m == msg { + n++ + } + } + return n +} + +// probeResp is a scripted `viewer { app }` response with the given capability. +func probeResp(app bool) scriptedResponse { + return scriptedResponse{status: 200, body: `{"data":{"viewer":{"app":` + strconv.FormatBool(app) + `}}}`} +} + +// teamResp is a scripted team-key->id lookup response. +func teamResp(id string) scriptedResponse { + return scriptedResponse{status: 200, body: `{"data":{"teams":{"nodes":[{"id":"` + id + `"}]}}}`} +} + +const degradeMsg = "linear: actor attribution unavailable; degrading to stamp-only" + +// decodeGraphQLReq parses a recorded request body into its query + variables. +func decodeGraphQLReq(t *testing.T, body string) (string, map[string]any) { + t.Helper() + var req struct { + Query string `json:"query"` + Variables map[string]any `json:"variables"` + } + if err := json.Unmarshal([]byte(body), &req); err != nil { + t.Fatalf("decode graphql request: %v (body=%q)", err, body) + } + return req.Query, req.Variables +} + +// --- item 1: issueCreate golden (probe passes -> createAsUser present) ------- + +func TestLinearCreateIssueRequestGolden(t *testing.T) { + rt := &scriptedRoundTripper{responses: []scriptedResponse{ + teamResp("team-uuid-1"), + probeResp(true), // actor probe: capable + {status: 200, body: `{"data":{"issueCreate":{"issue":{ + "number":42,"title":"a bug","description":"stamped body", + "url":"https://linear.app/x/issue/SEA-42","state":{"name":"Todo","type":"unstarted"}, + "labels":{"nodes":[]},"creator":null,"updatedAt":"2026-08-01T12:30:00Z"}}}}`}, + }} + l := newTestLinear(rt, &fakeTokenSource{token: "sekret"}, slog.New(&capturingHandler{})) + + got, err := l.CreateIssue(context.Background(), "SEA", + CreateIssue{Title: "a bug", Body: "stamped body"}) + if err != nil { + t.Fatalf("CreateIssue: %v", err) + } + + // The create request is the 3rd (probe, team lookup, then create). + createReq := rt.requests[2] + if h := createReq.Header.Get("Authorization"); h != "Bearer sekret" { + t.Errorf("Authorization = %q, want Bearer sekret", h) + } + query, vars := decodeGraphQLReq(t, readReqBody(t, createReq)) + if !strings.Contains(query, "issueCreate(input: $input)") { + t.Errorf("query missing issueCreate mutation: %q", query) + } + input, ok := vars["input"].(map[string]any) + if !ok { + t.Fatalf("input variable missing/not an object: %#v", vars["input"]) + } + if input["teamId"] != "team-uuid-1" { + t.Errorf("teamId = %v, want team-uuid-1", input["teamId"]) + } + if input["title"] != "a bug" || input["description"] != "stamped body" { + t.Errorf("title/description wrong: %#v", input) + } + if input["createAsUser"] != attributionUser { + t.Errorf("createAsUser = %v, want %q (probe passed)", input["createAsUser"], attributionUser) + } + if input["displayIconUrl"] != attributionIconURL { + t.Errorf("displayIconUrl = %v, want %q", input["displayIconUrl"], attributionIconURL) + } + if got.Number != 42 || got.Title != "a bug" || got.Body != "stamped body" || + got.State != "open" || got.URL != "https://linear.app/x/issue/SEA-42" { + t.Errorf("decoded Issue = %+v", got) + } +} + +// --- item 2: commentCreate golden -------------------------------------------- + +func TestLinearCommentOnIssueRequestGolden(t *testing.T) { + rt := &scriptedRoundTripper{responses: []scriptedResponse{ + {status: 200, body: `{"data":{"issues":{"nodes":[{"id":"issue-uuid-9"}]}}}`}, // resolve issue id + probeResp(true), // actor probe + {status: 200, body: `{"data":{"commentCreate":{"comment":{ + "id":"comment-uuid","url":"https://linear.app/x/issue/SEA-7#comment-1", + "body":"a reply","user":null}}}}`}, + }} + l := newTestLinear(rt, &fakeTokenSource{token: "t"}, slog.New(&capturingHandler{})) + + got, err := l.CommentOnIssue(context.Background(), "SEA", 7, "a reply") + if err != nil { + t.Fatalf("CommentOnIssue: %v", err) + } + + commentReq := rt.requests[2] + query, vars := decodeGraphQLReq(t, readReqBody(t, commentReq)) + if !strings.Contains(query, "commentCreate(input: $input)") { + t.Errorf("query missing commentCreate mutation: %q", query) + } + input := vars["input"].(map[string]any) + if input["issueId"] != "issue-uuid-9" { + t.Errorf("issueId = %v, want issue-uuid-9", input["issueId"]) + } + if input["body"] != "a reply" { + t.Errorf("body = %v, want a reply", input["body"]) + } + if input["createAsUser"] != attributionUser { + t.Errorf("createAsUser = %v, want %q", input["createAsUser"], attributionUser) + } + if got.Body != "a reply" || got.URL != "https://linear.app/x/issue/SEA-7#comment-1" { + t.Errorf("decoded Comment = %+v", got) + } +} + +// --- item 3: team-key -> id resolved once, then cached ----------------------- + +func TestLinearTeamIDResolvedOnceThenCached(t *testing.T) { + rt := &scriptedRoundTripper{responses: []scriptedResponse{ + teamResp("team-uuid-1"), // team lookup (once) + probeResp(true), // probe (once) + {status: 200, body: `{"data":{"issueCreate":{"issue":{"number":1,"state":{"type":"unstarted"},"labels":{"nodes":[]},"creator":null}}}}`}, + // Second CreateIssue: probe cached, team cached -> ONLY the create request. + {status: 200, body: `{"data":{"issueCreate":{"issue":{"number":2,"state":{"type":"unstarted"},"labels":{"nodes":[]},"creator":null}}}}`}, + }} + l := newTestLinear(rt, &fakeTokenSource{token: "t"}, slog.New(&capturingHandler{})) + + if _, err := l.CreateIssue(context.Background(), "SEA", CreateIssue{Title: "one"}); err != nil { + t.Fatalf("CreateIssue 1: %v", err) + } + callsAfterFirst := rt.calls + if callsAfterFirst != 3 { + t.Fatalf("first CreateIssue issued %d requests, want 3 (probe+team+create)", callsAfterFirst) + } + + if _, err := l.CreateIssue(context.Background(), "SEA", CreateIssue{Title: "two"}); err != nil { + t.Fatalf("CreateIssue 2: %v", err) + } + if extra := rt.calls - callsAfterFirst; extra != 1 { + t.Fatalf("second CreateIssue issued %d requests, want 1 (create only; team+probe cached)", extra) + } + // The second create's team id came from cache, not a new lookup. + _, vars := decodeGraphQLReq(t, readReqBody(t, rt.requests[3])) + if input := vars["input"].(map[string]any); input["teamId"] != "team-uuid-1" { + t.Errorf("cached teamId = %v, want team-uuid-1", input["teamId"]) + } +} + +// --- item 4: read-query mapping (GetIssue + ListIssues incl. filter state) --- + +func TestLinearGetIssueMapping(t *testing.T) { + rt := &scriptedRoundTripper{responses: []scriptedResponse{ + {status: 200, body: `{"data":{"issues":{"nodes":[{ + "number":7,"title":"a bug","description":"raw body", + "url":"https://linear.app/x/issue/SEA-7","state":{"name":"Done","type":"completed"}, + "labels":{"nodes":[{"name":"bug"},{"name":"p1"}]}, + "creator":{"displayName":"alice"},"updatedAt":"2026-08-01T12:30:00Z"}]}}}`}, + }} + l := newTestLinear(rt, &fakeTokenSource{token: "t"}, slog.New(&capturingHandler{})) + + got, err := l.GetIssue(context.Background(), "SEA", 7) + if err != nil { + t.Fatalf("GetIssue: %v", err) + } + if got.Number != 7 || got.Title != "a bug" { + t.Errorf("scalars wrong: %+v", got) + } + if got.Body != "raw body" { + t.Errorf("body not raw/untouched: %q", got.Body) + } + if got.State != "closed" { + t.Errorf("State = %q, want closed (type=completed)", got.State) + } + if got.ForgeAccount != "alice" { + t.Errorf("ForgeAccount = %q, want alice", got.ForgeAccount) + } + if strings.Join(got.Labels, ",") != "bug,p1" { + t.Errorf("Labels = %v", got.Labels) + } + want := time.Date(2026, 8, 1, 12, 30, 0, 0, time.UTC) + if !got.UpdatedAt.Equal(want) { + t.Errorf("UpdatedAt = %v, want %v", got.UpdatedAt, want) + } + + // The read query filters by team key + number. + _, vars := decodeGraphQLReq(t, readReqBody(t, rt.requests[0])) + filter := vars["filter"].(map[string]any) + team := filter["team"].(map[string]any)["key"].(map[string]any) + if team["eq"] != "SEA" { + t.Errorf("team key filter = %v, want SEA", team["eq"]) + } + if num := filter["number"].(map[string]any); num["eq"] != float64(7) { + t.Errorf("number filter = %v, want 7", num["eq"]) + } +} + +func TestLinearGetIssueNotFound(t *testing.T) { + rt := &scriptedRoundTripper{responses: []scriptedResponse{ + {status: 200, body: `{"data":{"issues":{"nodes":[]}}}`}, + }} + l := newTestLinear(rt, &fakeTokenSource{token: "t"}, slog.New(&capturingHandler{})) + + _, err := l.GetIssue(context.Background(), "SEA", 999) + var se *StatusError + if !errors.As(err, &se) || se.Status != 404 { + t.Fatalf("err = %v, want *StatusError 404", err) + } +} + +func TestLinearListIssuesFilterAndPagination(t *testing.T) { + rt := &scriptedRoundTripper{responses: []scriptedResponse{ + {status: 200, body: `{"data":{"issues":{ + "nodes":[{"number":1,"state":{"type":"started"},"labels":{"nodes":[]},"creator":null}], + "pageInfo":{"hasNextPage":true,"endCursor":"CUR1"}}}}`}, + {status: 200, body: `{"data":{"issues":{ + "nodes":[{"number":2,"state":{"type":"completed"},"labels":{"nodes":[]},"creator":null}], + "pageInfo":{"hasNextPage":false,"endCursor":""}}}}`}, + }} + l := newTestLinear(rt, &fakeTokenSource{token: "t"}, slog.New(&capturingHandler{})) + + got, err := l.ListIssues(context.Background(), "SEA", IssueFilter{State: "open", Labels: []string{"bug"}}) + if err != nil { + t.Fatalf("ListIssues: %v", err) + } + if len(got) != 2 || got[0].Number != 1 || got[1].Number != 2 { + t.Fatalf("walk concatenation wrong: %+v", got) + } + if got[0].State != "open" || got[1].State != "closed" { + t.Errorf("state mapping wrong: %q, %q", got[0].State, got[1].State) + } + + // Page 1 request: open-state filter -> type nin closed; label AND sub-filter. + _, vars := decodeGraphQLReq(t, readReqBody(t, rt.requests[0])) + filter := vars["filter"].(map[string]any) + stateType := filter["state"].(map[string]any)["type"].(map[string]any) + if _, ok := stateType["nin"]; !ok { + t.Errorf("open state should map to type nin closed; got %#v", stateType) + } + if _, ok := filter["and"]; !ok { + t.Errorf("label filter should produce an `and` of some-sub-filters; got %#v", filter) + } + if _, ok := vars["after"]; ok { + t.Errorf("page 1 should not send an after cursor; got %v", vars["after"]) + } + + // Page 2 request carries the endCursor from page 1. + _, vars2 := decodeGraphQLReq(t, readReqBody(t, rt.requests[1])) + if vars2["after"] != "CUR1" { + t.Errorf("page 2 after = %v, want CUR1", vars2["after"]) + } +} + +func TestLinearListIssuesClosedState(t *testing.T) { + rt := &scriptedRoundTripper{responses: []scriptedResponse{ + {status: 200, body: `{"data":{"issues":{"nodes":[],"pageInfo":{"hasNextPage":false,"endCursor":""}}}}`}, + }} + l := newTestLinear(rt, &fakeTokenSource{token: "t"}, slog.New(&capturingHandler{})) + + if _, err := l.ListIssues(context.Background(), "SEA", IssueFilter{State: "closed"}); err != nil { + t.Fatalf("ListIssues: %v", err) + } + _, vars := decodeGraphQLReq(t, readReqBody(t, rt.requests[0])) + stateType := vars["filter"].(map[string]any)["state"].(map[string]any)["type"].(map[string]any) + if _, ok := stateType["in"]; !ok { + t.Errorf("closed state should map to type in closed; got %#v", stateType) + } +} + +// --- item 5: ErrUnsupported for the five PR/review ops ----------------------- + +func TestLinearUnsupportedOps(t *testing.T) { + rt := &scriptedRoundTripper{} + l := newTestLinear(rt, &fakeTokenSource{token: "t"}, slog.New(&capturingHandler{})) + ctx := context.Background() + + if _, err := l.CreatePullRequest(ctx, "SEA", CreatePR{}); !errors.Is(err, ErrUnsupported) { + t.Errorf("CreatePullRequest err = %v, want ErrUnsupported", err) + } + if _, err := l.CommentOnPullRequest(ctx, "SEA", 1, "x"); !errors.Is(err, ErrUnsupported) { + t.Errorf("CommentOnPullRequest err = %v, want ErrUnsupported", err) + } + if _, err := l.SubmitReview(ctx, "SEA", 1, SubmitReview{}); !errors.Is(err, ErrUnsupported) { + t.Errorf("SubmitReview err = %v, want ErrUnsupported", err) + } + if _, err := l.GetPullRequest(ctx, "SEA", 1); !errors.Is(err, ErrUnsupported) { + t.Errorf("GetPullRequest err = %v, want ErrUnsupported", err) + } + if _, err := l.Checks(ctx, "SEA", 1); !errors.Is(err, ErrUnsupported) { + t.Errorf("Checks err = %v, want ErrUnsupported", err) + } + // The unsupported ops make no wire call. + if rt.calls != 0 { + t.Errorf("unsupported ops issued %d requests, want 0", rt.calls) + } +} + +// --- item 6: 429 -> resource_exhausted (ErrBudgetExhausted) ------------------ + +func TestLinear429MapsToBudgetExhausted(t *testing.T) { + base := time.Date(2026, 8, 9, 12, 0, 0, 0, time.UTC) + rt := &scriptedRoundTripper{responses: []scriptedResponse{ + {status: 429, body: `{"errors":[{"message":"rate limited"}]}`, headers: map[string]string{"Retry-After": "30"}}, + }} + ts := &fakeTokenSource{token: "t"} + l := newTestLinear(rt, ts, slog.New(&capturingHandler{})) + l.now = func() time.Time { return base } + + // A read triggers the 429 directly (no probe on the read path). + _, err := l.GetIssue(context.Background(), "SEA", 7) + if !errors.Is(err, ErrBudgetExhausted) { + t.Fatalf("err = %v, want ErrBudgetExhausted", err) + } + if ts.invalidated != 0 { + t.Errorf("rate-limit must not Invalidate; got %d", ts.invalidated) + } + // The gate is armed: the next call fails fast without a request. + if _, err := l.GetIssue(context.Background(), "SEA", 8); !errors.Is(err, ErrBudgetExhausted) { + t.Fatalf("gated call err = %v, want ErrBudgetExhausted", err) + } + if rt.calls != 1 { + t.Errorf("gate issued a request: calls = %d, want 1", rt.calls) + } +} + +func TestLinearGraphQLRateLimitedCodeOn400(t *testing.T) { + rt := &scriptedRoundTripper{responses: []scriptedResponse{ + {status: 400, body: `{"errors":[{"message":"complex","extensions":{"code":"RATELIMITED"}}]}`}, + }} + l := newTestLinear(rt, &fakeTokenSource{token: "t"}, slog.New(&capturingHandler{})) + + _, err := l.GetIssue(context.Background(), "SEA", 7) + if !errors.Is(err, ErrBudgetExhausted) { + t.Fatalf("err = %v, want ErrBudgetExhausted (RATELIMITED code)", err) + } +} + +// --- item 7: GraphQL errors on HTTP 200 -> *StatusError ---------------------- + +func TestLinearGraphQLErrorsOn200(t *testing.T) { + rt := &scriptedRoundTripper{responses: []scriptedResponse{ + {status: 200, body: `{"errors":[{"message":"Field bad"},{"message":"also bad"}]}`}, + }} + l := newTestLinear(rt, &fakeTokenSource{token: "t"}, slog.New(&capturingHandler{})) + + _, err := l.GetIssue(context.Background(), "SEA", 7) + var se *StatusError + if !errors.As(err, &se) { + t.Fatalf("err = %v, want *StatusError", err) + } + if se.Status != 200 { + t.Errorf("Status = %d, want 200 (GraphQL error on a 200)", se.Status) + } + if se.Message != "Field bad; also bad" { + t.Errorf("Message = %q, want joined messages", se.Message) + } +} + +// --- item 8: actor probe FAILS -> no createAsUser + the exact log line ------- + +func TestLinearActorProbeDegradesToStampOnly(t *testing.T) { + cap := &capturingHandler{} + rt := &scriptedRoundTripper{responses: []scriptedResponse{ + teamResp("team-uuid-1"), + probeResp(false), // probe: NOT an actor=app token + {status: 200, body: `{"data":{"issueCreate":{"issue":{"number":1,"state":{"type":"unstarted"},"labels":{"nodes":[]},"creator":null}}}}`}, + // Second create: probe cached -> team+create only, still no createAsUser. + {status: 200, body: `{"data":{"issueCreate":{"issue":{"number":2,"state":{"type":"unstarted"},"labels":{"nodes":[]},"creator":null}}}}`}, + }} + l := newTestLinear(rt, &fakeTokenSource{token: "t"}, slog.New(cap)) + + if _, err := l.CreateIssue(context.Background(), "SEA", CreateIssue{Title: "x"}); err != nil { + t.Fatalf("CreateIssue 1: %v", err) + } + _, vars := decodeGraphQLReq(t, readReqBody(t, rt.requests[2])) + input := vars["input"].(map[string]any) + if _, ok := input["createAsUser"]; ok { + t.Errorf("degraded write must NOT set createAsUser; got %#v", input) + } + if _, ok := input["displayIconUrl"]; ok { + t.Errorf("degraded write must NOT set displayIconUrl; got %#v", input) + } + if !cap.has(degradeMsg) { + t.Fatalf("expected the exact degrade log line %q; got %v", degradeMsg, cap.msgs) + } + + // The probe (and its log line) fire exactly once, even across writes. + if _, err := l.CreateIssue(context.Background(), "SEA", CreateIssue{Title: "y"}); err != nil { + t.Fatalf("CreateIssue 2: %v", err) + } + if n := cap.count(degradeMsg); n != 1 { + t.Errorf("degrade line logged %d times, want exactly 1", n) + } +} + +// --- item 9: 401 -> TokenSource.Invalidate + StatusError --------------------- + +func TestLinear401Invalidates(t *testing.T) { + rt := &scriptedRoundTripper{responses: []scriptedResponse{ + {status: 401, body: `{"errors":[{"message":"Authentication required"}]}`}, + }} + ts := &fakeTokenSource{token: "t"} + l := newTestLinear(rt, ts, slog.New(&capturingHandler{})) + + _, err := l.GetIssue(context.Background(), "SEA", 7) + var se *StatusError + if !errors.As(err, &se) || se.Status != 401 { + t.Fatalf("err = %v, want *StatusError 401", err) + } + if ts.invalidated != 1 { + t.Errorf("401 must Invalidate; got %d", ts.invalidated) + } +} + +func TestLinearGraphQLAuthErrorCodeInvalidates(t *testing.T) { + rt := &scriptedRoundTripper{responses: []scriptedResponse{ + {status: 200, body: `{"errors":[{"message":"bad token","extensions":{"code":"AUTHENTICATION_ERROR"}}]}`}, + }} + ts := &fakeTokenSource{token: "t"} + l := newTestLinear(rt, ts, slog.New(&capturingHandler{})) + + _, err := l.GetIssue(context.Background(), "SEA", 7) + var se *StatusError + if !errors.As(err, &se) { + t.Fatalf("err = %v, want *StatusError", err) + } + if ts.invalidated != 1 { + t.Errorf("AUTHENTICATION_ERROR must Invalidate; got %d", ts.invalidated) + } +} + +// --- misc: token error short-circuits before any request -------------------- + +func TestLinearTokenErrorNoWire(t *testing.T) { + rt := &scriptedRoundTripper{} + tokErr := errors.New("resolve failed") + l := newTestLinear(rt, &fakeTokenSource{err: tokErr}, slog.New(&capturingHandler{})) + + _, err := l.GetIssue(context.Background(), "SEA", 7) + if !errors.Is(err, tokErr) { + t.Fatalf("err = %v, want token error", err) + } + if rt.calls != 0 { + t.Errorf("issued a request despite token error: calls = %d", rt.calls) + } +} + +func TestLinearBodyLimit(t *testing.T) { + l := newTestLinear(&scriptedRoundTripper{}, &fakeTokenSource{token: "t"}, slog.New(&capturingHandler{})) + if got := l.BodyLimit(); got != linearBodyLimit { + t.Errorf("BodyLimit() = %d, want %d", got, linearBodyLimit) + } +} + +func TestLinearName(t *testing.T) { + l := newTestLinear(&scriptedRoundTripper{}, &fakeTokenSource{token: "t"}, slog.New(&capturingHandler{})) + if got := l.Name(); got != "linear" { + t.Errorf("Name() = %q, want linear", got) + } +} From 71941d7ce9e99f96e8e8c9338222c41c2f7d73c2 Mon Sep 17 00:00:00 2001 From: seal Date: Tue, 18 Aug 2026 23:20:34 -0400 Subject: [PATCH 2/2] fix(forge): re-probe transient actor failures + pin Linear budget gate (RIG-2209 review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review M1: actorAttribution cached a TRANSIENT probe failure (a network blip, HTTP 5xx, or an already-armed rate gate) as a permanent not-capable, poisoning attribution to stamp-only for the whole process even after the token/budget recovered. The probe is meant to reflect the token's nature, not runtime state — so only cache an authoritative answer (probeErr == nil); a transient error degrades this one write and re-probes on the next. Review M2/L2: add success-path budget-gate coverage (a 200 whose X-Ratelimit-Requests-Remaining equals the reserve arms the gate; above-reserve leaves it open) and both untested rateLimitReset branches (Retry-After HTTP-date, and the X-Ratelimit-Requests-Reset epoch-ms fallback). Review L1: ListIssues could loop forever if Linear ever returned hasNextPage=true with an empty endCursor (advancing on an empty cursor drops the after variable and refetches page 1); terminate on a malformed page instead. New M1 regression test is red-green verified (fails against the permanent-cache behavior, passes with the fix). All four gates green (gofmt/vet/-race/lint); 21 Linear tests pass race-clean. Refs RIG-2209. Co-authored-by: Matt Wilkinson --- go/internal/forge/linear.go | 41 ++++---- go/internal/forge/linear_test.go | 154 +++++++++++++++++++++++++++++-- 2 files changed, 171 insertions(+), 24 deletions(-) diff --git a/go/internal/forge/linear.go b/go/internal/forge/linear.go index 593b832f..df9551ff 100644 --- a/go/internal/forge/linear.go +++ b/go/internal/forge/linear.go @@ -262,10 +262,14 @@ func (l *Linear) ListIssues(ctx context.Context, repo string, f IssueFilter) ([] for _, n := range out.Issues.Nodes { all = append(all, n.toIssue()) } - if !out.Issues.PageInfo.HasNextPage { + next := out.Issues.PageInfo.EndCursor + // Terminate on end-of-pages OR a malformed page (hasNextPage with an + // empty cursor): advancing on an empty cursor would drop the `after` + // variable and refetch page 1 forever. + if !out.Issues.PageInfo.HasNextPage || next == "" { break } - after = out.Issues.PageInfo.EndCursor + after = next } return all, nil } @@ -533,11 +537,14 @@ func (l *Linear) resolveIssueID(ctx context.Context, repo string, number uint64) } // actorAttribution reports whether writes may set createAsUser, running the -// one-time capability probe on first call and caching the result. The probe +// capability probe on first call and caching an AUTHORITATIVE result. The probe // queries `viewer { app }`: an actor=app OAuth token authenticates AS the app, -// so viewer.app is true; a plain user/API-key token reports false. A probe that -// errors is treated as not-capable (degrade, never block the write). On the -// not-capable transition it emits the named degrade log line EXACTLY once. +// so viewer.app is true; a plain user/API-key token reports false. The probe is +// meant to reflect the token's NATURE, not a transient runtime state — so a +// probe that ERRORS (network blip, HTTP 5xx, or an already-armed rate gate) +// degrades THIS write to stamp-only WITHOUT caching, letting a later write +// re-probe once the transient condition clears; only a clean answer +// (probeErr == nil) is cached. On the first degrade it emits the named log line. func (l *Linear) actorAttribution(ctx context.Context) bool { l.mu.Lock() if l.probeDone { @@ -556,24 +563,26 @@ func (l *Linear) actorAttribution(ctx context.Context) bool { } `json:"viewer"` } probeErr := l.doGraphQL(ctx, query, nil, &out) - capable := probeErr == nil && out.Viewer.App l.mu.Lock() defer l.mu.Unlock() if l.probeDone { - // A concurrent caller finished the probe first; honor its result. + // A concurrent caller finished an authoritative probe first; honor it. return l.actorCapable } + if probeErr != nil { + // Transient failure — degrade this write but do NOT cache, so a later + // write re-probes. Log the degrade line once per transient occurrence. + l.log.Warn("linear: actor attribution unavailable; degrading to stamp-only", "probe_error", probeErr) + return false + } + // Authoritative answer: cache it. A definitive not-capable also degrades. l.probeDone = true - l.actorCapable = capable - if !capable { - if probeErr != nil { - l.log.Warn("linear: actor attribution unavailable; degrading to stamp-only", "probe_error", probeErr) - } else { - l.log.Warn("linear: actor attribution unavailable; degrading to stamp-only") - } + l.actorCapable = out.Viewer.App + if !l.actorCapable { + l.log.Warn("linear: actor attribution unavailable; degrading to stamp-only") } - return capable + return l.actorCapable } // applyAttribution sets createAsUser/displayIconUrl on a mutation input when the diff --git a/go/internal/forge/linear_test.go b/go/internal/forge/linear_test.go index ed22ddfd..c90e4397 100644 --- a/go/internal/forge/linear_test.go +++ b/go/internal/forge/linear_test.go @@ -73,10 +73,9 @@ func probeResp(app bool) scriptedResponse { return scriptedResponse{status: 200, body: `{"data":{"viewer":{"app":` + strconv.FormatBool(app) + `}}}`} } -// teamResp is a scripted team-key->id lookup response. -func teamResp(id string) scriptedResponse { - return scriptedResponse{status: 200, body: `{"data":{"teams":{"nodes":[{"id":"` + id + `"}]}}}`} -} +// teamResp is a scripted team-key->id lookup response (the id is fixed; tests +// assert behavior, not the specific UUID). +var teamResp = scriptedResponse{status: 200, body: `{"data":{"teams":{"nodes":[{"id":"team-uuid-1"}]}}}`} const degradeMsg = "linear: actor attribution unavailable; degrading to stamp-only" @@ -97,7 +96,7 @@ func decodeGraphQLReq(t *testing.T, body string) (string, map[string]any) { func TestLinearCreateIssueRequestGolden(t *testing.T) { rt := &scriptedRoundTripper{responses: []scriptedResponse{ - teamResp("team-uuid-1"), + teamResp, probeResp(true), // actor probe: capable {status: 200, body: `{"data":{"issueCreate":{"issue":{ "number":42,"title":"a bug","description":"stamped body", @@ -184,8 +183,8 @@ func TestLinearCommentOnIssueRequestGolden(t *testing.T) { func TestLinearTeamIDResolvedOnceThenCached(t *testing.T) { rt := &scriptedRoundTripper{responses: []scriptedResponse{ - teamResp("team-uuid-1"), // team lookup (once) - probeResp(true), // probe (once) + teamResp, // team lookup (once) + probeResp(true), // probe (once) {status: 200, body: `{"data":{"issueCreate":{"issue":{"number":1,"state":{"type":"unstarted"},"labels":{"nodes":[]},"creator":null}}}}`}, // Second CreateIssue: probe cached, team cached -> ONLY the create request. {status: 200, body: `{"data":{"issueCreate":{"issue":{"number":2,"state":{"type":"unstarted"},"labels":{"nodes":[]},"creator":null}}}}`}, @@ -427,7 +426,7 @@ func TestLinearGraphQLErrorsOn200(t *testing.T) { func TestLinearActorProbeDegradesToStampOnly(t *testing.T) { cap := &capturingHandler{} rt := &scriptedRoundTripper{responses: []scriptedResponse{ - teamResp("team-uuid-1"), + teamResp, probeResp(false), // probe: NOT an actor=app token {status: 200, body: `{"data":{"issueCreate":{"issue":{"number":1,"state":{"type":"unstarted"},"labels":{"nodes":[]},"creator":null}}}}`}, // Second create: probe cached -> team+create only, still no createAsUser. @@ -524,3 +523,142 @@ func TestLinearName(t *testing.T) { t.Errorf("Name() = %q, want linear", got) } } + +// linearIssueNodeResp is a minimal valid issues-query response (GetIssue reads +// the first node); enough fields to decode through toIssue without error. +func linearIssueNodeResp(number int) string { + return `{"data":{"issues":{"nodes":[{"number":` + strconv.Itoa(number) + + `,"state":{"type":"unstarted"},"labels":{"nodes":[]},"creator":null}]}}}` +} + +// --- budget recording on the SUCCESS path (proactive gate) ------------------- + +// A 200 whose X-Ratelimit-Requests-Remaining EQUALS the reserve arms the gate, +// so the next call fails fast without a wire request. Pins the `remaining > +// reserve` boundary, the recordBudget call site, and the epoch-ms reset parse. +func TestLinearBudgetGateRemainingAtReserveArms(t *testing.T) { + base := time.Date(2026, 8, 9, 12, 0, 0, 0, time.UTC) + resetMS := strconv.FormatInt(base.Add(time.Minute).UnixMilli(), 10) + rt := &scriptedRoundTripper{responses: []scriptedResponse{ + {status: 200, body: linearIssueNodeResp(7), headers: map[string]string{ + "X-Ratelimit-Requests-Remaining": strconv.Itoa(reserve), // == reserve -> arms + "X-Ratelimit-Requests-Reset": resetMS, + }}, + }} + l := newTestLinear(rt, &fakeTokenSource{token: "t"}, slog.New(&capturingHandler{})) + l.now = func() time.Time { return base } + + if _, err := l.GetIssue(context.Background(), "SEA", 7); err != nil { + t.Fatalf("GetIssue: %v", err) + } + if _, err := l.GetIssue(context.Background(), "SEA", 8); !errors.Is(err, ErrBudgetExhausted) { + t.Fatalf("gated call err = %v, want ErrBudgetExhausted", err) + } + if rt.calls != 1 { + t.Errorf("gate issued a request: calls = %d, want 1", rt.calls) + } +} + +// A 200 with remaining ABOVE the reserve leaves the gate open: the next call +// issues a real request. Guards against a `>`→`>=` regression at the boundary. +func TestLinearBudgetGateRemainingAboveReserveStaysOpen(t *testing.T) { + base := time.Date(2026, 8, 9, 12, 0, 0, 0, time.UTC) + resetMS := strconv.FormatInt(base.Add(time.Minute).UnixMilli(), 10) + rt := &scriptedRoundTripper{responses: []scriptedResponse{ + {status: 200, body: linearIssueNodeResp(7), headers: map[string]string{ + "X-Ratelimit-Requests-Remaining": strconv.Itoa(reserve + 1), // > reserve -> open + "X-Ratelimit-Requests-Reset": resetMS, + }}, + {status: 200, body: linearIssueNodeResp(8)}, + }} + l := newTestLinear(rt, &fakeTokenSource{token: "t"}, slog.New(&capturingHandler{})) + l.now = func() time.Time { return base } + + if _, err := l.GetIssue(context.Background(), "SEA", 7); err != nil { + t.Fatalf("GetIssue 1: %v", err) + } + if _, err := l.GetIssue(context.Background(), "SEA", 8); err != nil { + t.Fatalf("GetIssue 2 (gate should be open): %v", err) + } + if rt.calls != 2 { + t.Errorf("calls = %d, want 2 (gate stayed open)", rt.calls) + } +} + +// --- rateLimitReset: both non-delta-seconds branches ------------------------- + +// A 429 with Retry-After as an HTTP-date arms the gate at exactly that instant. +func TestLinearRateLimitResetHTTPDate(t *testing.T) { + base := time.Date(2026, 8, 9, 12, 0, 0, 0, time.UTC) + at := base.Add(45 * time.Second) + rt := &scriptedRoundTripper{responses: []scriptedResponse{ + {status: 429, body: `{"errors":[{"message":"rate limited"}]}`, headers: map[string]string{ + "Retry-After": at.Format(http.TimeFormat), + }}, + }} + l := newTestLinear(rt, &fakeTokenSource{token: "t"}, slog.New(&capturingHandler{})) + l.now = func() time.Time { return base } + + if _, err := l.GetIssue(context.Background(), "SEA", 7); !errors.Is(err, ErrBudgetExhausted) { + t.Fatalf("err = %v, want ErrBudgetExhausted", err) + } + if !l.resetAt.Equal(at) { + t.Errorf("resetAt = %v, want %v (Retry-After HTTP-date)", l.resetAt, at) + } +} + +// A 429 with no Retry-After falls back to X-Ratelimit-Requests-Reset (epoch ms). +func TestLinearRateLimitResetHeaderFallback(t *testing.T) { + base := time.Date(2026, 8, 9, 12, 0, 0, 0, time.UTC) + at := base.Add(90 * time.Second) + rt := &scriptedRoundTripper{responses: []scriptedResponse{ + {status: 429, body: `{"errors":[{"message":"rate limited"}]}`, headers: map[string]string{ + "X-Ratelimit-Requests-Reset": strconv.FormatInt(at.UnixMilli(), 10), + }}, + }} + l := newTestLinear(rt, &fakeTokenSource{token: "t"}, slog.New(&capturingHandler{})) + l.now = func() time.Time { return base } + + if _, err := l.GetIssue(context.Background(), "SEA", 7); !errors.Is(err, ErrBudgetExhausted) { + t.Fatalf("err = %v, want ErrBudgetExhausted", err) + } + if !l.resetAt.Equal(time.UnixMilli(at.UnixMilli())) { + t.Errorf("resetAt = %v, want %v (epoch-ms fallback)", l.resetAt, at) + } +} + +// --- actor probe: a TRANSIENT error is not cached; a later write re-probes --- + +func TestLinearActorProbeTransientErrorReprobed(t *testing.T) { + cap := &capturingHandler{} + rt := &scriptedRoundTripper{responses: []scriptedResponse{ + teamResp, + // Write 1's probe hits a transient 500 -> degrade this write, do NOT cache. + {status: 500, body: `{"errors":[{"message":"internal"}]}`}, + {status: 200, body: `{"data":{"issueCreate":{"issue":{"number":1,"state":{"type":"unstarted"},"labels":{"nodes":[]},"creator":null}}}}`}, + // Write 2 re-probes (transient cleared) -> capable -> createAsUser set. + probeResp(true), + {status: 200, body: `{"data":{"issueCreate":{"issue":{"number":2,"state":{"type":"unstarted"},"labels":{"nodes":[]},"creator":null}}}}`}, + }} + l := newTestLinear(rt, &fakeTokenSource{token: "t"}, slog.New(cap)) + + if _, err := l.CreateIssue(context.Background(), "SEA", CreateIssue{Title: "x"}); err != nil { + t.Fatalf("CreateIssue 1: %v", err) + } + _, vars1 := decodeGraphQLReq(t, readReqBody(t, rt.requests[2])) + if _, ok := vars1["input"].(map[string]any)["createAsUser"]; ok { + t.Errorf("transient-probe-fail write must NOT set createAsUser") + } + if !cap.has(degradeMsg) { + t.Errorf("transient probe failure must emit the degrade line") + } + + if _, err := l.CreateIssue(context.Background(), "SEA", CreateIssue{Title: "y"}); err != nil { + t.Fatalf("CreateIssue 2: %v", err) + } + _, vars2 := decodeGraphQLReq(t, readReqBody(t, rt.requests[4])) + if vars2["input"].(map[string]any)["createAsUser"] != attributionUser { + t.Errorf("re-probe capable write must set createAsUser=%q (proves no permanent cache); got %#v", + attributionUser, vars2["input"]) + } +}