Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
Python
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167"""Real daemon read-only /context snapshot against an observed provider request."""import contextlibimport http.serverimport jsonimport osfrom pathlib import Pathimport subprocessimport tempfileimport threadingimport timeimport urllib.request
ROOT = Path(__file__).resolve().parents[2]
class Provider(http.server.BaseHTTPRequestHandler): requests = []
def log_message(self, *_): pass
catalog = json.dumps({"fixture-cloud": { "id": "fixture-cloud", "name": "Fixture Cloud", "env": ["FIXTURE_API_KEY"], "models": {"fixture": {"id": "fixture", "limit": {"context": 200000, "output": 8000}, "modalities": {"input": ["text"]}}}, }})
def do_GET(self): body = self.catalog.encode() self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body)
def do_POST(self): request = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) self.requests.append(request) response = {"type": "response.completed", "response": { "id": "fixture", "status": "completed", "output": [{ "type": "message", "role": "assistant", "status": "completed", "content": [{"type": "output_text", "text": "done", "annotations": []}], }], "usage": {"input_tokens": 20, "output_tokens": 1}, }} body = ("data: " + json.dumps(response) + "\n\n").encode() self.send_response(200) self.send_header("Content-Type", "text/event-stream") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body)
def run(endpoint): with tempfile.TemporaryDirectory(prefix="albedo-context-") as directory: root = Path(directory) home, workspace, user_home = root/"state", root/"workspace", root/"user" for path in (home, workspace, user_home): path.mkdir(mode=0o700) secret = "credential-must-never-appear-in-inspector" (home/"config.json").write_text(json.dumps({"active": "fixture", "providers": {"fixture": { "baseUrl": endpoint, "apiKey": secret, "model": "fixture", "protocol": "responses", }}})) # The catalog provider is the same local server, so no test reaches the network. (home/"extensions.json").write_text(json.dumps({"models": { "url": endpoint.rsplit("/", 1)[0] + "/models.json", "refreshHours": 24, }})) env = dict(os.environ, HOME=str(user_home), ALBEDO_HOME=str(home), ALBEDO_PARENT_PID=str(os.getpid())) connection = None
def cli(*args): result = subprocess.run(["node", "cli/bin/albedo.mjs", *args], cwd=ROOT, env=env, capture_output=True, text=True, timeout=45) assert result.returncode == 0, result.stdout + result.stderr + (home/"daemon.log").read_text() return result.stdout
def api(path, data=None): request = urllib.request.Request(f"http://127.0.0.1:{connection['port']}" + path, headers={"Authorization": "Bearer " + connection["token"], "Content-Type": "application/json"}, data=None if data is None else json.dumps(data).encode()) with urllib.request.urlopen(request, timeout=25) as response: return json.load(response)
def stop(): if not connection: return with contextlib.suppress(Exception): api("/shutdown", {})
try: cli("sessions") connection = json.loads((home/"daemon.json").read_text()) assert "session_context" in api("/health")["capabilities"] session = json.loads(cli("new", str(workspace)))["session"] route = f"/sessions/{session}/context"
before = api(route) assert before == {"state": "pending", "reason": "runtime session has not prepared a provider request"} assert Provider.requests == [], "inspection must not contact the provider"
api(f"/sessions/{session}/events", {"content": "inspect exact composition"}) deadline = time.monotonic() + 20 while time.monotonic() < deadline and api(f"/sessions/{session}/status")["running"]: time.sleep(.025) assert len(Provider.requests) == 1 sent = Provider.requests[0]
snapshot = api(route) assert snapshot["state"] == "ready" assert snapshot["provider"] == "fixture" and snapshot["model"] == "fixture" # The catalog is stored trimmed to the fields lookup reads. served = json.loads(Provider.catalog)["fixture-cloud"] assert json.loads((home/"models.json").read_text()) == {"fixture-cloud": { "env": served["env"], "models": served["models"]}} assert snapshot["context_window_tokens"] == 200_000, snapshot compaction = snapshot["compaction"] assert compaction["estimate_method"] == "local byte-based estimate; not provider token usage" assert compaction["status"] == "not_needed", compaction assert compaction["input_limit_tokens"] == 200_000, compaction assert compaction["trigger_free_percent"] == 10, compaction assert "fixture-cloud" in compaction["source"] and "models.dev catalog" in compaction["source"], compaction labels = [section["label"] for section in snapshot["sections"]] assert labels[0] == "system instructions" and labels[-2:] == ["prepared conversation", "tool schemas"] assert all(label.startswith("extension context · ") for label in labels[1:-2]) assert all(section["preview"] and len(section["preview"]) <= 180 for section in snapshot["sections"]) assert secret not in json.dumps(snapshot)
pages = {} for section in snapshot["sections"]: page = api(f"{route}/{section['id']}/0") assert len(page["content"].encode()) <= 32_000 pages[section["id"]] = page["content"] assert pages["instructions"] == sent["instructions"] assert json.loads(pages["tools"]) == sent["tools"] assert "inspect exact composition" in pages["history"] assert secret not in json.dumps(pages) assert len(Provider.requests) == 1, "summary/page inspection must not contact the provider"
changed = api(f"/sessions/{session}/commands", {"name": "/model", "args": {"model": "changed-model"}}) assert changed["result"]["model"] == "changed-model" assert api(route) == { "state": "pending", "reason": "runtime session has not prepared a provider request", }, "a snapshot for the old model must not survive model selection" assert len(Provider.requests) == 1, "model invalidation must not contact the provider"
# A model the catalog does not list stays explicitly unknown. api(f"/sessions/{session}/events", {"content": "a model outside the catalog"}) deadline = time.monotonic() + 20 while time.monotonic() < deadline and api(f"/sessions/{session}/status")["running"]: time.sleep(.025) unknown = api(route) assert "context_window_tokens" not in unknown, unknown assert unknown["compaction"]["status"] == "unknown", unknown["compaction"] print("context pending/ready sources, exact schema reuse, bounds, read-only behavior, and model invalidation passed") finally: stop()
if __name__ == "__main__": server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Provider) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() try: run(f"http://127.0.0.1:{server.server_port}/v1") finally: server.shutdown()