antcolony

17.3 KB

1. // agent.hl — `agent`: the program on ONE host that starts the workers (mission 035, ticket antcolony#14; concept
2. // docs/scheduler-agent.md §1 + §3). The scheduler (`run --serve`) decides WHAT runs; this agent runs it here, on the host
3. // that holds the project's dev folder, and reports back. It holds no data of its own: tickets stay the only state — a
4. // worker's lease, heartbeat and report go to tickets directly (work.hl), the agent only keeps in-flight session bookkeeping.
5. //
6. // Connects OUT: every --poll s one `POST <scheduler>/agent/sync` (HTTPS, `Authorization: Bearer <agent token>`):
7. // → { host, capabilities: [what this computer offers, e.g. chrome, gpu, deploy], parallel, accepting, why, running: [{ref, session, folder, phase, ports}], events: [buffered, not yet acked] }
8. // ← { ack: [event ids], commands: [{kind: start|resume, project, number, session, folder, brief}] }
9. // Hybriel has no WebSocket client (hybriel#109), so this is request/answer instead of one held-open socket; the effects the
10. // concept asks for are the same: the agent reconnects on its own, sessions keep running while the scheduler is not
11. // reachable, `SESSION END` events are buffered (until acked) and delivered on reconnect, a resent command is ignored.
12. // Quota is the agent's (concept §3): `claude -p /usage` is read here (at most every --quota-every s, only while a slot is
13. // free); at/above the reserve, above the weekly limit, or after a usage limit was hit → `accepting: false` + why, and a
14. // command that arrives anyway is refused (event `refused`). Ports: the agent's own pool, one disjoint range per session.
15. // Stop: the `colony` wrapper writes $COLONY_STOP_FILE: first signal → accept nothing, running sessions finish;
16. // second → they are parked (resumed by the next agent / scheduler round). Never silent: every line starts with AGENT / SESSION END / LIMIT.
17. import { env } from 'hl:proc'
18. import { exists, readFile, writeFile, mkDir } from 'hl:fs'
19. import { fetch } from 'hl:fetch'
20. import { every, now } from 'hl:time'
21. import { readToken, readAgentToken, userAgent } from './tickets.hl'
22. import { runWork, runResume, hostName, runsDir } from './work.hl'
23. import { setting, isWholeNumber } from './claude.hl'
24. import { parseCapabilities } from './util.hl'
25. import { probeUsage } from './quota.hl'
26. import { portRange } from './brief.hl'
27. import { passOptions, money, ago } from './daemon.hl'
29. static LIMIT_RETRY_MS = 900000
31. static runAgent = (argv) => {
32. let home = env('COLONY_HOME')
33. if (home == null || home == '') {
34. console.log('agent: REFUSED — COLONY_HOME is not set (run through ./colony, which sets it)')
35. return false
36. }
37. let schedText = setting(argv, '--scheduler', 'COLONY_SCHEDULER_URL', '')
38. if (schedText == '') {
39. console.log('agent: REFUSED — --scheduler URL (COLONY_SCHEDULER_URL) is not set: where is the scheduler?')
40. return false
41. }
42. while (schedText.endsWith('/')) { schedText = schedText.slice(0, schedText.length - 1) }
43. let pollText = setting(argv, '--poll', 'COLONY_AGENT_POLL', '5')
44. let parallelText = setting(argv, '--parallel', 'COLONY_PARALLEL', '1')
45. let reserveText = setting(argv, '--reserve', 'COLONY_QUOTA_RESERVE', '70')
46. let weekText = setting(argv, '--week-limit', 'COLONY_WEEK_LIMIT', '84')
47. let poolText = setting(argv, '--port-pool', 'COLONY_PORT_POOL', '8700-8799')
48. let perText = setting(argv, '--ports-per-session', 'COLONY_PORTS_PER_SESSION', '10')
49. let everyText = setting(argv, '--quota-every', 'COLONY_AGENT_QUOTA_EVERY', '60')
50. let capText = setting(argv, '--capabilities', 'COLONY_CAPABILITIES', '')
51. let accountText = setting(argv, '--account', 'COLONY_CLAUDE_ACCOUNT', '')
52. let claudeBin = setting(argv, '--claude', 'COLONY_CLAUDE', 'claude')
53. let quotaBin = setting(argv, '--quota-claude', 'COLONY_QUOTA_CLAUDE', claudeBin)
54. if (!isWholeNumber(pollText) || pollText == '0') {
55. console.log('agent: REFUSED — --poll must be whole seconds ≥ 1, got ' + pollText)
56. return false
57. }
58. if (!isWholeNumber(parallelText) || parallelText == '0') {
59. console.log('agent: REFUSED — --parallel must be a whole number ≥ 1, got ' + parallelText)
60. return false
61. }
62. if (!isWholeNumber(reserveText) || toNumber(reserveText) > 100) {
63. console.log('agent: REFUSED — --reserve must be a percentage 0–100, got ' + reserveText)
64. return false
65. }
66. if (!isWholeNumber(weekText) || toNumber(weekText) > 100) {
67. console.log('agent: REFUSED — --week-limit must be a percentage 0–100, got ' + weekText)
68. return false
69. }
70. if (!isWholeNumber(everyText) || everyText == '0') {
71. console.log('agent: REFUSED — --quota-every must be whole seconds ≥ 1, got ' + everyText)
72. return false
73. }
74. let pool = portRange(poolText)
75. if (pool == null) {
76. console.log('agent: REFUSED — --port-pool must be A-B (1 ≤ A ≤ B ≤ 65535), got ' + poolText)
77. return false
78. }
79. if (!isWholeNumber(perText) || perText == '0') {
80. console.log('agent: REFUSED — --ports-per-session must be a whole number ≥ 1, got ' + perText)
81. return false
82. }
83. let caps = parseCapabilities(capText)
84. let poll = toNumber(pollText)
85. let parallel = toNumber(parallelText)
86. let reserve = toNumber(reserveText)
87. let weekLimit = toNumber(weekText)
88. let per = toNumber(perText)
89. let quotaEvery = toNumber(everyText)
90. let size = pool.to - pool.from + 1
91. let slots = (size - size % per) / per
92. if (slots < parallel) {
93. console.log('agent: REFUSED — the port pool ' + poolText + ' has ' + slots + ' range(s) of ' + per + ' ports, --parallel ' + parallel + ' needs ' + parallel)
94. return false
95. }
96. let at = readAgentToken()
97. if (at.error != null) {
98. console.log('agent: REFUSED — ' + at.error)
99. return false
100. }
101. let tk = readToken()
102. if (tk.error != null) {
103. console.log('agent: REFUSED — ' + tk.error + ' (the workers of this host lease and post to tickets)')
104. return false
105. }
106. let host = hostName()
107. // antcolony#16: the name of the Claude account this host is logged in to; hosts sharing one account give it the same name
108. // (default: the host's own name = its own account). The scheduler pools the quota of all agents with the same name.
109. let account = accountText.trim() == '' ? host : accountText.trim()
110. let runs = runsDir(home)
111. mkDir(runs, 448)
112. mkDir(runs + '/.briefs', 448)
113. let stopFile = env('COLONY_STOP_FILE')
114. let pass = passOptions(argv)
116. let running = []
117. let live = {}
118. let events = []
119. let handled = []
120. let stopping = null
121. let finished = false
122. let syncing = false
123. let connected = null
124. let probing = false
125. let lastProbe = null
126. let reading = null
127. let quotaOk = false
128. let quotaWhy = 'the quota is not read yet'
129. let limitUntil = null
130. let loop = null
131. let watch = null
132. let n = 0
134. console.log('AGENT: started on ' + host + ' → scheduler ' + schedText + ' · poll ' + poll + ' s · parallel ' + parallel + ' · offers ' + (caps.length == 0 ? '— (nothing announced)' : caps.join(', ')) + ' · quota reserve ' + reserve + '% of the 5 h window, no start above ' + weekLimit + '% of the week (probe: ' + quotaBin + ' -p /usage, every ' + quotaEvery + ' s while a slot is free) · ports ' + pool.from + '-' + pool.to + ', ' + per + ' per session · runs ' + runs + '/ · stop file ' + (stopFile == null ? '— (no wrapper: SIGTERM kills without parking)' : stopFile) + ' · Claude account ' + account + (accountText.trim() == '' ? ' (own — none named: not shared)' : ' (quota shared with the agents of the same name)'))
136. let slotFree = (i) => {
137. for (r of running) { if (r.slot == i) { return false } }
138. return true
139. }
140. let freeSlot = () => {
141. let i = 0
142. while (i < slots) {
143. if (slotFree(i)) { return i }
144. i = i + 1
145. }
146. return -1
147. }
148. let rangeOf = (i) => {
149. let a = pool.from + i * per
150. return a + '-' + (a + per - 1)
151. }
152. let known = (s) => {
153. for (r of running) { if (r.session == s) { return true } }
154. return handled.includes(s)
155. }
156. let folderBusy = (f) => {
157. for (r of running) { if (r.folder == f) { return true } }
158. return false
159. }
160. let pushEvent = (ev) => {
161. for (e of events) { if (e.id == ev.id) { return null } }
162. events.push(ev)
163. return null
164. }
166. // the same rules as the single-host `run` (daemon.hl quotaBlock + the reserve / week checks of iterate)
167. let quotaVerdict = (q) => {
168. if (q == null || q.error != null || q.percent == null) { return 'the quota is UNKNOWN (' + (q == null ? 'no reading' : q.error) + ')' }
169. let qt = 'quota ' + q.percent + '% of 5 h' + (q.resetsMs == null ? '' : ' (resets ' + ago(q.resetsMs) + ')') + (q.weekly == null ? '' : ', week ' + q.weekly + '%')
170. if (q.percent >= 100) { return qt + ' → the 5 h limit is reached (100%)' }
171. if (q.weekly != null && q.weekly >= 100) { return qt + ' → the weekly limit is reached (100%)' }
172. if (reserve < 100 && q.percent >= reserve) { return qt + ' ≥ reserve ' + reserve + '%' }
173. if (weekLimit < 100 && (q.weekly == null || q.weekly > weekLimit)) { return qt + (q.weekly == null ? ', week UNKNOWN' : ' > week limit ' + weekLimit + '%') }
174. return null
175. }
176. let pauseForLimit = (resetMs, why) => {
177. let until = resetMs == null ? now() + LIMIT_RETRY_MS : resetMs
178. if (limitUntil == null) {
179. limitUntil = until
180. console.log('LIMIT: usage limit reached — no start until ' + ago(until) + (resetMs == null ? ' (no reset time known → ' + (LIMIT_RETRY_MS / 60000) + ' min, then try once)' : '') + ' · ' + why)
181. } else if (until > limitUntil) {
182. limitUntil = until
183. console.log('LIMIT: the pause is extended until ' + ago(until) + ' · ' + why)
184. }
185. return null
186. }
187. let accepting = () => {
188. if (stopping != null) { return { ok = false why = 'stop requested' } }
189. if (limitUntil != null) {
190. if (now() < limitUntil) { return { ok = false why = 'usage limit reached until ' + ago(limitUntil) } }
191. console.log('LIMIT: resumed — the pause until ' + ago(limitUntil) + ' is over; the quota is read again')
192. limitUntil = null
193. lastProbe = null
194. }
195. if (!quotaOk) { return { ok = false why = quotaWhy } }
196. return { ok = true why = null }
197. }
199. let flushed = false
200. let sync = null
201. let checkEnd = () => {
202. if (finished || stopping == null || running.length > 0) { return null }
203. // one last try to hand over what the sessions reported while stopping
204. if (events.length > 0 && !flushed) {
205. flushed = true
206. sync()
207. return null
208. }
209. finished = true
210. if (loop != null) { loop.stop() }
211. if (watch != null) { watch.stop() }
212. console.log('AGENT STOPPED — stop requested (' + stopping + '), no session running' + (events.length > 0 ? ' · ' + events.length + ' event(s) not delivered (the scheduler was not reachable)' : ''))
213. return null
214. }
216. let sessionEnded = (entry, r) => {
217. let out = []
218. for (x of running) { if (x.session != entry.session) { out.push(x) } }
219. running = out
220. let total = (r.workerCost == null ? 0 : r.workerCost) + (r.controllerCost == null ? 0 : r.controllerCost)
221. console.log('SESSION END ' + entry.ref + ' ' + entry.session + ' · ' + r.outcome + ' · cost ' + money(total) + ' (worker ' + money(r.workerCost) + ' + controller ' + money(r.controllerCost) + ') · ports ' + entry.ports + ' freed · running ' + running.length + '/' + parallel)
222. pushEvent({ id = entry.session + ':ended' kind = 'ended' session = entry.session ref = entry.ref outcome = r.outcome workerCost = r.workerCost == null ? 0 : r.workerCost controllerCost = r.controllerCost == null ? 0 : r.controllerCost ports = entry.ports limited = r.limited == true })
223. if (r.limited == true) { pauseForLimit(r.resetMs, entry.ref + ' ' + entry.session + ' was parked on it (' + r.limitText + ')') }
224. checkEnd()
225. return null
226. }
228. let refuse = (c, why) => {
229. console.log('AGENT: refused ' + c.kind + ' ' + c.project + '#' + c.number + ' (' + c.session + ') — ' + why)
230. pushEvent({ id = c.session + ':refused' kind = 'refused' session = c.session ref = c.project + '#' + c.number why = why })
231. return null
232. }
234. let startCommand = (c) => {
235. if (known(c.session)) { return null }
236. let ref = c.project + '#' + c.number
237. let a = accepting()
238. if (!a.ok) { return refuse(c, 'this agent is not accepting: ' + a.why) }
239. let slot = freeSlot()
240. if (slot < 0) { return refuse(c, 'no slot / port range free (' + running.length + '/' + parallel + ' running)') }
241. let ports = rangeOf(slot)
242. let entry = { ref = ref session = c.session folder = c.folder slot = slot ports = ports proc = null phase = 'worker' parkReason = null }
243. // antcolony#28: process, phase and park reason live in `live[session]`, not on `entry` (a push into `running` copies it)
244. live[c.session] = { proc = null phase = 'worker' parkReason = null }
245. let hooks = {
246. onProc = (p, phase) => {
247. live[c.session].proc = p
248. live[c.session].phase = phase
249. return null
250. }
251. parkReason = () => { return live[c.session].parkReason }
252. }
253. handled.push(c.session)
254. let ok = false
255. if (c.kind == 'resume') {
256. if (!exists(runs + '/' + c.session + '/session.json')) { return refuse(c, 'no run folder ' + runs + '/' + c.session + ' on ' + host + ' (parked on another host?)') }
257. let sj = JSON.parse(readFile(runs + '/' + c.session + '/session.json'))
258. entry.folder = sj.folder
259. if (folderBusy(entry.folder)) { return refuse(c, 'dev folder ' + entry.folder + ' is busy') }
260. running.push(entry)
261. ok = runResume(runs + '/' + c.session, pass, (r) => {
262. sessionEnded(entry, r)
263. return null
264. }, hooks, ports)
265. } else {
266. if (c.folder != null && folderBusy(c.folder)) { return refuse(c, 'dev folder ' + c.folder + ' is busy') }
267. let briefFile = runs + '/.briefs/' + c.session + '.md'
268. writeFile(briefFile, c.brief, 384)
269. running.push(entry)
270. let wargv = ['work' c.project '' + c.number '--post' '--session' c.session '--ports' ports '--brief-file' briefFile]
271. for (o of pass) { wargv.push(o) }
272. ok = runWork(wargv, (r) => {
273. sessionEnded(entry, r)
274. return null
275. }, hooks)
276. }
277. if (!ok) {
278. let out = []
279. for (x of running) { if (x.session != entry.session) { out.push(x) } }
280. running = out
281. return refuse(c, 'the worker did not start (see the lines above)')
282. }
283. console.log('AGENT: ' + (c.kind == 'resume' ? 'resumed ' : 'started ') + ref + ' (' + c.session + ', ports ' + ports + ') · running ' + running.length + '/' + parallel)
284. return null
285. }
287. // quota: read at most every --quota-every s, only while a slot is free and nothing is paused / stopping
288. let maybeProbe = () => {
289. if (probing || stopping != null || running.length >= parallel) { return null }
290. if (limitUntil != null && now() < limitUntil) { return null }
291. if (lastProbe != null && now() - lastProbe < quotaEvery * 1000) { return null }
292. probing = true
293. lastProbe = now()
294. probeUsage(quotaBin, runs, (q) => {
295. probing = false
296. reading = q == null || q.error != null || q.percent == null ? null : { percent = q.percent weekly = q.weekly resetsMs = q.resetsMs at = now() }
297. let v = quotaVerdict(q)
298. let was = quotaOk
299. quotaOk = v == null
300. quotaWhy = v == null ? null : v
301. if (quotaOk != was || n == 0) { console.log('AGENT: ' + (quotaOk ? 'quota ok (' + q.percent + '% of 5 h' + (q.weekly == null ? '' : ', week ' + q.weekly + '%') + ') → accepting work' : 'not accepting work — ' + v)) }
302. return null
303. })
304. return null
305. }
307. sync = () => {
308. if (syncing || finished) { return null }
309. syncing = true
310. n = n + 1
311. maybeProbe()
312. let a = accepting()
313. let list = []
314. for (r of running) { list.push({ ref = r.ref session = r.session folder = r.folder phase = r.phase ports = r.ports }) }
315. let payload = { host = host account = account quota = reading limitUntil = limitUntil capabilities = caps parallel = parallel accepting = a.ok why = a.why running = list events = events }
316. let res = fetch(schedText + '/agent/sync', { method = 'POST' headers = { 'user-agent' = userAgent() authorization = 'Bearer ' + at.token } json = payload timeoutMs = 15000 })
317. if (res == null || res.status != 200) {
318. let why = res == null ? 'no answer' : 'answered ' + res.status
319. if (connected != false) { console.log('AGENT: scheduler ' + schedText + ' not reachable (' + why + ') — ' + running.length + ' session(s) keep running, ' + events.length + ' event(s) buffered, trying again every ' + poll + ' s') }
320. connected = false
321. syncing = false
322. return null
323. }
324. if (connected != true) { console.log('AGENT: connected to ' + schedText + (connected == false ? ' again' : '') + ' · ' + events.length + ' buffered event(s) delivered') }
325. connected = true
326. let j = res.json()
327. let keep = []
328. for (e of events) { if (!j.ack.includes(e.id)) { keep.push(e) } }
329. events = keep
330. for (c of j.commands) { startCommand(c) }
331. syncing = false
332. checkEnd()
333. return null
334. }
336. let checkStop = () => {
337. if (finished || stopFile == null || stopFile == '' || !exists(stopFile)) { return null }
338. let mode = readFile(stopFile).trim()
339. if (stopping == null) {
340. stopping = 'finish'
341. console.log('AGENT: stop requested — accepting nothing more; ' + running.length + ' running session(s) finish (a second SIGTERM/SIGINT parks them instead)')
342. }
343. if (mode == 'park' && stopping != 'park') {
344. stopping = 'park'
345. console.log('AGENT: second stop signal — parking ' + running.length + ' running session(s)')
346. for (r of running) {
347. if (live[r.session] != null) {
348. live[r.session].parkReason = 'the agent was stopped (second SIGTERM/SIGINT)'
349. if (live[r.session].proc != null) {
350. console.log('AGENT: killing ' + live[r.session].phase + ' pid ' + live[r.session].proc.pid + ' of ' + r.ref + ' ' + r.session + ' → parked')
351. live[r.session].proc.kill()
352. }
353. }
354. }
355. }
356. checkEnd()
357. return null
358. }
360. watch = every(0.5)
361. on watch.tick(x) {
362. checkStop()
363. return null
364. }
365. loop = every(poll)
366. on loop.tick(x) {
367. sync()
368. return null
369. }
370. sync()
371. return true
372. }