deck/collectors/darwin-collect.py

263 lines · python

"""
darwin-collect  - cross-host session collector for the Darwin Deck (v2).
 
Runs centrally on CT215. Over root SSH it reads the *newest* session log on
each remote agent host, normalizes new activity into contract-§1 events, and
POSTs them to the Deck `/event`. Change-only: per (host,session) byte offset is
persisted on local ext4 so each run only ships new records (idempotent upserts
on the Deck side make re-sends harmless anyway).
 
Collected hosts:
  - ct247 (REDACTED-IP) - OpenCode / GLM, ~/.local/share/opencode/log/opencode.log
                            (key=value text log; model glm-5.2, session.id=ses_*)
  - ct241 (REDACTED-IP) - Codex CLI, newest ~/.codex/sessions/YYYY/MM/DD/rollout-*.jsonl
                            (rollout JSONL; model from turn_context, tokens from token_count)
 
State (local ext4, NOT CIFS): /var/tmp/darwin-dev/collectors/state/<host>.json
Deck:  http://REDACTED-IP:9893/event   (override DARWIN_DECK_URL)
 
Designed to be invoked by a systemd timer (darwin-collect.timer) every 30s, or
looped. Each invocation is a single pass and exits. Never raises on a dead host
or dead Deck - the bus must survive.
"""
from __future__ import annotations
 
import json
import os
import re
import subprocess
import sys
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
 
DECK_URL = os.environ.get("DARWIN_DECK_URL", "http://REDACTED-IP:9893").rstrip("/")
STATE_DIR = Path(os.environ.get("DARWIN_STATE_DIR", "/var/tmp/darwin-dev/collectors/state"))
SSH = ["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=6", "-o", "StrictHostKeyChecking=no"]
 
HOSTS = {
    "ct247": "REDACTED-IP",
    "ct241": "REDACTED-IP",
}
 
 
def _now() -> str:
    return datetime.now(timezone.utc).isoformat()
 
 
def _post(event: dict) -> bool:
    data = json.dumps(event).encode()
    req = urllib.request.Request(
        DECK_URL + "/event", data=data,
        headers={"Content-Type": "application/json"}, method="POST",
    )
    try:
        urllib.request.urlopen(req, timeout=5).read()
        return True
    except Exception as e:
        sys.stderr.write(f"[collect] POST failed: {e}\n")
        return False
 
 
def _ssh(ip: str, cmd: str, timeout: int = 25) -> str | None:
    try:
        out = subprocess.run(
            SSH + [f"root@{ip}", cmd],
            capture_output=True, text=True, timeout=timeout,
        )
        if out.returncode != 0:
            return None
        return out.stdout
    except Exception:
        return None
 
 
def _load_state(host: str) -> dict:
    f = STATE_DIR / f"{host}.json"
    if f.exists():
        try:
            return json.loads(f.read_text())
        except Exception:
            return {}
    return {}
 
 
def _save_state(host: str, state: dict) -> None:
    STATE_DIR.mkdir(parents=True, exist_ok=True)
    (STATE_DIR / f"{host}.json").write_text(json.dumps(state))
 
 
_KV = re.compile(r'(\w[\w.]*)=("(?:[^"\\]|\\.)*"|\S+)')
LOG_247 = "/root/.local/share/opencode/log/opencode.log"
 
 
def _kv(line: str) -> dict:
    d = {}
    for k, v in _KV.findall(line):
        if v and v[0] == '"':
            v = v[1:-1]
        d[k] = v
    return d
 
 
def collect_opencode(host: str, ip: str) -> int:
    state = _load_state(host)
    offset = int(state.get("offset", 0))
    size_out = _ssh(ip, f"stat -c %s {LOG_247} 2>/dev/null")
    if not size_out or not size_out.strip().isdigit():
        return 0
    size = int(size_out.strip())
    if size < offset:
        offset = 0
    if size <= offset:
        return 0
    chunk = _ssh(ip, f"tail -c +{offset + 1} {LOG_247} 2>/dev/null")
    if chunk is None:
        return 0
 
    sessions = state.get("sessions", {})
    posted = 0
    for line in chunk.splitlines():
        if "session.id=" not in line and "run=" not in line:
            continue
        kv = _kv(line)
        sid = kv.get("session.id")
        msg = kv.get("message", "")
        model = kv.get("llm.model", "glm-5.2")
        ts = kv.get("timestamp", _now())
        if not sid:
            continue
        s = sessions.setdefault(sid, {"model": model, "started": False})
        if model and model != "":
            s["model"] = model
        aid = f"opencode-{sid}"
        base = {
            "ts": ts, "agent_id": aid, "parent_id": None,
            "label": f"GLM {sid[:12]}", "model": s["model"],
            "host": host, "session": sid,
        }
        if not s["started"]:
            s["started"] = True
            ev = dict(base, type="spawned", status="running")
            if _post(ev):
                posted += 1
        if "loop" in msg or "shell tool" in msg or "creating" in msg:
            ev = dict(base, type="tool_call", status="running",
                      tool=kv.get("step") and f"step:{kv['step']}" or "loop")
            if _post(ev):
                posted += 1
        if "exiting loop" in msg or "disposing instance" in msg:
            ev = dict(base, type="returned", status="done")
            if _post(ev):
                posted += 1
    state["offset"] = size
    state["sessions"] = sessions
    _save_state(host, state)
    return posted
 
 
