pi_agent_rust 0.8.0

Native AI coding agent CLI - Rust port of Pi Agent
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104

"""Framed subprocess peer for active workspace diagnostics."""
import json
import pathlib
import sys

root = pathlib.Path.cwd()
mode = sys.argv[1]
documents = {}


def send(message):
    body = json.dumps(message).encode("utf-8")
    sys.stdout.buffer.write(
        ("Content-Length: %d\r\n\r\n" % len(body)).encode("ascii") + body
    )
    sys.stdout.buffer.flush()


def items(uri):
    text = documents[uri]["text"]
    if "broken" not in text:
        return []
    message = "broken declaration"
    if mode == "oversize":
        message = "x" * (220 * 1024)
    return [{
        "range": {"start": {"line": 0, "character": 0},
                  "end": {"line": 0, "character": 1}},
        "severity": 1, "message": message,
    }]


while True:
    headers = {}
    while True:
        line = sys.stdin.buffer.readline()
        if not line:
            sys.exit(0)
        if line in (b"\r\n", b"\n"):
            break
        key, value = line.decode("ascii").split(":", 1)
        headers[key.lower()] = value.strip()
    message = json.loads(sys.stdin.buffer.read(int(headers["content-length"])))
    method = message.get("method")
    params = message.get("params") or {}
    with (root / "requests.jsonl").open("a", encoding="utf-8") as log:
        log.write(json.dumps({"method": method, "params": params}) + "\n")
    if method is None:
        if message.get("id") == "unsolicited-edit":
            (root / "edit-response.json").write_text(json.dumps(message["result"]), encoding="utf-8")
        continue
    if method == "exit":
        sys.exit(0)
    if method == "initialized":
        send({"jsonrpc": "2.0", "method": "experimental/serverStatus",
              "params": {"quiescent": True}})
    if method == "textDocument/didOpen":
        doc = params["textDocument"]
        documents[doc["uri"]] = doc.copy()
    elif method == "textDocument/didChange":
        doc = params["textDocument"]
        documents[doc["uri"]]["version"] = doc["version"]
        documents[doc["uri"]]["text"] = params["contentChanges"][-1]["text"]
    if method in ("textDocument/didOpen", "textDocument/didChange") and mode == "push":
        doc = params["textDocument"]
        send({"jsonrpc": "2.0", "method": "textDocument/publishDiagnostics",
              "params": {"uri": doc["uri"], "version": doc["version"],
                         "diagnostics": items(doc["uri"])}})
        send({"jsonrpc": "2.0", "method": "experimental/serverStatus",
              "params": {"quiescent": True}})
    if "id" not in message:
        continue
    result = None
    if method == "initialize":
        capabilities = {"textDocumentSync": {"openClose": True, "change": 1, "save": True}}
        if mode not in ("push", "pending"):
            capabilities["diagnosticProvider"] = {
                "identifier": "scan", "interFileDependencies": True,
                "workspaceDiagnostics": False,
            }
        result = {"capabilities": capabilities}
    elif method == "textDocument/diagnostic":
        uri = params["textDocument"]["uri"]
        if mode == "unsolicited" and uri.endswith("/a.scan"):
            send({"jsonrpc": "2.0", "id": "unsolicited-edit", "method": "workspace/applyEdit",
                  "params": {"edit": {"changes": {uri: [{
                      "range": {"start": {"line": 0, "character": 0},
                                "end": {"line": 0, "character": 6}},
                      "newText": "changed",
                  }]}}}})
        if mode == "hang":
            continue
        if mode == "error" and uri.endswith("/a.scan"):
            send({"jsonrpc": "2.0", "id": message["id"],
                  "error": {"code": -32603, "message": "fixture diagnostic failure"}})
            continue
        if mode == "drift" and uri.endswith("/b.scan"):
            (root / "a.scan").write_text("external change\n", encoding="utf-8")
        result = {"kind": "full", "resultId": uri + ":" + str(documents[uri]["version"]),
                  "items": items(uri)}
    if method == "textDocument/diagnostic":
        send({"jsonrpc": "2.0", "method": "experimental/serverStatus",
              "params": {"quiescent": True}})
    send({"jsonrpc": "2.0", "id": message["id"], "result": result})