Something went wrong. Try again.
small gleam coding and (not yet) persistent agent daemon with a detachable cli forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208"""Bounded file-access evidence and before/after diffs for the terminal view.
This is observation, not a sandbox. External processes and native writes may bypass it."""
from __future__ import annotations
from collections.abc import Callable, Iteratorfrom typing import Protocol, castimport albedo_shellimport contextlibimport contextvarsimport difflibimport osimport shleximport sys
GUARD = contextvars.ContextVar("albedo_trace_guard", default=False)LIMIT = 128 * 1024
@contextlib.contextmanagerdef unobserved() -> Iterator[None]: """Harness bookkeeping inside this block, and tasks created in it, stay out of the trace.""" token = GUARD.set(True) try: yield finally: GUARD.reset(token)
def text(value: object) -> str: return ( os.fsdecode(value) if isinstance(value, (str, bytes, os.PathLike)) else str(value) )
def command(args: object) -> str: """The command as a person would type it: no `sh -c` wrapper, no store path on argv[0].""" if not isinstance(args, (list, tuple)): return text(args) argv = [text(arg) for arg in args] if not argv: return "" script = albedo_shell.shell_script(argv) if script is not None: return script name = os.path.basename(argv[0]) if os.path.isabs(argv[0]) else argv[0] return shlex.join([name, *argv[1:]])
class Trace: def __init__(self): self.activities: dict[tuple[str, str], dict[str, str]] = {} self.before: dict[str, str | None] = {} self.truncated: bool = False
def activity(self, kind: str, target: object) -> None: if len(self.activities) >= 64: self.truncated = True else: target = str(target)[:1000] self.activities[(kind, target)] = {"kind": kind, "target": target}
def renamed(self, source: str, destination: str) -> None: if source == destination: return # Atomic editors write a temporary file then replace the real target. # The temporary file is not an edit; snapshot the destination before rename. self.writing(destination) self.before.pop(source, None)
def writing(self, path: str) -> None: if path in self.before: return if len(self.before) >= 16: self.truncated = True return self.before[path] = snapshot(path)
def finish(self) -> dict[str, object]: with unobserved(): changes: list[dict[str, str | int]] = [] for path, before in self.before.items(): after = snapshot(path) if before == after: continue if before is None or after is None: changes.append( { "path": path[:1000], "kind": "unavailable", "reason": "binary, unreadable, or larger than 128 KiB", } ) continue lines = list( difflib.unified_diff( before.splitlines(True), after.splitlines(True), fromfile=path, tofile=path, ) ) diff = "".join(lines) self.truncated |= len(diff) > 16000 changes.append( { "path": path[:1000], "kind": "diff", "diff": diff[:16000], "added": sum( line.startswith("+") and not line.startswith("+++") for line in lines ), "removed": sum( line.startswith("-") and not line.startswith("---") for line in lines ), } ) return { "activities": list(self.activities.values()), "changes": changes, "truncated": self.truncated, }
def snapshot(path: str) -> str | None: try: if not os.path.exists(path): return "" if not os.path.isfile(path) or os.path.getsize(path) > LIMIT: return None with open(path, "rb") as file: value = file.read(LIMIT + 1) return ( value.decode("utf-8") if len(value) <= LIMIT and b"\0" not in value else None ) except (OSError, UnicodeError): return None
class TracedCapture(Protocol): trace: Trace
def _no_capture() -> None: return None
current: Callable[[], TracedCapture | None] = _no_capture
def note(kind: str, target: str) -> None: """Record what a harness tool did for the running cell, as it was asked for: the command run() started, the pattern files.find searched.""" capture = current() if capture is not None: capture.trace.activity(kind, target)
def install(get_capture: Callable[[], TracedCapture | None]) -> None: global current current = get_capture
def audit(event: str, args: tuple[object, ...]) -> None: capture = get_capture() if capture is None or GUARD.get(): return token = GUARD.set(True) try: if event == "open" and isinstance(args[0], (str, bytes)): path = os.path.abspath(os.fsdecode(args[0])) if path == "/proc" or path.startswith("/proc/"): return # process supervision is harness bookkeeping, not user exploration flags = args[2] if not isinstance(flags, int): raise TypeError("invalid open flags") if flags & ( os.O_WRONLY | os.O_RDWR | os.O_CREAT | os.O_TRUNC | os.O_APPEND ): capture.trace.writing(path) else: capture.trace.activity("read", path) elif event in ("os.rename", "os.replace") and len(args) >= 2: source, destination = ( os.path.abspath(os.fsdecode(cast(str | bytes, path))) for path in args[:2] ) capture.trace.renamed(source, destination) elif event in ("os.listdir", "os.scandir"): path = os.fsdecode(cast(str | bytes, args[0])) if args else "." if path != "/proc" and not path.startswith("/proc/"): capture.trace.activity("list", path) elif event == "subprocess.Popen": capture.trace.activity("run", command(args[1])) except Exception: capture.trace.truncated = True finally: GUARD.reset(token)
sys.addaudithook(audit)