def collect_codex(host: str, ip: str) -> int:
    newest = _ssh(
        ip,
        "ls -t /root/.codex/sessions/*/*/*/rollout-*.jsonl "
        "/home/zion/.codex/sessions/*/*/*/rollout-*.jsonl 2>/dev/null | head -1",
    )
    if not newest or not newest.strip():
        return 0
    path = newest.strip().splitlines()[0]
 
    state = _load_state(host)
    sess = state.get("path") == path and state or {"path": path, "offset": 0}
    offset = int(sess.get("offset", 0))
 
    size_out = _ssh(ip, f"stat -c %s '{path}' 2>/dev/null")
    if not size_out or not size_out.strip().isdigit():
        return 0
    size = int(size_out.strip())
    if size < offset:
        offset = 0
    if size <= offset:
        return 0
    chunk = _ssh(ip, f"tail -c +{offset + 1} '{path}' 2>/dev/null")
    if chunk is None:
        return 0
 
    m = re.search(r"rollout-[\dT-]+-([0-9a-f-]+)\.jsonl", path)
    sid = m.group(1) if m else Path(path).stem
    aid = f"codex-{sid}"
    model = sess.get("model", "gpt-5.5")
    tokens = int(sess.get("tokens", 0))
    started = sess.get("started", False)
 
    posted = 0
    for raw in chunk.splitlines():
        raw = raw.strip()
        if not raw:
            continue
        try:
            o = json.loads(raw)
        except Exception:
            continue
        t = o.get("type")
        p = o.get("payload", {})
        ts = o.get("timestamp", _now())
        base = {
            "agent_id": aid, "parent_id": None, "host": host, "session": sid,
        }
        if t == "session_meta":
            sid2 = p.get("id", sid)
            base["session"] = sid2
            if not started:
                started = True
                ev = dict(base, ts=ts, type="spawned", status="running",
                          label=f"Codex {sid2[:12]}", model=model)
                if _post(ev):
                    posted += 1
        elif t == "turn_context":
            model = p.get("model", model)
        elif t == "event_msg":
            st = p.get("type")
            if st == "token_count":
                tu = p.get("info", {}).get("total_token_usage", {})
                tokens = max(tokens, int(tu.get("total_tokens", tokens)))
                ev = dict(base, ts=ts, type="token_tick", status="running",
                          label=f"Codex {sid[:12]}", model=model, tokens=tokens)
                if _post(ev):
                    posted += 1
            elif st in ("function_call", "mcp_tool_call_end", "patch_apply_end", "agent_message"):
                if not started:
                    started = True
                    _post(dict(base, ts=ts, type="spawned", status="running",
                               label=f"Codex {sid[:12]}", model=model))
                ev = dict(base, ts=ts, type="tool_call", status="running",
                          label=f"Codex {sid[:12]}", model=model,
                          tokens=tokens, tool=st)
                if _post(ev):
                    posted += 1
            elif st == "task_complete":
                ev = dict(base, ts=ts, type="returned", status="done",
                          label=f"Codex {sid[:12]}", model=model, tokens=tokens)
                if _post(ev):
                    posted += 1
 
    _save_state(host, {"path": path, "offset": size, "model": model,
                       "tokens": tokens, "started": started})
    return posted
 
 
def main() -> int:
    total = 0
    for host, ip in HOSTS.items():
        try:
            if host == "ct247":
                total += collect_opencode(host, ip)
            elif host == "ct241":
                total += collect_codex(host, ip)
        except Exception as e:
            sys.stderr.write(f"[collect] {host} error: {e}\n")
    print(f"[collect] posted {total} events at {_now()}")
    return 0
 
 
if __name__ == "__main__":
    sys.exit(main())

