# pr_lint.exs: draft a pull request description with Together AI chat completions, or lint one. # elixir pr_lint.exs --draft "retry webhook sends, add jitter, cap at 5 tries" # elixir pr_lint.exs samples/robotic.md # elixir pr_lint.exs --json samples/robotic.md # Exit 0 = ready to open, 1 = rewrite it, 2 = the HTTP call or the input failed. # Elixir + Erlang/OTP stdlib only (:httpc, :ssl, JSON). No Hex packages. defmodule PrLint do @max_title 72 @max_bullets 8 @preamble ~r/\A\s*(sure|certainly|absolutely|of course|here(?:'s| is| are))\b/i @boilerplate ~r/\A\s*(#+\s*)?(summary\s*:?\s*)?(this (pr|pull request) (aims|seeks|is designed) to|in this (pr|pull request),? (we|i))/i @filler ~r/\b(seamless(ly)?|leverag(e|es|ing)|robust|comprehensive|streamlin(e|es|ed)|enhanc(e|es|ed) the overall|delve|elevate|game[- ]changer|cutting[- ]edge|as an ai|ensure a smooth|holistic|paradigm)\b/i @testing ~r/^\s*(#+\s*)?(\*\*)?(testing|tested|test plan|how to test|verification)\b/im @bullet ~r/^\s*[-*+]\s+/m def fence, do: String.duplicate("`", 3) def lint(text) do t = String.trim(text) if t == "" do ["empty description"] else title = t |> String.split("\n", parts: 2) |> hd() |> String.trim_leading("#") |> String.trim() bullets = length(Regex.scan(@bullet, t)) emoji = t |> String.to_charlist() |> Enum.count(&emoji?/1) [ {String.length(title) > @max_title, "first line is #{String.length(title)} chars (keep it to #{@max_title})"}, {Regex.match?(@preamble, t), "model preamble (Here is / Sure / Certainly)"}, {String.starts_with?(t, fence()), "whole description wrapped in a code fence"}, {Regex.match?(@boilerplate, t), "boilerplate opener (This PR aims to ...)"}, {Regex.match?(@filler, t), "robotic filler phrase"}, {not Regex.match?(@testing, t), "no Testing / Test plan line"}, {bullets > @max_bullets, "#{bullets} bullets (reviewers skim; keep it to #{@max_bullets})"}, {emoji > 1, "more than one emoji (#{emoji})"} ] |> Enum.filter(&elem(&1, 0)) |> Enum.map(&elem(&1, 1)) end end defp emoji?(c), do: c in 0x1F300..0x1FAFF or c in 0x2600..0x27BF def draft(notes) do key = System.get_env("TOGETHER_API_KEY") || die("TOGETHER_API_KEY is not set") base = System.get_env("TOGETHER_BASE_URL", "https://api.together.ai") |> String.trim_trailing("/") model = System.get_env("TOGETHER_MODEL", "Qwen/Qwen3.5-9B") body = JSON.encode!(%{ "model" => model, "max_tokens" => 400, "messages" => [ %{"role" => "system", "content" => "You write GitHub pull request descriptions. Plain text or light markdown. " <> "First line: what changed, under 72 characters. Then why, then a Testing line. " <> "No preamble, no code fence around the answer, no marketing words."}, %{"role" => "user", "content" => "Change notes: " <> notes} ] }) {:ok, _} = Application.ensure_all_started(:inets) {:ok, _} = Application.ensure_all_started(:ssl) url = String.to_charlist(base <> "/v1/chat/completions") headers = [{~c"authorization", String.to_charlist("Bearer " <> key)}] http_opts = [ timeout: 30_000, connect_timeout: 10_000, ssl: [ verify: :verify_peer, cacerts: :public_key.cacerts_get(), customize_hostname_check: [match_fun: :public_key.pkix_verify_hostname_match_fun(:https)] ] ] case :httpc.request(:post, {url, headers, ~c"application/json", body}, http_opts, body_format: :binary) do {:ok, {{_, status, _}, _h, resp}} when status in 200..299 -> case JSON.decode(resp) do {:ok, %{"choices" => [%{"message" => %{"content" => c}} | _]}} when is_binary(c) and c != "" -> c _ -> die("no choices[0].message.content in the response") end {:ok, {{_, status, _}, _h, resp}} -> die("HTTP #{status}: #{String.slice(resp, 0, 200)}") {:error, reason} -> die("request failed: #{inspect(reason)}") end end def die(msg) do IO.puts(:stderr, "pr_lint: " <> msg) System.halt(2) end def main(args) do {json?, rest} = {"--json" in args, Enum.reject(args, &(&1 == "--json"))} {source, text} = case rest do ["--draft", notes | _] -> {"together", draft(notes)} [path | _] -> case File.read(path) do {:ok, t} -> {path, t} {:error, why} -> die("cannot read #{path}: #{why}") end [] -> die("usage: elixir pr_lint.exs [--json] (--draft \"notes\" | file.md)") end problems = lint(text) if json? do IO.puts(JSON.encode!(%{"source" => source, "ok" => problems == [], "problems" => problems})) else if source == "together", do: IO.puts(text <> "\n---") if problems == [], do: IO.puts("ok: #{source}"), else: Enum.each(problems, &IO.puts("#{source}: #{&1}")) end System.halt(if problems == [], do: 0, else: 1) end end unless System.get_env("PR_LINT_NO_MAIN"), do: PrLint.main(System.argv()) 

