Repository navigation
Expand file tree
/
Copy pathhost.js
More file actions
2856 lines (2685 loc) · 116 KB
/
Copy pathhost.js
File metadata and controls
2856 lines (2685 loc) · 116 KB
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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
// magpie's plugin host: runs OpenCode (v1) server plugins under Bun and
// answers magpie over stdin/stdout, one JSON message a line.
//
// magpie → host: {id, method, params}; the host answers {id, result} or
// {id, error}. A fetch streams its reply first: {id, event: "head", status,
// headers}, then {id, event: "chunk", data} (base64) as the body comes, then
// {id, result: null}. {method: "abort", params: {id}} cancels one, and
// {method: "credit", params: {id, n}} gives a fetch back the n charged bytes
// its reader took (the payload it read, and a frame's overhead once the frame
// is done), so it may send that much more. A fetch starts with the window
// magpie gave it and sends no more than that ahead of its reader, so a slow
// reader pauses its upstream instead of the host buffering without bound.
// host → magpie, unasked: {event: "log", level, message} and {event:
// "toast", ...} (what a plugin logs or shows), {event: "auth", provider}
// (a sign-in the host saved or refreshed).
//
// The plugins get what OpenCode hands them (PluginInput): a client whose
// auth.set saves to magpie's plugin-auth.json in OpenCode's auth.json shape,
// tui.showToast and app.log, config.get; Bun's $; the folder magpie keeps
// its files in. What else a plugin asks the client for answers {data:
// undefined}.
//
// A provider may be signed in to more than once: the first account is kept
// under the provider's id, as OpenCode keeps it, the others under
// id#<slot>. Everything a request does runs in its account's scope, so the
// plugin's getAuth, client.auth.set and loader see that account alone.
import { AsyncLocalStorage } from "node:async_hooks"
import { createHash, randomUUID } from "node:crypto"
import dns from "node:dns"
import fs from "node:fs"
import http2 from "node:http2"
import net from "node:net"
import path from "node:path"
import readline from "node:readline"
import { Duplex } from "node:stream"
import tls from "node:tls"
import { pathToFileURL } from "node:url"
const rpcWrite = process.stdout.write.bind(process.stdout)
// Everything the child writes goes through one ordered queue: a line goes out
// only when the stream can take it, so a parent that stops reading bounds what
// Bun buffers instead of it growing without bound. A caller may cancel its own
// wait, which drops its line if it has not been written yet, so an aborted
// request leaves no listener or closure behind. A write failure or a closed
// stream fails every line still waiting rather than leaving them pending.
const outq = [] // {line, settle}
let pumping = false
let outFail = null
function waitDrain() {
return new Promise((resolve, reject) => {
const done = () => { clear(); resolve() }
const bad = (e) => { clear(); reject(e ?? new Error("the host's stdout closed")) }
const clear = () => {
process.stdout.removeListener("drain", done)
process.stdout.removeListener("error", bad)
process.stdout.removeListener("close", bad)
}
process.stdout.once("drain", done)
process.stdout.once("error", bad)
process.stdout.once("close", bad)
})
}
function pump() {
if (pumping) return
pumping = true
while (outq.length) {
const item = outq[0]
if (outFail) {
outq.shift()
item.settle(outFail)
continue
}
// write returns false when the stream is full, but the line was still
// taken: wait for it to drain before the next one, never write it twice
const ok = rpcWrite(item.line)
outq.shift()
item.settle(null)
if (!ok) {
waitDrain().then(
() => { pumping = false; pump() },
(e) => { outFail = e; pumping = false; pump() },
)
return
}
}
pumping = false
}
// queueLine queues one line, giving a promise that settles with null once it is
// written or with the error that stopped it, and a way to give the wait up.
function queueLine(line) {
const item = { line, settle: () => {} }
const promise = new Promise((resolve) => { item.settle = (err) => resolve(err ?? null) })
outq.push(item)
pump()
return {
promise,
cancel: () => {
const i = outq.indexOf(item)
if (i >= 0) {
outq.splice(i, 1)
item.settle(null)
}
},
}
}
// send queues one message, ignoring whether it was written.
const send = (msg) => queueLine(JSON.stringify(msg) + "\n").promise
// sendCancelable is send, with a way to give the wait up. A string is a
// line made already (chunkLine).
const sendCancelable = (msg) => queueLine(typeof msg === "string" ? msg : JSON.stringify(msg) + "\n")
// chunkLine is JSON.stringify({id, event: "chunk", data}) + "\n", made by
// joining: base64 has nothing JSON escapes, and stringify scanning every
// byte of it for something to escape cost more than the base64 itself.
const chunkLine = (id, data) => '{"id":' + JSON.stringify(id) + ',"event":"chunk","data":"' + data + '"}\n'
// A host cancellation suppresses the answer; a local abort still owes Go
// its error. The record, rather than the signal, owns pending wire output.
function transitionFetch(record, state) {
if (record.state === "finished") return
if (record.state !== "cancelled") record.state = state
if (state === "streaming") return
record.gate.stop()
const body = record.unconsumedBody
record.unconsumedBody = undefined
try { Promise.resolve(body?.cancel?.()).catch(() => {}) } catch {}
if (state === "finishing") return
record.pendingSend?.cancel()
record.readyCleanup?.()
record.readyCleanup = null
if (state === "cancelled") record.controller.abort()
else record.state = "finished"
}
function checkFetch(record) {
if (record.state === "cancelled" || record.state === "finished") throw new DOMException("Aborted", "AbortError")
record.controller.signal.throwIfAborted()
}
async function sendForFetch(record, msg) {
if (record.state === "cancelled" || record.state === "finished") return
const ctl = record.controller
const pendingSend = sendCancelable(msg)
record.pendingSend = pendingSend
const onAbort = () => pendingSend.cancel()
ctl.signal.addEventListener("abort", onAbort, { once: true })
try {
const err = await pendingSend.promise
if (err) throw err
} finally {
record.pendingSend = null
ctl.signal.removeEventListener("abort", onAbort)
}
}
// The largest decoded bytes one frame carries: the reader credits in larger
// steps than this, so frames stay small enough that a credit round trip is
// cheap without one frame dwarfing the window.
const MAX_FRAME = 64 << 10
// FRAME_OVERHEAD is what one queued frame costs the window besides its bytes,
// as Go counts it too: the message struct and the decoded slice behind it. It
// keeps a producer of tiny frames from filling a window with structs.
const FRAME_OVERHEAD = 2048
// HEAD_MAX and HEAD_ENTRIES are magpie's own head-envelope policy, not HTTP's:
// a fetch's head carries at most HEAD_MAX header bytes (the UTF-8 bytes of
// every name and value, plus HEAD_ENTRY_COST an entry) and HEAD_ENTRIES
// entries, counted the same by Go before it is queued, so a vendor's unbounded
// header set is not a free 0-charge line.
const HEAD_MAX = 64 << 10
const HEAD_ENTRIES = 256
const HEAD_ENTRY_COST = 4
// headCost is a head's header cost, as Go counts it too: the UTF-8 bytes of
// every name and value, plus HEAD_ENTRY_COST an entry.
function headCost(h) {
let n = 0
for (const [k, v] of Object.entries(h)) n += Buffer.byteLength(k) + Buffer.byteLength(v) + HEAD_ENTRY_COST
const entries = Object.keys(h).length
return { n, entries }
}
// makeGate is one fetch's credit: the child may have `window` decoded bytes
// outstanding, each frame costing its bytes plus FRAME_OVERHEAD. take returns
// how many bytes of a frame may go now (-1 when the fetch was stopped), spend
// takes the frame's cost, add is the reader giving credit back.
function makeGate(window) {
let free = window
let stopped = false
let wake = null
const fit = (want) => Math.min(want, free - FRAME_OVERHEAD)
return {
take(want) {
if (stopped) return Promise.resolve(-1)
const n = fit(want)
if (n > 0) return Promise.resolve(n)
return new Promise((resolve) => { wake = (ok) => resolve(ok ? fit(want) : -1) })
},
spend(cost) { free -= cost },
add(n) {
free += n
if (wake && free > FRAME_OVERHEAD) { const w = wake; wake = null; w(true) }
},
stop() {
stopped = true
if (wake) { const w = wake; wake = null; w(false) }
},
}
}
// A plugin writing to stdout would break the protocol: everything it
// prints goes to stderr, which magpie logs.
const toErr = (...xs) => process.stderr.write(xs.map((x) => (typeof x === "string" ? x : Bun.inspect(x))).join(" ") + "\n")
console.log = console.info = console.debug = console.warn = toErr
process.stdout.write = (chunk, enc, cb) => process.stderr.write(chunk, enc, cb)
// Nor may a program a plugin starts have them: one given the host's stdin
// reads magpie's messages away (pi-devin-plus runs `devin auth login` with
// stdio "inherit", which took the next requests as its answer and gave up,
// "Login canceled"), and its stdout is the host's answers. It reads
// nothing instead and writes to stderr. node:child_process starts every
// program through Bun.spawn and Bun.spawnSync, so these see all of them.
//
// And it has the proxy the plugin's own fetches take, as a built-in's CLI
// had magpie's (netproxy.Env): the host has its *_PROXY only as
// MAGPIE_*_PROXY (see below), so `grok login`, which the Grok plugin runs,
// went out with none and, where x.ai is reached only through one, printed
// no link to open (𝕏 on Discord). One the plugin set itself is kept.
// Node's fetch goes past *_PROXY unless NODE_USE_ENV_PROXY=1 says to take
// them, so a plugin downloading in a Node program it starts went out with
// none and failed where the vendor is reached only through one (larchsis
// on 𝕏, Windows 11, a proxy set in magpie only). It is told to, unless
// the plugin said otherwise.
const own = (v, fd) => v === "inherit" || v === fd || v === (fd ? process.stdout : process.stdin)
const PROXY_VARS = ["HTTPS_PROXY", "HTTP_PROXY", "ALL_PROXY"]
function proxied(env) {
env = { ...(env ?? process.env) }
if (Object.keys(env).some((k) => PROXY_VARS.includes(k.toUpperCase()) && env[k])) return env
const mine = via.getStore()
if (mine === "direct") return env
const set = mine ? { HTTPS_PROXY: mine, HTTP_PROXY: mine } : { HTTPS_PROXY: globalProxy.https, HTTP_PROXY: globalProxy.http, NO_PROXY: proxyVar("NO_PROXY") }
if (set.HTTPS_PROXY || set.HTTP_PROXY) set.NODE_USE_ENV_PROXY = "1"
for (const [k, v] of Object.entries(set)) {
if (!v || Object.keys(env).some((e) => e.toUpperCase() === k)) continue
env[k] = v
// Windows' names are one whatever their case
if (process.platform !== "win32" && k.endsWith("_PROXY") && k !== "NODE_USE_ENV_PROXY") env[k.toLowerCase()] = v
}
return env
}
const guard = (o) => {
if (o != null && typeof o !== "object") return o
o = { ...o }
o.env = proxied(o.env)
if (Array.isArray(o.stdio)) {
o.stdio = [...o.stdio]
if (own(o.stdio[0], 0)) o.stdio[0] = "ignore"
if (own(o.stdio[1], 1)) o.stdio[1] = 2
}
if (own(o.stdin, 0)) o.stdin = "ignore"
if (own(o.stdout, 1)) o.stdout = 2
return o
}
for (const name of ["spawn", "spawnSync"]) {
const run = Bun[name]
Bun[name] = function (a, b) {
return Array.isArray(a) ? run.call(this, a, guard(b)) : run.call(this, guard(a))
}
}
let authPath = ""
let modelsDevPath = ""
let piPath = "" // pi.js, which loads pi's extensions
let directory = process.cwd()
let userConfig = {}
// prefer is the plugin the user picked for a provider id more than one
// plugin signs in to: id → its spec (plugins.json's prefer)
let prefer = {}
// chosen are the user's own values of the plugins' options (auth.options),
// provider id → key → true/false (settings.json's pluginOptions, #2038)
let chosen = {}
const hooks = [] // {spec, hooks}
const gatewayHooks = [] // {spec, options, onRequest}
const loaded = [] // {spec, id, error}
const loaders = new Map() // account → options the auth loader returned
// infos are info's answers a request is given, by account: built once, not
// for every fetch, which asked the plugin's models hook (its vendor, once
// its own list's time was up) before each request went. One goes with its
// account's sign-in (forgetAccount), and all with the options the user set
// (setProviderOptions) or a new models.dev catalog; a listing (providers)
// puts its own in.
const infos = new Map() // account → {gen, md, info}
let infoGen = 0
// forgetAccount drops what was made of the sign-in at key: its loader's
// options and its info, which the next request makes again.
function forgetAccount(key) {
loaders.delete(key)
infos.delete(key)
infoGen++
}
const sessions = new Map() // oauth sign-in in progress → its authorize result
const inflight = new Map() // fetch id → its lifecycle and owned resources
const renewing = new Map() // account → its sign-in's renewal under way
const unrenewed = new Map() // account → its last renewal that failed: {at, secret, gone}
let config = { provider: {} } // what the plugins' config hooks made of it
// loadWait bounds one plugin's load (its import and its plugin function)
// and one config hook; listWait bounds one provider's models hook in a
// listing. A plugin stuck on its vendor (a sign-in answered 401 and tried
// again, a vendor that never answers) is then given up on alone: the
// plugins load side by side and every other one is listed meanwhile,
// rather than all waiting behind it until magpie's own deadline fails the
// whole host or the whole list. magpie sets both at init.
let loadWait = 20 * 1000
let listWait = 20 * 1000
// late is what a step that took longer than ms gives instead of its own
// result: an error saying so. The step goes on by itself; what it gives
// later is dropped.
class Late extends Error {}
function inTime(step, ms, what) {
let timer
const late = new Promise((_, no) => {
timer = setTimeout(() => no(Object.assign(new Late(`${what} didn't finish in ${Math.round(ms / 1000)}s`), { plain: true })), ms)
})
return Promise.race([Promise.resolve().then(step), late]).finally(() => clearTimeout(timer))
}
// ---- a provider's own proxy ---------------------------------------------------
// A request made for a provider or an account with a proxy of its own
// (#237) goes through it, and so does every fetch the plugin makes for
// it; one for a provider set to "direct" goes through none. Bun reads
// *_PROXY once, and a fetch given no proxy option can't be told to skip
// them, so the host is started with them as MAGPIE_*_PROXY and each
// fetch is given magpie's own here when it has none of its own.
// Loopback never goes through one.
const via = new AsyncLocalStorage() // the proxy: "", "direct" or its URL
const bunFetch = globalThis.fetch
function proxyVar(k) {
return process.env["MAGPIE_" + k] || process.env["MAGPIE_" + k.toLowerCase()] || ""
}
const globalProxy = {
https: proxyVar("HTTPS_PROXY") || proxyVar("ALL_PROXY"),
http: proxyVar("HTTP_PROXY") || proxyVar("ALL_PROXY"),
no: proxyVar("NO_PROXY")
.split(",")
.map((s) => s.trim().toLowerCase().replace(/^\*?\./, ""))
.filter(Boolean),
}
function hostOf(input) {
try {
const u = new URL(typeof input === "string" || input instanceof URL ? input : input.url)
return { protocol: u.protocol, host: u.hostname.toLowerCase().replace(/^\[|\]$/g, "") }
} catch {
return null
}
}
function loopback(host) {
return host === "localhost" || host.endsWith(".localhost") || host === "::1" || /^127\./.test(host)
}
// A host on the user's own network — another computer's magpie, an LLM
// relay on the NAS — is reached directly, whatever proxy the host was
// started with, as magpie's own requests reach it (netproxy.onLAN, lan.go,
// #982, #1572): a proxy is for reaching vendors; one elsewhere can't reach
// the user's network, and one here only goes there directly itself. A
// provider's own proxy (via) is still taken as it is. The rule is lan.go's:
// keep the two the same.
// lanWait is how long a name is looked up for before it is taken as not on
// the network; lanKeep is how long what it resolved to is kept.
const lanWait = 400
const lanKeep = 60_000
const lanSeen = new Map() // host → {lan: Promise<boolean>, at}
// ipBytes is the 16 bytes of the IPv6 address s (an IPv4 one as
// ::ffff:a.b.c.d), null when s is no address.
function ipBytes(s) {
if (net.isIPv4(s)) return [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff, ...s.split(".").map(Number)]
if (!net.isIPv6(s)) return null
s = s.replace(/%.*$/, "")
const v4 = s.match(/(\d+\.\d+\.\d+\.\d+)$/)
if (v4) {
const [a, b, c, d] = v4[1].split(".").map(Number)
s = s.slice(0, -v4[1].length) + ((a << 8) | b).toString(16) + ":" + ((c << 8) | d).toString(16)
}
const [head, tail] = s.includes("::") ? s.split("::") : [s, null]
const words = (x) => (x ? x.split(":").map((w) => parseInt(w, 16)) : [])
const h = words(head)
const t = words(tail)
const all = tail === null ? h : [...h, ...Array(8 - h.length - t.length).fill(0), ...t]
return all.flatMap((w) => [w >> 8, w & 0xff])
}
// lanIP is netproxy.lanIP: a private (10/8, 172.16/12, 192.168/16,
// fc00::/7), link-local (169.254/16, fe80::/10), loopback or carrier-grade
// NAT (Tailscale's 100.64/10) address.
function lanIP(s) {
const b = ipBytes(s)
if (!b) return false
if (b.slice(0, 10).every((x) => x === 0) && b[10] === 0xff && b[11] === 0xff) {
const [p, q] = b.slice(12)
return p === 10 || (p === 172 && (q & 0xf0) === 16) || (p === 192 && q === 168) || (p === 169 && q === 254) || p === 127 || (p === 100 && (q & 0xc0) === 64)
}
return (b[0] & 0xfe) === 0xfc || (b[0] === 0xfe && (b[1] & 0xc0) === 0x80) || b.every((x, i) => x === (i === 15 ? 1 : 0))
}
// onLAN is netproxy.onLAN: whether host is on the user's own network — an
// address lanIP takes, a .local name, or a name every address of which is
// one of those. A name that doesn't resolve here (in lanWait) is not: the
// proxy is what reaches it.
function onLAN(host) {
host = host.toLowerCase().replace(/\.$/, "")
if (!host) return Promise.resolve(false)
if (net.isIP(host)) return Promise.resolve(lanIP(host))
if (host.endsWith(".local")) return Promise.resolve(true)
const seen = lanSeen.get(host)
if (seen && Date.now() - seen.at < lanKeep) return seen.lan
let timer
const lan = Promise.race([
dns.promises.lookup(host, { all: true, verbatim: true }).then(
(as) => as.length > 0 && as.every((a) => lanIP(a.address)),
() => false,
),
new Promise((r) => (timer = setTimeout(() => r(false), lanWait))),
]).finally(() => clearTimeout(timer))
if (lanSeen.size >= 256) lanSeen.clear()
lanSeen.set(host, { lan, at: Date.now() })
return lan
}
// proxyOf is the proxy a request to input goes through, "" for none, and
// whether it is the global one, which a host on the user's own network
// goes around (proxyFor).
function proxyOf(input) {
const u = hostOf(input)
if (!u || loopback(u.host)) return { p: "" }
const own = via.getStore()
if (own === "direct") return { p: "" }
if (own) return { p: own }
if (globalProxy.no.some((n) => n === "*" || u.host === n || u.host.endsWith("." + n))) return { p: "" }
const p = u.protocol === "https:" ? globalProxy.https : u.protocol === "http:" ? globalProxy.http : ""
return { p, global: !!p, host: u.host }
}
// proxyFor is the proxy a fetch of input goes through, "" for none.
async function proxyFor(input) {
const { p, global, host } = proxyOf(input)
return global && (await onLAN(host)) ? "" : p
}
// reach is what a check's fetches came to: whether one got an answer, and
// the first that got none
const reach = new AsyncLocalStorage()
async function sent(input, init) {
if (init && "proxy" in init) return bunFetch(input, init)
const p = await proxyFor(input)
if (!p) return bunFetch(input, init)
return bunFetch(input, { ...init, proxy: p }).catch(async (e) => {
if (e?.name === "AbortError" || init?.signal?.aborted) throw e
const why = await proxyRefuses(p, input)
if (why) throw Object.assign(new Error(`proxyconnect tcp: ${why}`), { code: e?.code, cause: e })
throw e
})
}
// proxyRefuses is why the proxy p won't carry a request to input, "" when
// it will: Bun says a proxy that is down, or that turns the tunnel away,
// only as a connection that failed, which a vendor's own failure is too.
// The tunnel is asked for again, as Go's transport asks for it, so a
// failure is said in its words ("proxyconnect …") and the gateway takes it
// for the proxy's, asking the next account rather than resting this one.
function proxyRefuses(p, input) {
let pu, target
try {
pu = new URL(p)
const u = new URL(typeof input === "string" || input instanceof URL ? input : input.url)
target = `${u.hostname.includes(":") ? `[${u.hostname.replace(/^\[|\]$/g, "")}]` : u.hostname}:${u.port || (u.protocol === "http:" ? 80 : 443)}`
} catch {
return Promise.resolve("")
}
const host = pu.hostname.replace(/^\[|\]$/g, "")
const port = Number(pu.port || (pu.protocol === "https:" ? 443 : 80))
return new Promise((resolve) => {
let got = ""
let refused
const done = (why) => {
sock.destroy()
resolve(why)
}
const opts = { host, port, servername: net.isIP(host) ? undefined : host }
const sock = pu.protocol === "https:" ? tls.connect(opts) : net.connect(opts)
sock.setTimeout(10_000, () => done(`dial tcp ${host}:${port}: i/o timeout`))
sock.on("error", (e) => done(`dial tcp ${host}:${port}: ${e?.message ?? e}`))
sock.once(pu.protocol === "https:" ? "secureConnect" : "connect", () => {
let req = `CONNECT ${target} HTTP/1.1\r\nHost: ${target}\r\n`
if (pu.username) {
const cred = `${decodeURIComponent(pu.username)}:${decodeURIComponent(pu.password)}`
req += `Proxy-Authorization: Basic ${Buffer.from(cred).toString("base64")}\r\n`
}
sock.write(req + "\r\n")
})
sock.on("data", (d) => {
got += d.toString("latin1")
const end = got.indexOf("\r\n\r\n")
if (end < 0 && got.length < 8192) return
const status = got.slice(0, got.indexOf("\r\n")).replace(/^HTTP\/\d(\.\d)?\s+/, "")
if (/^2\d\d/.test(status)) return done("")
// a bridge to a SOCKS proxy says why after its status
refused ??= () => {
const body = end < 0 ? "" : got.slice(end + 4, end + 304).trim()
return body ? `${status}: ${body}` : status
}
setTimeout(() => done(refused()), 50)
})
sock.on("end", () => done(refused ? refused() : got ? "" : `${host}:${port} closed the connection`))
})
}
// listing is what a models hook's fetches came to: whether it asked any,
// and whether the last it asked answered
const listing = new AsyncLocalStorage()
function asked(input, init) {
const l = listing.getStore()
if (!l) return sent(input, init)
l.tried = true
return sent(input, init).then(
(res) => ((l.lastOk = res.ok), (l.said = res.ok ? "" : `HTTP ${res.status}`), res),
(e) => {
l.lastOk = false
l.said = e?.message ?? String(e)
throw e
},
)
}
// MAGPIE_ONLY are the headers magpie tells a plugin's fetch where the agent
// that asked works with (WorkDirHeader and WorkGitHeader in
// internal/provider): for the plugin to read, never for its vendor, so a
// plugin that sends on the headers it was given sends them nowhere.
const MAGPIE_ONLY = ["x-magpie-work-dir", "x-magpie-work-git"]
function unsaid(input, init) {
if (input instanceof Request && MAGPIE_ONLY.some((k) => input.headers.has(k))) {
const h = new Headers(input.headers)
for (const k of MAGPIE_ONLY) h.delete(k)
input = new Request(input, { headers: h })
}
if (init?.headers) {
const h = new Headers(init.headers)
if (MAGPIE_ONLY.some((k) => h.has(k))) {
for (const k of MAGPIE_ONLY) h.delete(k)
init = { ...init, headers: h }
}
}
return [input, init]
}
globalThis.fetch = Object.assign(function fetch(input, init) {
;[input, init] = unsaid(input, init)
const r = reach.getStore()
if (!r) return asked(input, init)
return asked(input, init).then(
(res) => ((r.reached = true), res),
(e) => {
r.failed ??= e
throw e
},
)
}, bunFetch)
// An HTTP/2 session a plugin opens itself goes through the proxy its
// fetches take, as a built-in's requests on Go's transport did. Cursor's
// plugin runs agent.v1.AgentService/Run on node:http2, which Bun connects
// directly, *_PROXY or not: where Cursor is reached only through magpie's
// proxy (the system's, or one set in magpie), its sign-in, models and
// usage came through and every chat went out without it (#1035).
const h2connect = http2.connect
http2.connect = function connect(authority, options, listener) {
if (typeof options === "function") [listener, options] = [options, undefined]
let u
try {
u = typeof authority === "string" ? new URL(authority) : authority instanceof URL ? authority : new URL(`${authority.protocol ?? "https:"}//${authority.host ?? authority.hostname}`)
} catch {
return h2connect.call(this, authority, options, listener)
}
const { p, global } = options?.createConnection ? { p: "" } : proxyOf(u.href)
if (!p) return h2connect.call(this, authority, options, listener)
const host = u.hostname.replace(/^\[|\]$/g, "")
const port = Number(u.port || (u.protocol === "http:" ? 80 : 443))
// the session is returned now, and its connection made once it is known
// whether host is on the user's own network (direct) or not (the proxy)
const raw = tunnel(global ? onLAN(host).then((lan) => (lan ? "" : p)) : p, host, port)
const createConnection = () =>
u.protocol === "http:" ? raw : tls.connect({ ...options, socket: raw, servername: options?.servername || (net.isIP(host) ? undefined : host), ALPNProtocols: ["h2"] })
const session = h2connect.call(this, authority, { ...options, createConnection }, listener)
// the tunnel's failure is the session's: Bun's TLS over a stream leaves
// the stream's errors unheard (one unheard ends the host), and says a
// tunnel that never came as "h2 is not supported"
raw.on("error", (e) => session.destroy(e))
return session
}
// tunnel is a connection to host:port through the HTTP proxy which names
// (a proxy, or a promise of one), or straight to it when that is "" (a host
// on the user's own network, onLAN). A SOCKS proxy is handed to the host as
// an HTTP bridge to it, netproxy.ForBun: its CONNECT is asked first, and
// what is written waits for the tunnel. A proxy that turns it away fails
// it in Go's words ("proxyconnect …").
function tunnel(which, host, port) {
const target = `${host.includes(":") ? `[${host}]` : host}:${port}`
let ready = false
let sock = null
let ended = false
let gone = false
const queued = []
const conn = new Duplex({
read() {},
write(chunk, enc, cb) {
if (ready) sock.write(chunk, cb)
else queued.push([chunk, cb])
},
final(cb) {
if (ready) sock.end()
else ended = true
cb()
},
destroy(err, cb) {
gone = true
sock?.destroy()
cb(err)
},
})
const open = () => {
ready = true
sock.on("data", (d) => conn.push(d))
for (const [c, cb] of queued.splice(0)) sock.write(c, cb)
if (ended) sock.end()
}
Promise.resolve(which).then((p) => {
if (gone) return
if (!p) {
sock = net.connect({ host, port })
sock.once("connect", open)
sock.on("end", () => conn.push(null))
sock.on("error", (e) => conn.destroy(e))
return
}
const pu = new URL(p)
const ph = pu.hostname.replace(/^\[|\]$/g, "")
const pp = Number(pu.port || (pu.protocol === "https:" ? 443 : 80))
sock = pu.protocol === "https:" ? tls.connect({ host: ph, port: pp, servername: net.isIP(ph) ? undefined : ph }) : net.connect({ host: ph, port: pp })
sock.once(pu.protocol === "https:" ? "secureConnect" : "connect", () => {
let req = `CONNECT ${target} HTTP/1.1\r\nHost: ${target}\r\n`
if (pu.username) {
const cred = `${decodeURIComponent(pu.username)}:${decodeURIComponent(pu.password)}`
req += `Proxy-Authorization: Basic ${Buffer.from(cred).toString("base64")}\r\n`
}
sock.write(req + "\r\n")
})
let got = Buffer.alloc(0)
const head = (chunk) => {
got = Buffer.concat([got, chunk])
const end = got.indexOf("\r\n\r\n")
if (end < 0) {
if (got.length > 8192) conn.destroy(new Error(`proxyconnect tcp: ${ph}:${pp} answered CONNECT with no HTTP`))
return
}
sock.off("data", head)
const status = got.subarray(0, got.indexOf("\r\n")).toString("latin1").replace(/^HTTP\/\d(\.\d)?\s+/, "")
if (!/^2\d\d/.test(status)) {
const body = got.subarray(end + 4, end + 304).toString("utf8").trim()
return conn.destroy(new Error(`proxyconnect tcp: ${body ? `${status}: ${body}` : status}`))
}
if (got.length > end + 4) conn.push(got.subarray(end + 4))
open()
}
sock.on("data", head)
sock.on("end", () => conn.push(null))
sock.on("error", (e) => conn.destroy(ready ? e : new Error(`proxyconnect tcp: dial tcp ${ph}:${pp}: ${e?.message || e?.code || e}`)))
}).catch((e) => conn.destroy(e))
return conn
}
// ---- auth.json ---------------------------------------------------------------
// A crash can leave plugin-auth.json at its full length with every byte
// zero (#1505). Such a file is unknown, never "no sign-ins": it is read
// from its last good generation, plugin-auth.json.bak, which each write
// keeps first; a write over one that doesn't read copies it aside
// (plugin-auth.json.bad-<time>) first, and never writes none over it. As
// internal/lastgood does for magpie's own files.
// parseAuth is the sign-ins in text, null when it isn't one JSON object.
function parseAuth(text) {
try {
const v = JSON.parse(text)
return v && typeof v === "object" && !Array.isArray(v) ? v : null
} catch {
return null
}
}
function readFileSteady(p) {
for (let i = 0; ; i++) {
try {
return fs.readFileSync(p)
} catch (e) {
if (e?.code === "ENOENT") return null
// Windows refuses a file being renamed over for a moment: read as
// none, the next setAuth would write the others' accounts away
if (i >= 50) throw e
pause(10)
}
}
}
// describeBad says what is wrong with a file that doesn't read.
function describeBad(buf) {
if (buf.length > 0 && buf.every((b) => b === 0)) return `${buf.length} bytes, every one zero`
if (buf.toString("utf8").trim() === "") return "it is empty"
return "it doesn't parse"
}
let authRecovered = false
let authUnread = false
// readAuthState is the sign-ins and whether they are known: a file not
// there is none, known; one that doesn't read is its .bak's, known, or
// none, unknown.
function readAuthState(buf = readFileSteady(authPath)) {
if (buf === null) return { all: {}, known: true }
const v = parseAuth(buf.toString("utf8"))
if (v) return { all: v, known: true, own: true }
const why = describeBad(buf)
let bak = null
try {
bak = parseAuth(fs.readFileSync(authPath + ".bak", "utf8"))
} catch {}
if (bak) {
if (!authRecovered) {
authRecovered = true
toErr(`plugin-auth.json can't be read (${why}); the sign-ins in plugin-auth.json.bak are used`)
send({ event: "recovered", said: why })
}
return { all: bak, known: true }
}
if (!authUnread) {
authUnread = true
toErr(`plugin-auth.json can't be read (${why}) and has no good backup`)
}
return { all: {}, known: false }
}
// authSeen is plugin-auth.json as last parsed, kept while the file has the
// same time, size and inode: every request read and parsed it two or three
// times (its account, its renewal, the plugin's getAuth), ~0.6 ms each for
// a dozen sign-ins. While the file's time is too recent for a second write
// to stamp a later one (Linux's clock is a tick coarse), its bytes are kept
// and compared too, as Go's filememo does.
let authSeen = null // {mtimeMs, size, ino, raw, all}
const AUTH_SETTLED = 2000
// readAuth is the sign-ins, shared: the caller must not change them, and
// hands a plugin a copy (authCopy).
function readAuth() {
let st = null
try {
st = fs.statSync(authPath)
} catch {}
const e = authSeen
const same = st && e && e.mtimeMs === st.mtimeMs && e.size === st.size && e.ino === st.ino
if (same && !e.raw) return e.all
const buf = readFileSteady(authPath)
if (same && buf && buf.equals(e.raw)) {
if (Date.now() - e.mtimeMs > AUTH_SETTLED) e.raw = null
return e.all
}
const s = readAuthState(buf)
authSeen = st && s.own ? { mtimeMs: st.mtimeMs, size: st.size, ino: st.ino, raw: Date.now() - st.mtimeMs > AUTH_SETTLED ? null : buf, all: s.all } : null
return s.all
}
// authCopy is readAuth's sign-ins, or the one at key, the plugin's own to
// change.
function authCopy(key) {
const all = readAuth()
const v = key === undefined ? all : all[key]
return v === undefined ? undefined : structuredClone(v)
}
// writeDurable replaces p with data through a temp file flushed to the
// disk before it is renamed over p: a rename reaches the disk on its own
// schedule, the data behind it later, and a machine that went down between
// the two left plugin-auth.json all zero (#1505). A write or flush that
// fails leaves p as it was.
function writeDurable(p, data) {
const tmp = p + ".tmp-" + process.pid
try {
const f = fs.openSync(tmp, "w", 0o600)
try {
for (let off = 0; off < data.length; ) off += fs.writeSync(f, data, off, data.length - off)
fs.fsyncSync(f)
} finally {
fs.closeSync(f)
}
// and a file held open by a reader refuses to be renamed over
for (let i = 0; ; i++) {
try {
return fs.renameSync(tmp, p)
} catch (e) {
if (i >= 50 || !["EPERM", "EACCES", "EBUSY"].includes(e?.code)) throw e
pause(10)
}
}
} catch (e) {
try {
fs.unlinkSync(tmp)
} catch {}
throw e
}
}
// badStamp is the time a bad file is kept under, as Go's lastgood names it.
function badStamp(d = new Date()) {
const p = (n) => String(n).padStart(2, "0")
return `${d.getFullYear()}${p(d.getMonth() + 1)}${p(d.getDate())}-${p(d.getHours())}${p(d.getMinutes())}${p(d.getSeconds())}`
}
// keepBad copies buf, a plugin-auth.json that doesn't read, aside: two
// within one second are both kept (-2, …), the same bytes kept already
// not again.
function keepBad(buf) {
const stamp = authPath + ".bad-" + badStamp()
for (let i = 1; i < 1000; i++) {
const name = i > 1 ? `${stamp}-${i}` : stamp
let f
try {
f = fs.openSync(name, "wx", 0o600)
} catch (e) {
if (e?.code !== "EEXIST") throw e
let old = null
try {
old = fs.readFileSync(name)
} catch {}
if (old && old.equals(buf)) return name
continue
}
try {
for (let off = 0; off < buf.length; ) off += fs.writeSync(f, buf, off, buf.length - off)
fs.fsyncSync(f)
} catch (e) {
fs.closeSync(f)
try {
fs.unlinkSync(name)
} catch {}
throw e
}
fs.closeSync(f)
toErr(`plugin-auth.json can't be read (${describeBad(buf)}); it is kept as ${path.basename(name)}`)
return name
}
throw new Error("plugin-auth.json can't be read and there is no free name to keep it under")
}
// keepAuth is called before plugin-auth.json is replaced: the file there
// now becomes plugin-auth.json.bak when it reads, or is copied aside when
// it doesn't. A failure stops the write.
function keepAuth() {
const buf = readFileSteady(authPath)
if (buf === null) return
if (!parseAuth(buf.toString("utf8"))) {
keepBad(buf)
return
}
const bak = authPath + ".bak"
let old = null
try {
old = fs.readFileSync(bak)
} catch {}
if (!old || !old.equals(buf)) writeDurable(bak, buf)
}
function writeAuth(all) {
fs.mkdirSync(path.dirname(authPath), { recursive: true })
keepAuth()
writeDurable(authPath, Buffer.from(JSON.stringify(all, null, 2) + "\n"))
}
function pause(ms) {
Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, ms)
}
// Two hosts run at once while one is restarted (the old one finishing its
// calls), and magpie writes the file when no host runs: each change to it
// is made under plugin-auth.json.lock, read afresh and written back with
// only its own accounts changed, so a host never writes another's newer
// token away with what it read before. Nothing awaits under the lock; one
// held longer than AUTH_LOCK_STALE was left by a host that died.
const AUTH_LOCK_STALE = 10 * 1000
function lockAuth() {
const lock = authPath + ".lock"
fs.mkdirSync(path.dirname(authPath), { recursive: true })
for (const start = Date.now(); ; ) {
try {
fs.closeSync(fs.openSync(lock, "wx", 0o600))
return () => {
try {
fs.unlinkSync(lock)
} catch {}
}
} catch (e) {
if (e?.code !== "EEXIST" && e?.code !== "EPERM" && e?.code !== "EACCES") throw e
}
let old = false
try {
old = Date.now() - fs.statSync(lock).mtimeMs > AUTH_LOCK_STALE
} catch {}
if (old || Date.now() - start > 2 * AUTH_LOCK_STALE) {
try {
fs.unlinkSync(lock)
} catch {}
}
pause(5)
}
}
// changeAuth runs change on the file as it is now, under the lock, and
// writes it back when change says it changed it.
function changeAuth(change) {
const unlock = lockAuth()
try {
const { all, known } = readAuthState()
const out = change(all)
// sign-ins that don't read are never written over with none: a
// sign-out of one of them waits for the file to be put right
if (out !== false && (known || Object.keys(all).length > 0)) writeAuth(all)
return out
} finally {
unlock()
}
}
function setAuth(key, info) {
changeAuth((all) => {
all[key] = info
})
forgetAccount(key)
send({ event: "auth", provider: providerOf(key), account: key })
}
function removeAuth(key) {
changeAuth((all) => {
delete all[key]
})
forgetAccount(key)
send({ event: "auth", provider: providerOf(key), account: key })
}
// ---- accounts ----------------------------------------------------------------
// scope is the account a request is for: {provider, key}.
const scope = new AsyncLocalStorage()
const providerOf = (key) => key.split("#")[0]
// accountsOf are the keys provider's accounts are kept under, the one under
// its own id first.
function accountsOf(all, provider) {
return Object.keys(all)
.filter((k) => k === provider || k.startsWith(provider + "#"))
.sort((a, b) => (a === provider ? -1 : b === provider ? 1 : a < b ? -1 : a > b ? 1 : 0))
}
// keyFor is where the plugin's id is kept in this scope: the scope's
// account when the plugin names the scope's provider, else its own id.
function keyFor(id) {
const s = scope.getStore()
return s && s.provider === id ? s.key : id
}
// accountKey is the account a request names, the provider's first when it
// names none.