351 lines
15 KiB
Python
351 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""Opt-in Codex hooks: relay completed messages to a selected Fenrir override.
|
|
|
|
No speech, model feedback, tool output, or transcript copies are produced. A
|
|
single detached follower per conversation reads the transcript supplied by
|
|
Codex and keeps at most nine completed assistant messages in memory.
|
|
"""
|
|
|
|
import collections
|
|
import fcntl
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import socket
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
|
|
MAX_MESSAGE = 1024 * 1024
|
|
MAX_LINE = 4 * MAX_MESSAGE
|
|
STATUSES = {
|
|
"SessionStart": "idle", "UserPromptSubmit": "working",
|
|
"PermissionRequest": "permission", "PostToolUse": "working",
|
|
"Stop": "done", "Interrupt": "interrupted", "SessionEnd": "closed",
|
|
}
|
|
|
|
|
|
def process_info(pid):
|
|
# The process name may itself contain spaces or parentheses.
|
|
fields = Path(f"/proc/{pid}/stat").read_text().rsplit(")", 1)[1].split()
|
|
return int(fields[1]), int(fields[4]), fields[19]
|
|
|
|
|
|
def codex_process():
|
|
pid = os.getppid()
|
|
while pid > 1:
|
|
parent, tty, started = process_info(pid)
|
|
executable = os.path.basename(os.readlink(f"/proc/{pid}/exe"))
|
|
if executable == "codex":
|
|
return {"pid": pid, "started": started, "tty": tty}
|
|
pid = parent
|
|
raise ValueError("no Codex ancestor; run this command as a Codex hook")
|
|
|
|
|
|
def private_directory():
|
|
runtimeRoot = Path(os.environ.get("XDG_RUNTIME_DIR", tempfile.gettempdir()))
|
|
directory = runtimeRoot / f"fenrir-codex-{os.getuid()}"
|
|
directory.mkdir(mode=0o700, exist_ok=True)
|
|
info = directory.lstat()
|
|
if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid() or info.st_mode & 0o077:
|
|
raise ValueError("unsafe bridge runtime directory")
|
|
return directory
|
|
|
|
|
|
def write_status(path, status):
|
|
write_json(path, {"status": status})
|
|
|
|
|
|
def write_json(path, data):
|
|
descriptor, temporary = tempfile.mkstemp(prefix=".status-", dir=path.parent)
|
|
try:
|
|
with os.fdopen(descriptor, "w") as stream:
|
|
json.dump(data, stream)
|
|
os.replace(temporary, path)
|
|
finally:
|
|
if os.path.exists(temporary):
|
|
os.unlink(temporary)
|
|
|
|
|
|
def hook(payload):
|
|
eventName = payload.get("hook_event_name")
|
|
if eventName not in STATUSES:
|
|
return
|
|
session = payload.get("session_id", "")
|
|
transcript = payload.get("transcript_path")
|
|
if not isinstance(session, str) or not session or not isinstance(transcript, str):
|
|
raise ValueError("Codex hook did not supply a session and transcript")
|
|
owner = codex_process()
|
|
window = os.environ.get("WINDOWID", "")
|
|
# Linux virtual consoles use major 4, minor 1..63. Never guess a VT from a PTY.
|
|
terminal = 1 if window else os.minor(owner["tty"])
|
|
if not window and (os.major(owner["tty"]) != 4 or not 1 <= terminal <= 63):
|
|
raise ValueError("no X11 window or Linux console identifies this session")
|
|
directory = private_directory()
|
|
key = hashlib.sha256(f"{session}:{owner['pid']}:{owner['started']}".encode()).hexdigest()
|
|
statusPath = directory / f"{key}.json"
|
|
status = STATUSES[eventName]
|
|
if eventName == "Stop":
|
|
lines = str(payload.get("last_assistant_message") or "").lstrip().splitlines()
|
|
if lines and lines[0].strip() == "I have a question.":
|
|
status = "question"
|
|
write_status(statusPath, status)
|
|
if eventName not in ("SessionStart", "UserPromptSubmit"):
|
|
return
|
|
terminalKey = hashlib.sha256(f"{window}:{terminal}".encode()).hexdigest()
|
|
activePath = directory / f"active-{terminalKey}.json"
|
|
descriptor = os.open(str(activePath) + ".lock", os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
|
|
with os.fdopen(descriptor, "w") as lock:
|
|
fcntl.flock(lock, fcntl.LOCK_EX)
|
|
active = json.loads(activePath.read_text()) if activePath.exists() else {}
|
|
if active.get("key") != key:
|
|
write_json(activePath, {"key": key, "epoch": max(time.time_ns(), active.get("epoch", 0) + 1)})
|
|
config = dict(owner, window=window, terminal=terminal, session=session,
|
|
transcript=transcript, status_path=str(statusPath),
|
|
lock_path=str(directory / f"{key}.lock"), active_path=str(activePath), key=key)
|
|
subprocess.Popen([sys.executable, str(Path(__file__).resolve()), "--watch", json.dumps(config)],
|
|
stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL, start_new_session=True, close_fds=True)
|
|
|
|
|
|
def completed_message(record, offset):
|
|
if record.get("type") != "event_msg":
|
|
return None
|
|
payload = record.get("payload", {})
|
|
if payload.get("type") == "agent_message":
|
|
text = payload.get("message")
|
|
return (str(offset), clean_message(text)) if isinstance(text, str) and text.strip() else None
|
|
if payload.get("type") == "item_completed":
|
|
item = payload.get("item", {})
|
|
if item.get("type") == "AgentMessage":
|
|
text = "".join(part.get("text", "") for part in item.get("content", [])
|
|
if part.get("type") == "Text")
|
|
return (str(offset), clean_message(text)) if text.strip() else None
|
|
return None
|
|
|
|
|
|
def clean_message(text):
|
|
# Codex appends machine-readable memory attribution to some replies.
|
|
return re.sub(r"<oai-mem-citation>.*?</oai-mem-citation>", "", text, flags=re.DOTALL).strip()
|
|
|
|
|
|
def record_status(record):
|
|
if record.get("type") != "event_msg":
|
|
return None
|
|
return {"task_started": "working", "task_complete": "done",
|
|
"turn_aborted": "interrupted"}.get(record.get("payload", {}).get("type"))
|
|
|
|
|
|
def reader_sockets(config, registry=None):
|
|
if registry is None:
|
|
registry = Path(tempfile.gettempdir()) / "fenrirscreenreader-instances"
|
|
for path in registry.glob("*.json"):
|
|
try:
|
|
info = path.lstat()
|
|
if not stat.S_ISREG(info.st_mode) or info.st_uid not in (0, os.getuid()):
|
|
continue
|
|
if info.st_size > 16384:
|
|
continue
|
|
data = json.loads(path.read_text())
|
|
if time.time() - data["updated_at"] > 30:
|
|
continue
|
|
if config["window"]:
|
|
matches = (data["screen_driver"] == "ptyDriver"
|
|
and data.get("x11_window_id") == config["window"]
|
|
and info.st_uid == os.getuid())
|
|
else:
|
|
matches = data["screen_driver"] == "vcsaDriver"
|
|
if not matches:
|
|
continue
|
|
try:
|
|
os.kill(data["pid"], 0)
|
|
except PermissionError:
|
|
pass # A user may connect to the root console reader.
|
|
for name in data["socket_files"]:
|
|
# An explicit socket_file is also private. Never use the shared
|
|
# router: it may belong to another window or console reader.
|
|
if Path(name).is_absolute() and Path(name).name != "fenrirscreenreader-daemon.sock":
|
|
socketInfo = os.lstat(name)
|
|
if stat.S_ISSOCK(socketInfo.st_mode) and socketInfo.st_uid == info.st_uid:
|
|
yield name, data["pid"]
|
|
except (OSError, ValueError, KeyError, TypeError):
|
|
continue
|
|
|
|
|
|
def send_event(destination, event):
|
|
path, expectedPID = destination
|
|
encoded = ("APPLICATION CODEX " + json.dumps(event, ensure_ascii=False) + "\n").encode()
|
|
if len(encoded) > 8129:
|
|
raise ValueError("bridge packet exceeds Fenrir protocol limit")
|
|
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection:
|
|
connection.settimeout(0.5)
|
|
connection.connect(path)
|
|
import struct
|
|
peerPID, _, _ = struct.unpack("3i", connection.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12))
|
|
if peerPID != expectedPID:
|
|
return False
|
|
connection.sendall(encoded)
|
|
connection.shutdown(socket.SHUT_WR)
|
|
return connection.recv(1024).strip() == b""
|
|
|
|
|
|
def message_events(base, message):
|
|
identifier, text, baseline = message
|
|
# Bound JSON expansion as well as UTF-8 length (including control characters).
|
|
chunks, current, size = [], [], 0
|
|
for char in text:
|
|
charSize = len(json.dumps(char, ensure_ascii=False).encode()) - 2
|
|
if size + charSize > 5000 and current:
|
|
chunks.append("".join(current))
|
|
current, size = [], 0
|
|
current.append(char)
|
|
size += charSize
|
|
if current:
|
|
chunks.append("".join(current))
|
|
return [dict(base, kind="message", id=identifier, text=chunk, part=index,
|
|
parts=len(chunks), baseline=baseline) for index, chunk in enumerate(chunks)]
|
|
|
|
|
|
class Follower:
|
|
def __init__(self, session):
|
|
self.session = session
|
|
self.offset = 0
|
|
self.history = collections.deque(maxlen=9)
|
|
self.status = "idle"
|
|
self.verified = False
|
|
|
|
def read_available(self, stream, baseline=False):
|
|
stream.seek(self.offset)
|
|
while True:
|
|
offset = stream.tell()
|
|
line = stream.readline(MAX_LINE + 1)
|
|
if len(line) > MAX_LINE:
|
|
raise ValueError("Codex transcript record exceeds 4 MiB")
|
|
if not line or not line.endswith(b"\n"):
|
|
return # A partial JSON record is not a completed message.
|
|
self.offset = stream.tell()
|
|
record = json.loads(line)
|
|
if not self.verified:
|
|
if record.get("type") != "session_meta" or record.get("payload", {}).get("id") != self.session:
|
|
raise ValueError("Codex transcript belongs to another session")
|
|
self.verified = True
|
|
continue
|
|
message = completed_message(record, offset)
|
|
if message:
|
|
if len(message[1].encode()) > MAX_MESSAGE:
|
|
raise ValueError("Codex message exceeds 1 MiB")
|
|
self.history.append([message[0], message[1], baseline])
|
|
status = record_status(record)
|
|
if status:
|
|
if status == "done" and self.history and self.history[-1][1].splitlines()[0].strip() == "I have a question.":
|
|
status = "question"
|
|
self.status = status
|
|
|
|
|
|
def watch(config):
|
|
descriptor = os.open(config["lock_path"], os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
|
|
with os.fdopen(descriptor, "w") as lock:
|
|
try:
|
|
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
except BlockingIOError:
|
|
return
|
|
follower = Follower(config["session"])
|
|
base = {key: config[key] for key in ("window", "terminal", "session")}
|
|
base["version"] = 1
|
|
active = json.loads(Path(config["active_path"]).read_text())
|
|
if active.get("key") != config["key"]:
|
|
return
|
|
base["epoch"] = active["epoch"]
|
|
statusStamp, lastSend, lastState = None, 0, None
|
|
stream = None
|
|
try:
|
|
while True:
|
|
if json.loads(Path(config["active_path"]).read_text()).get("key") != config["key"]:
|
|
break # Another conversation now owns this terminal.
|
|
alive = False
|
|
try:
|
|
alive = process_info(config["pid"])[2] == config["started"]
|
|
except OSError:
|
|
pass
|
|
if stream is None and alive:
|
|
try:
|
|
descriptor = os.open(config["transcript"], os.O_RDONLY | os.O_NOFOLLOW)
|
|
stream = os.fdopen(descriptor, "rb")
|
|
info = os.fstat(stream.fileno())
|
|
if not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid():
|
|
raise ValueError("unsafe Codex transcript")
|
|
follower.read_available(stream, baseline=True)
|
|
except FileNotFoundError:
|
|
pass # SessionStart may precede the first durable record.
|
|
elif stream is not None:
|
|
follower.read_available(stream)
|
|
statusPath = Path(config["status_path"])
|
|
stamp = statusPath.stat().st_mtime_ns
|
|
if stamp != statusStamp:
|
|
follower.status = json.loads(statusPath.read_text())["status"]
|
|
statusStamp = stamp
|
|
if not alive:
|
|
follower.status = "closed"
|
|
state = (follower.status, tuple((message[0], message[2]) for message in follower.history))
|
|
if state != lastState or time.monotonic() - lastSend >= 2:
|
|
for destination in reader_sockets(config):
|
|
try:
|
|
if not send_event(destination, dict(base, kind="status", status=follower.status)):
|
|
continue
|
|
success = True
|
|
for message in follower.history:
|
|
for event in message_events(base, message):
|
|
if not send_event(destination, event):
|
|
success = False
|
|
break
|
|
if not success:
|
|
break
|
|
if success:
|
|
for message in follower.history:
|
|
message[2] = True
|
|
except (OSError, ValueError):
|
|
continue
|
|
lastState, lastSend = state, time.monotonic()
|
|
if not alive or follower.status == "closed":
|
|
break
|
|
time.sleep(0.2)
|
|
finally:
|
|
if stream is not None:
|
|
stream.close()
|
|
# Keep the lock inode stable: another watcher may already have it open.
|
|
# Runtime state contains status only and is overwritten on the next hook.
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) == 3 and sys.argv[1] == "--watch":
|
|
config = json.loads(sys.argv[2])
|
|
try:
|
|
watch(config)
|
|
except (OSError, ValueError, KeyError, TypeError):
|
|
write_status(Path(config["status_path"]), "error")
|
|
event = {key: config[key] for key in ("window", "terminal", "session")}
|
|
active = json.loads(Path(config["active_path"]).read_text())
|
|
if active.get("key") != config["key"]:
|
|
return
|
|
event.update(version=1, epoch=active["epoch"], kind="status", status="error")
|
|
for destination in reader_sockets(config):
|
|
try:
|
|
send_event(destination, event)
|
|
except OSError:
|
|
pass
|
|
return
|
|
try:
|
|
hook(json.load(sys.stdin))
|
|
except (OSError, ValueError, KeyError, TypeError) as error:
|
|
print(f"Fenrir Codex bridge: {error}", file=sys.stderr)
|
|
# Neutral output: no context injection, approval decisions, or continuation.
|
|
print("{}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|