Something went wrong. Try again.
small gleam coding and persistent agent daemon with a detachable cli and a canvas ui forked from okami.mom/albedo
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788"""Detached kernel identities retained beside their sockets."""
from __future__ import annotations
import jsonimport osfrom pathlib import Pathimport statimport tempfile
import albedo_proc
NAME = "identity.json"
def write(run_dir: str, groups: dict[str, dict[str, object]] | None = None) -> None: """Publish this kernel's identity without storing its attach token.""" directory = Path(run_dir).resolve() identity = { "home": str(directory.parent.parent), "run_dir": str(directory), "cwd": os.getcwd(), "pid": os.getpid(), "pgid": os.getpgrp(), "leader": albedo_proc.leader_token(os.getpid()), "groups": { identity: { key: group[key] for key in ("pid", "pgid", "leader") if key in group } for identity, group in (groups or {}).items() }, } fd, temporary = tempfile.mkstemp(dir=directory, prefix=".identity-") try: with os.fdopen(fd, "w") as target: json.dump(identity, target) os.replace(temporary, directory / NAME) finally: if os.path.exists(temporary): os.unlink(temporary)
def discover(home: str, known: list[str]) -> list[dict[str, str]]: """Return verified local kernels absent from the daemon's ownership records.""" root = Path(home).resolve() runs = root / "run" found = [] excluded = {Path(path).resolve() for path in known} if not runs.is_dir(): return found for directory in runs.iterdir(): if directory.is_symlink() or directory in excluded or not directory.is_dir(): continue try: if not stat.S_ISSOCK((directory / "kernel.sock").stat().st_mode): continue with (directory / NAME).open() as source: identity = json.loads(source.read(1024 * 1024 + 1)) if not isinstance(identity, dict): continue pid = identity.get("pid") leader = identity.get("leader") if ( not isinstance(pid, int) or pid <= 1 or identity.get("pgid") != pid or not isinstance(leader, str) or identity.get("home") != str(root) or identity.get("run_dir") != str(directory) or not isinstance(identity.get("cwd"), str) or albedo_proc.leader_token(pid) != leader or os.getpgid(pid) != pid ): continue os.kill(pid, 0) owned = json.dumps( { key: identity.get(key, {}) for key in ("pid", "pgid", "leader", "groups") } ) found.append( {"run_dir": str(directory), "cwd": identity["cwd"], "owned": owned} ) except (OSError, ValueError): continue return found