1	"""
2	darwin-collect - cross-host session collector for the Darwin Deck (v2).
3
4	Runs centrally on CT215. Over root SSH it reads the newest session log on
5	each remote agent host, normalizes new activity into contract-§1 events, and
6	POSTs them to the Deck `/event`. Change-only: per (host,session) byte offset is
7	persisted on local ext4 so each run only ships new records (idempotent upserts
8	on the Deck side make re-sends harmless anyway).
9
10	Collected hosts:
11	- ct247 (REDACTED-IP) - OpenCode / GLM, ~/.local/share/opencode/log/opencode.log
12	(key=value text log; model glm-5.2, session.id=ses_*)
13	- ct241 (REDACTED-IP) - Codex CLI, newest ~/.codex/sessions/YYYY/MM/DD/rollout-*.jsonl
14	(rollout JSONL; model from turn_context, tokens from token_count)
15
16	State (local ext4, NOT CIFS): /var/tmp/darwin-dev/collectors/state/<host>.json
17	Deck: http://REDACTED-IP:9893/event (override DARWIN_DECK_URL)
18
19	Designed to be invoked by a systemd timer (darwin-collect.timer) every 30s, or
20	looped. Each invocation is a single pass and exits. Never raises on a dead host
21	or dead Deck - the bus must survive.
22	"""
23	from __future__ import annotations
24
25	import json
26	import os
27	import re
28	import subprocess
29	import sys
30	import urllib.request
31	from datetime import datetime, timezone
32	from pathlib import Path
33
34	DECK_URL = os.environ.get("DARWIN_DECK_URL", "http://REDACTED-IP:9893").rstrip("/")
35	STATE_DIR = Path(os.environ.get("DARWIN_STATE_DIR", "/var/tmp/darwin-dev/collectors/state"))
36	SSH = ["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=6", "-o", "StrictHostKeyChecking=no"]
37
38	HOSTS = {
39	"ct247": "REDACTED-IP",
40	"ct241": "REDACTED-IP",
41	}
42
43
44	def _now() -> str:
45	return datetime.now(timezone.utc).isoformat()
46
47
48	def _post(event: dict) -> bool:
49	data = json.dumps(event).encode()
50	req = urllib.request.Request(
51	DECK_URL + "/event", data=data,
52	headers={"Content-Type": "application/json"}, method="POST",
53	)
54	try:
55	urllib.request.urlopen(req, timeout=5).read()
56	return True
57	except Exception as e:
58	sys.stderr.write(f"[collect] POST failed: {e}\n")
59	return False
60
61
62	def _ssh(ip: str, cmd: str, timeout: int = 25) -> str \| None:
63	try:
64	out = subprocess.run(
65	SSH + [f"root@{ip}", cmd],
66	capture_output=True, text=True, timeout=timeout,
67	)
68	if out.returncode != 0:
69	return None
70	return out.stdout
71	except Exception:
72	return None
73
74
75	def _load_state(host: str) -> dict:
76	f = STATE_DIR / f"{host}.json"
77	if f.exists():
78	try:
79	return json.loads(f.read_text())
80	except Exception:
81	return {}
82	return {}
83
84
85	def _save_state(host: str, state: dict) -> None:
86	STATE_DIR.mkdir(parents=True, exist_ok=True)
87	(STATE_DIR / f"{host}.json").write_text(json.dumps(state))
88
89
90	_KV = re.compile(r'(\w[\w.])=("(?:[^"\\]\|\\.)"\|\S+)')
91	LOG_247 = "/root/.local/share/opencode/log/opencode.log"
92
93
94	def _kv(line: str) -> dict:
95	d = {}
96	for k, v in _KV.findall(line):
97	if v and v[0] == '"':
98	v = v[1:-1]
99	d[k] = v
100	return d
101
102
103	def collect_opencode(host: str, ip: str) -> int:
104	state = _load_state(host)
105	offset = int(state.get("offset", 0))
106	size_out = _ssh(ip, f"stat -c %s {LOG_247} 2>/dev/null")
107	if not size_out or not size_out.strip().isdigit():
108	return 0
109	size = int(size_out.strip())
110	if size < offset:
111	offset = 0
112	if size <= offset:
113	return 0
114	chunk = _ssh(ip, f"tail -c +{offset + 1} {LOG_247} 2>/dev/null")
115	if chunk is None:
116	return 0
117
118	sessions = state.get("sessions", {})
119	posted = 0
120	for line in chunk.splitlines():
121	if "session.id=" not in line and "run=" not in line:
122	continue
123	kv = _kv(line)
124	sid = kv.get("session.id")
125	msg = kv.get("message", "")
126	model = kv.get("llm.model", "glm-5.2")
127	ts = kv.get("timestamp", _now())
128	if not sid:
129	continue
130	s = sessions.setdefault(sid, {"model": model, "started": False})
131	if model and model != "":
132	s["model"] = model
133	aid = f"opencode-{sid}"
134	base = {
135	"ts": ts, "agent_id": aid, "parent_id": None,
136	"label": f"GLM {sid[:12]}", "model": s["model"],
137	"host": host, "session": sid,
138	}
139	if not s["started"]:
140	s["started"] = True
141	ev = dict(base, type="spawned", status="running")
142	if _post(ev):
143	posted += 1
144	if "loop" in msg or "shell tool" in msg or "creating" in msg:
145	ev = dict(base, type="tool_call", status="running",
146	tool=kv.get("step") and f"step:{kv['step']}" or "loop")
147	if _post(ev):
148	posted += 1
149	if "exiting loop" in msg or "disposing instance" in msg:
150	ev = dict(base, type="returned", status="done")
151	if _post(ev):
152	posted += 1
153	state["offset"] = size
154	state["sessions"] = sessions
155	_save_state(host, state)
156	return posted
157
158
159	def collect_codex(host: str, ip: str) -> int:
160	newest = _ssh(
161	ip,
162	"ls -t /root/.codex/sessions////rollout-.jsonl "
163	"/home/zion/.codex/sessions////rollout-.jsonl 2>/dev/null \| head -1",
164	)
165	if not newest or not newest.strip():
166	return 0
167	path = newest.strip().splitlines()[0]
168
169	state = _load_state(host)
170	sess = state.get("path") == path and state or {"path": path, "offset": 0}
171	offset = int(sess.get("offset", 0))
172
173	size_out = _ssh(ip, f"stat -c %s '{path}' 2>/dev/null")
174	if not size_out or not size_out.strip().isdigit():
175	return 0
176	size = int(size_out.strip())
177	if size < offset:
178	offset = 0
179	if size <= offset:
180	return 0
181	chunk = _ssh(ip, f"tail -c +{offset + 1} '{path}' 2>/dev/null")
182	if chunk is None:
183	return 0
184
185	m = re.search(r"rollout-[\dT-]+-([0-9a-f-]+)\.jsonl", path)
186	sid = m.group(1) if m else Path(path).stem
187	aid = f"codex-{sid}"
188	model = sess.get("model", "gpt-5.5")
189	tokens = int(sess.get("tokens", 0))
190	started = sess.get("started", False)
191
192	posted = 0
193	for raw in chunk.splitlines():
194	raw = raw.strip()
195	if not raw:
196	continue
197	try:
198	o = json.loads(raw)
199	except Exception:
200	continue
201	t = o.get("type")
202	p = o.get("payload", {})
203	ts = o.get("timestamp", _now())
204	base = {
205	"agent_id": aid, "parent_id": None, "host": host, "session": sid,
206	}
207	if t == "session_meta":
208	sid2 = p.get("id", sid)
209	base["session"] = sid2
210	if not started:
211	started = True
212	ev = dict(base, ts=ts, type="spawned", status="running",
213	label=f"Codex {sid2[:12]}", model=model)
214	if _post(ev):
215	posted += 1
216	elif t == "turn_context":
217	model = p.get("model", model)
218	elif t == "event_msg":
219	st = p.get("type")
220	if st == "token_count":
221	tu = p.get("info", {}).get("total_token_usage", {})
222	tokens = max(tokens, int(tu.get("total_tokens", tokens)))
223	ev = dict(base, ts=ts, type="token_tick", status="running",
224	label=f"Codex {sid[:12]}", model=model, tokens=tokens)
225	if _post(ev):
226	posted += 1
227	elif st in ("function_call", "mcp_tool_call_end", "patch_apply_end", "agent_message"):
228	if not started:
229	started = True
230	_post(dict(base, ts=ts, type="spawned", status="running",
231	label=f"Codex {sid[:12]}", model=model))
232	ev = dict(base, ts=ts, type="tool_call", status="running",
233	label=f"Codex {sid[:12]}", model=model,
234	tokens=tokens, tool=st)
235	if _post(ev):
236	posted += 1
237	elif st == "task_complete":
238	ev = dict(base, ts=ts, type="returned", status="done",
239	label=f"Codex {sid[:12]}", model=model, tokens=tokens)
240	if _post(ev):
241	posted += 1
242
243	_save_state(host, {"path": path, "offset": size, "model": model,
244	"tokens": tokens, "started": started})
245	return posted
246
247
248	def main() -> int:
249	total = 0
250	for host, ip in HOSTS.items():
251	try:
252	if host == "ct247":
253	total += collect_opencode(host, ip)
254	elif host == "ct241":
255	total += collect_codex(host, ip)
256	except Exception as e:
257	sys.stderr.write(f"[collect] {host} error: {e}\n")
258	print(f"[collect] posted {total} events at {_now()}")
259	return 0
260
261
262	if __name__ == "__main__":
263	sys.exit(main())