# PR_LINT_NO_MAIN=1 elixir pr_lint_test.exs ExUnit.start() Code.require_file("pr_lint.exs", __DIR__) defmodule PrLintTest do use ExUnit.Case, async: true test "clean sample passes" do assert PrLint.lint(File.read!("samples/clean.md")) == [] end test "robotic sample collects several problems" do p = PrLint.lint(File.read!("samples/robotic.md")) assert "model preamble (Here is / Sure / Certainly)" in p assert "robotic filler phrase" in p assert "no Testing / Test plan line" in p assert Enum.any?(p, &String.starts_with?(&1, "more than one emoji")) end test "bullet wall is flagged" do assert Enum.any?(PrLint.lint(File.read!("samples/long.md")), &String.contains?(&1, "11 bullets")) end test "boilerplate opener and long first line" do p = PrLint.lint("This PR aims to fix the thing that has been bothering everyone on the team since last March\n\nTesting: ran it") assert "boilerplate opener (This PR aims to ...)" in p assert Enum.any?(p, &String.starts_with?(&1, "first line is")) end test "fence wrapper and empty input" do f = PrLint.fence() assert "whole description wrapped in a code fence" in PrLint.lint(f <> "\nFix\n\nTesting: yes\n" <> f) assert PrLint.lint(" ") == ["empty description"] end end 

#!/usr/bin/env python3 # Local fake of Together's /v1/chat/completions endpoint. Standard library only. # usage: python3 mock_together.py MODE PORT (MODE: robotic | clean | untested | error | empty) import json import sys from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler args = sys.argv[1:] + ["robotic", "8799"][len(sys.argv[1:]):] MODE, PORT = args[0], int(args[1]) TICKS = "`" * 3 CANNED = { "robotic": "Certainly! Here is your PR description:\n" + TICKS + "\n" "This pull request aims to leverage a robust, comprehensive retry layer.\n" + TICKS, "clean": "Retry webhook sends with jitter, capped at 5 tries\n\n" "A single 502 used to drop the event. Sends now back off and stop after 5 attempts.\n\n" "Testing: backoff unit tests and a stub endpoint that fails twice.", "untested": "Retry webhook sends with jitter\n\nA single 502 used to drop the event.", "empty": "", } class FakeTogether(BaseHTTPRequestHandler): def reply(self, code, obj): raw = json.dumps(obj).encode() self.send_response(code) for k, v in (("Content-Type", "application/json"), ("Content-Length", str(len(raw)))): self.send_header(k, v) self.end_headers() self.wfile.write(raw) def do_POST(self): sent = self.rfile.read(int(self.headers.get("Content-Length") or 0)) try: req = json.loads(sent) except ValueError: req = {} bearer = self.headers.get("Authorization", "").startswith("Bearer ") roles = [m.get("role") for m in req.get("messages", [])] print(f"path={self.path} auth={'yes' if bearer else 'no'} ctype={self.headers.get('Content-Type')} " f"model={req.get('model')} max_tokens={req.get('max_tokens')} roles={roles}", flush=True) if MODE == "error": return self.reply(401, {"error": {"message": "Invalid API key", "type": "invalid_request_error"}}) choice = {"index": 0, "finish_reason": "stop", "message": {"role": "assistant", "content": CANNED[MODE]}} self.reply(200, {"id": "mock-1", "object": "chat.completion", "model": req.get("model"), "choices": [choice]}) def log_message(self, *_): return ThreadingHTTPServer(("127.0.0.1", PORT), FakeTogether).serve_forever() 

# needs Elixir 1.18+ (built-in JSON) and OTP 25+ (:public_key.cacerts_get) export PATH="$HOME/tools/elixir/bin:$HOME/tools/otp/bin:$PATH" # lint a saved description elixir pr_lint.exs samples/clean.md # ask Together for a draft, then lint it export TOGETHER_API_KEY=... elixir pr_lint.exs --draft "retry webhook sends, add jitter, cap at 5 tries" # point it at the mock python3 mock_together.py robotic 8799 & TOGETHER_API_KEY=test-key TOGETHER_BASE_URL=http://127.0.0.1:8799 \ elixir pr_lint.exs --draft "retry webhook sends" # unit tests PR_LINT_NO_MAIN=1 elixir pr_lint_test.exs