""" AI Test Plugin for Agentsd ========================== Manual verification plugin for AI agents and humans. It is intentionally not autoloaded by default; run it only when testing Agentsd runtime behavior. Usage: # Start agentsd first: # RUST_LOG=agentsd=info ./target/release/agentsd.exe python plugins/ai_test/ai_test_plugin.py """ import json import sys import time from concurrent.futures import ThreadPoolExecutor, TimeoutError from pathlib import Path import grpc ROOT = Path(__file__).resolve().parents[2] SDK_DIR = ROOT / "plugins" / "_sdk" sys.path.insert(0, str(SDK_DIR)) import agentsd_pb2 as pb # noqa: E402 import agentsd_pb2_grpc as rpc # noqa: E402 PLUGIN_ID = "ai-test" METADATA = (("x-plugin-id", PLUGIN_ID),) ADDR = "localhost:50051" class CheckRunner: def __init__(self): self.channel = grpc.insecure_channel(ADDR) self.bridge = rpc.PluginBridgeStub(self.channel) self.display = rpc.DisplayStub(self.channel) self.memory = rpc.MemoryStub(self.channel) self.fs = rpc.FSStub(self.channel) self.process = rpc.ProcessStub(self.channel) self.timer = rpc.TimerStub(self.channel) self.network = rpc.NetworkStub(self.channel) self.results = [] self.run_id = str(int(time.time() * 1000)) self.scratch_dir = ROOT / "target" / "release" / "data" / "ai_test" / self.run_id def close(self): self.channel.close() def record(self, name, ok, details=""): icon = "PASS" if ok else "FAIL" print(f"[{icon}] {name}: {details}") self.results.append({"name": name, "ok": bool(ok), "details": details}) def check(self, name, fn): try: details = fn() self.record(name, True, details or "ok") except Exception as exc: self.record(name, False, str(exc)) def register(self): resp = self.bridge.Register(pb.PluginInfo( id=PLUGIN_ID, name="AI Test Plugin", version="0.1.0", plugin_type="basic", syscalls=[ "display.text", "memory.write", "memory.read", "memory.list", "memory.delete", "fs.write", "fs.read", "fs.stat", "fs.list", "fs.rename", "fs.delete", "process.spawn", "process.stdout", "process.wait", "timer.once", ], depends=[], exports=[pb.ExportDef(name="run_checks", description="Run Agentsd verification checks")], )) if not resp.ok: raise RuntimeError(resp.error) return f"session={resp.session_id}" def display_text(self): resp = self.display.Text( pb.DisplayTextRequest(content=f"[ai-test] display check run={self.run_id}"), metadata=METADATA, ) if not resp.ok: raise RuntimeError(resp.error) return "display.text ok" def memory_roundtrip(self): key = f"ai_test.{self.run_id}.greeting" value = f"hello-{self.run_id}".encode("utf-8") resp = self.memory.Write(pb.MemoryWriteRequest(key=key, value=value), metadata=METADATA) if not resp.ok: raise RuntimeError(f"write: {resp.error}") resp = self.memory.Read(pb.MemoryReadRequest(key=key), metadata=METADATA) if not resp.found or resp.value != value: raise RuntimeError(f"read mismatch found={resp.found} value={resp.value!r}") resp = self.memory.List(pb.MemoryListRequest(prefix=f"ai_test.{self.run_id}", limit=10), metadata=METADATA) if key not in resp.keys: raise RuntimeError(f"list missing key; keys={list(resp.keys)} error={resp.error!r}") resp = self.memory.Delete(pb.MemoryKeyRequest(key=key), metadata=METADATA) if not resp.ok: raise RuntimeError(f"delete: {resp.error}") return f"key={key}" def fs_roundtrip(self): base = self.scratch_dir file_a = base / "sample.txt" file_b = base / "renamed.txt" data = f"AI_TEST_FS_OK {self.run_id}".encode("utf-8") resp = self.fs.Write(pb.FsWriteRequest(path=str(file_a), data=data, append=False), metadata=METADATA) if not resp.ok: raise RuntimeError(f"write: {resp.error}") resp = self.fs.Read(pb.FsReadRequest(path=str(file_a), offset=0, length=0), metadata=METADATA) if not resp.ok or resp.data != data: raise RuntimeError(f"read mismatch ok={resp.ok} data={resp.data!r} error={resp.error!r}") resp = self.fs.Stat(pb.FsPathRequest(path=str(file_a)), metadata=METADATA) if not resp.ok or not resp.is_file or resp.size != len(data): raise RuntimeError(f"stat mismatch ok={resp.ok} size={resp.size} error={resp.error!r}") resp = self.fs.List(pb.FsPathRequest(path=str(base)), metadata=METADATA) names = [e.name for e in resp.entries] if not resp.ok or "sample.txt" not in names: raise RuntimeError(f"list mismatch ok={resp.ok} names={names} error={resp.error!r}") resp = self.fs.Rename(pb.FsRenameRequest(**{"from": str(file_a), "to": str(file_b)}), metadata=METADATA) if not resp.ok: raise RuntimeError(f"rename: {resp.error}") resp = self.fs.Delete(pb.FsPathRequest(path=str(base)), metadata=METADATA) if not resp.ok: raise RuntimeError(f"delete: {resp.error}") return str(base) def process_roundtrip(self): code = "print('AI_TEST_PROCESS_OK')" resp = self.process.Spawn(pb.SpawnRequest( cmd=sys.executable, args=["-c", code], env={}, cwd=str(ROOT), ), metadata=METADATA) if not resp.ok: raise RuntimeError(resp.error) pid = resp.pid chunks = [] for chunk in self.process.Stdout(pb.PidRequest(pid=pid), metadata=METADATA): chunks.append(chunk.data) output = b"".join(chunks).decode("utf-8", errors="replace").strip() wait = self.process.Wait(pb.PidRequest(pid=pid), metadata=METADATA) if not wait.ok or wait.exit_code != 0: raise RuntimeError(f"wait failed ok={wait.ok} code={wait.exit_code} error={wait.error!r}") if output != "AI_TEST_PROCESS_OK": raise RuntimeError(f"stdout mismatch: {output!r}") return f"pid={pid} stdout={output}" def timer_event(self): stream = self.bridge.StreamCall( pb.CallRequest(target=PLUGIN_ID, fn="events", args=b""), metadata=METADATA, ) resp = self.timer.Once(pb.TimerOnceRequest(delay_ms=100, callback_id="ai-test-callback"), metadata=METADATA) if not resp.ok: raise RuntimeError(resp.error) with ThreadPoolExecutor(max_workers=1) as executor: future = executor.submit(next, stream) try: event = future.result(timeout=5) except TimeoutError: stream.cancel() raise RuntimeError("timed out waiting for timer event") payload = json.loads(event.result.decode("utf-8")) if payload.get("event") != "timer" or payload.get("callback_id") != "ai-test-callback": raise RuntimeError(f"unexpected payload: {payload}") return json.dumps(payload, ensure_ascii=False) def permission_denial(self): try: self.network.Available(pb.Empty(), metadata=METADATA) except grpc.RpcError as exc: if exc.code() == grpc.StatusCode.PERMISSION_DENIED: return exc.details() raise RuntimeError(f"wrong error: {exc.code().name} {exc.details()}") raise RuntimeError("network.available unexpectedly succeeded") def run(self): print(f"AI Test Plugin connecting to {ADDR}") print(f"Scratch dir: {self.scratch_dir}") self.check("register", self.register) self.check("display.text", self.display_text) self.check("memory roundtrip", self.memory_roundtrip) self.check("fs roundtrip", self.fs_roundtrip) self.check("process roundtrip", self.process_roundtrip) self.check("timer event", self.timer_event) self.check("permission denial", self.permission_denial) ok = all(r["ok"] for r in self.results) summary = {"plugin": PLUGIN_ID, "ok": ok, "results": self.results} print("AI_TEST_SUMMARY " + json.dumps(summary, ensure_ascii=False)) return 0 if ok else 1 def main(): runner = CheckRunner() try: return runner.run() finally: runner.close() if __name__ == "__main__": raise SystemExit(main())