Something went wrong. Try again.
small gleam coding and persistent agent daemon with a detachable cli and a canvas ui forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148"""Run gopls over changed Go files with staticcheck and every analyzer on.
`gopls check` cannot enable analyses, so this speaks LSP to `gopls serve`with the settings agents.md asks for. Run it inside `nix develop` so thepinned gopls matches the pinned Go:
python3 test/gopls_all.py # Go files changed against origin/main python3 test/gopls_all.py internal/tui/cards.go ... # paths under cli/
It prints one line per diagnostic and exits 1 when there are any."""
import jsonimport subprocessimport sysfrom pathlib import Pathfrom typing import Any
ROOT = Path(__file__).resolve().parents[1]MODULE = ROOT / "cli"
def changed_files() -> list[str]: """Go files under cli/ changed against origin/main, staged or not, plus new ones.""" names: set[str] = set() for args in ( ["diff", "--name-only", "origin/main...HEAD"], ["diff", "--name-only", "HEAD"], ["ls-files", "--others", "--exclude-standard"], ): listed = subprocess.run( ["git", "-C", str(ROOT), *args], capture_output=True, text=True, check=True ).stdout names.update(listed.split()) return sorted( name.removeprefix("cli/") for name in names if name.startswith("cli/") and name.endswith(".go") and (ROOT / name).exists() )
class Server: def __init__(self, settings: dict[str, Any]) -> None: self.settings = settings self.diagnostics: dict[str, list[dict[str, Any]]] = {} self.process = subprocess.Popen( ["gopls", "serve"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, cwd=MODULE, )
def send(self, message: dict[str, Any]) -> None: assert self.process.stdin body = json.dumps({"jsonrpc": "2.0", **message}).encode() self.process.stdin.write(b"Content-Length: %d\r\n\r\n" % len(body) + body) self.process.stdin.flush()
def receive(self) -> dict[str, Any]: assert self.process.stdout length = 0 while (line := self.process.stdout.readline()) not in (b"\r\n", b""): name, _, value = line.decode().partition(":") if name.lower() == "content-length": length = int(value) return json.loads(self.process.stdout.read(length))
def request(self, id: int, method: str, params: Any) -> dict[str, Any]: """Send a request and answer the server's own requests until its reply.""" self.send({"id": id, "method": method, "params": params}) while True: message = self.receive() if message.get("method") == "textDocument/publishDiagnostics": published = message["params"] self.diagnostics[published["uri"]] = published.get("diagnostics", []) elif "method" in message and "id" in message: items = message["params"].get("items", []) result = ( [self.settings for _ in items] if message["method"] == "workspace/configuration" else None ) self.send({"id": message["id"], "result": result}) elif message.get("id") == id: return message
def main(argv: list[str]) -> int: paths = [MODULE / name for name in (argv or changed_files())] if not paths: print("no changed Go files") return 0 analyzers = json.loads(subprocess.check_output(["gopls", "api-json"]))["Analyzers"] settings = { "staticcheck": True, "analyses": {analyzer["Name"]: True for analyzer in analyzers}, "diagnosticsDelay": "0s", } server = Server(settings) try: reply = server.request( 1, "initialize", { "processId": None, "rootUri": MODULE.as_uri(), "capabilities": {"workspace": {"configuration": True}}, "initializationOptions": settings, }, ) server.send({"method": "initialized", "params": {}}) for path in paths: document = { "uri": path.as_uri(), "languageId": "go", "version": 1, "text": path.read_text(), } server.send( {"method": "textDocument/didOpen", "params": {"textDocument": document}} ) commands = reply["result"]["capabilities"]["executeCommandProvider"]["commands"] command = next(name for name in commands if name.endswith("diagnose_files")) server.request( 2, "workspace/executeCommand", {"command": command, "arguments": [{"Files": [p.as_uri() for p in paths]}]}, ) server.request(3, "shutdown", None) finally: server.process.terminate() server.process.wait(timeout=10) found = 0 for path in paths: for row in server.diagnostics.get(path.as_uri(), []): line = row["range"]["start"]["line"] + 1 print( f"{path.relative_to(ROOT)}:{line}: [{row.get('source')}] {row['message']}" ) found += 1 print(f"{len(paths)} files, {len(analyzers)} analyzers, {found} diagnostics") return 1 if found else 0
if __name__ == "__main__": sys.exit(main(sys.argv[1:]))