Skip to content

Instantly share code, notes, and snippets.

@lukestanley
Created September 16, 2026 00:58
Show Gist options
  • Select an option

  • Save lukestanley/e3a1e07c412893707fcf6b8e2899286a to your computer and use it in GitHub Desktop.

Select an option

Save lukestanley/e3a1e07c412893707fcf6b8e2899286a to your computer and use it in GitHub Desktop.
Single HTML file CDN backed SWE Agent with Recursive Language Model (RLM) implementation
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Glass agent v4 - Luke Stanley</title>
<script type="module" src="https://pyscript.net/releases/2026.6.1/core.js"></script>
<style>
textarea { width: 100%; box-sizing: border-box; }
</style>
</head>
<body>
<h1>Glass agent</h1>
<p>A minimal ReAct harness with optional RLM mode. Default LLM model endpoint and model may need changing. Uses PyScript over CDN and just-bash.</p>
<details>
<summary>Model settings</summary>
<fieldset id="settings">
<p><label>Base URL <input id="endpoint" size="45"></label></p>
<p><label>Model <input id="model" size="30"></label></p>
<label>Headers (JSON)<textarea id="headers" rows="2">{}</textarea></label>
<label>Parameter overrides (JSON)<textarea id="parameters" rows="2">{}</textarea></label>
<p>Empty overrides use the model's configured defaults. Headers stay in page memory.</p>
</fieldset>
</details>
<p>
<select id="access" aria-label="Folder access">
<option value="read">Read access</option>
<option value="readwrite">Read and write access</option>
</select>
<button id="attach" disabled>Attach folder</button>
<button id="save" disabled>Save folder changes</button>
<input id="folder-input" type="file" webkitdirectory directory multiple hidden>
<span id="folder">No folder attached.</span>
</p>
<p id="folder-help">Attached files appear under <code>/project/shared</code>. Save writes changes back to that folder. Read access allows edits in memory only.</p>
<label>Task<textarea id="task" rows="3" placeholder="What shall we work on?"></textarea></label>
<p><label><input id="rlm" type="checkbox"> RLM: keep context in Python and allow model subcalls</label></p>
<label>Context<textarea id="context" rows="4" placeholder="Text to inspect or work with"></textarea></label>
<p>
<button id="run" disabled>Run</button>
<button id="stop" disabled>Stop</button>
<span id="status" role="status">Loading Python…</span>
</p>
<p>Each run starts a new conversation; files and Python variables persist. Stop resets the worker and discards unsaved changes.</p>
<details>
<summary>Latest model request</summary>
<textarea id="request" rows="10" readonly aria-label="Latest model request"></textarea>
</details>
<textarea id="transcript" rows="24" readonly aria-label="Transcript" placeholder="Model output, actions and observations appear here as plain text."></textarea>
<script id="worker-source" type="text/python">
import asyncio
import base64
import contextlib
import io
import json
import os
import shutil
import zipfile
from pathlib import Path, PurePosixPath
from pyodide.code import eval_code_async
from pyscript import config, sync
import pyodide_js
from just_bash import Bash
from just_bash.fs import ReadWriteFs, ReadWriteFsOptions
WORKSPACE = config["workspace"]
SHARED_FOLDER = WORKSPACE + "/shared"
native_folder = None
folder_writable = False
# Only these four methods need replacing: aiofiles expects threads.
class BrowserFiles(ReadWriteFs):
async def read_file(self, path, encoding="utf-8"):
return self._to_real_path(path).read_text(encoding=encoding)
async def read_file_bytes(self, path):
return self._to_real_path(path).read_bytes()
async def write_file(self, path, content, encoding="utf-8"):
target = self._to_real_path(path)
target.parent.mkdir(parents=True, exist_ok=True)
if isinstance(content, str):
target.write_text(content, encoding=encoding)
else:
target.write_bytes(content)
async def append_file(self, path, content, encoding="utf-8"):
target = self._to_real_path(path)
target.parent.mkdir(parents=True, exist_ok=True)
if isinstance(content, str):
with target.open("a", encoding=encoding) as output:
output.write(content)
else:
with target.open("ab") as output:
output.write(content)
Path(WORKSPACE).mkdir(exist_ok=True)
os.chdir(WORKSPACE)
shell = Bash(
fs=BrowserFiles(ReadWriteFsOptions(root="/")),
cwd=WORKSPACE,
unescape_html=False,
)
async def bash(command):
result = await shell.exec(command)
return {
"stdout": result.stdout,
"stderr": result.stderr,
"exit_code": result.exit_code,
}
async def llm_query(question):
return await sync.llm_query(str(question))
scope = {"bash": bash, "llm_query": llm_query, "context": "", "FINAL": None}
def set_context(text):
scope["context"] = text
scope["FINAL"] = None
async def mount_folder(handle, access):
global native_folder, folder_writable
native_folder = await pyodide_js.mountNativeFS(SHARED_FOLDER, handle)
folder_writable = access == "readwrite"
return "Folder attached at " + SHARED_FOLDER
async def reset_shared_folder():
global native_folder, folder_writable
native_folder = None
folder_writable = False
shared = Path(SHARED_FOLDER)
if shared.exists():
shutil.rmtree(shared)
shared.mkdir(parents=True)
def import_file(relative_path, encoded_data):
relative = PurePosixPath(relative_path)
if relative.is_absolute() or not relative.parts or ".." in relative.parts:
raise ValueError("Invalid imported path: " + relative_path)
target = Path(SHARED_FOLDER).joinpath(*relative.parts)
target.parent.mkdir(parents=True, exist_ok=True)
data = base64.b64decode(encoded_data)
target.write_bytes(data)
return len(data)
def export_folder_zip():
shared = Path(SHARED_FOLDER)
if not shared.exists():
raise FileNotFoundError("No imported folder is available.")
output = io.BytesIO()
with zipfile.ZipFile(output, "w", compression=zipfile.ZIP_DEFLATED) as archive:
for path in sorted(shared.rglob("*")):
if path.is_file():
archive.write(path, path.relative_to(shared).as_posix())
return base64.b64encode(output.getvalue()).decode("ascii")
async def save_folder():
if native_folder is None or not folder_writable:
raise PermissionError("Attach a folder with write access before saving.")
await native_folder.syncfs()
return "Folder changes saved."
async def execute(name, arguments):
code = json.loads(arguments)["code"]
if not isinstance(code, str):
raise ValueError("Tool code must be a string.")
if name == "bash":
return await bash(code)
if name != "python":
raise ValueError("Unknown tool: " + name)
output = io.StringIO()
with contextlib.redirect_stdout(output), contextlib.redirect_stderr(output):
results = await asyncio.gather(
eval_code_async(code, globals=scope),
return_exceptions=True,
)
result = results[0]
return {
"output": output.getvalue(),
"error": str(result) if isinstance(result, BaseException) else None,
"final": None if isinstance(result, BaseException) else scope["FINAL"],
}
async def run_tool(name, arguments):
result = (await asyncio.gather(execute(name, arguments), return_exceptions=True))[0]
if isinstance(result, BaseException):
result = {"error": str(result)}
return json.dumps(result)
__export__ = [
"set_context", "mount_folder", "reset_shared_folder", "import_file",
"export_folder_zip", "save_folder", "run_tool",
]
</script>
<script type="py">
import asyncio
import base64
import json
from js import AbortController, Blob, Object, TextDecoder, Uint8Array, URL, fetch
from pyodide.ffi import create_proxy, to_js
from pyscript import PyWorker, document, window
DEFAULT_ENDPOINT = "http://localhost:11434/v1"
DEFAULT_MODEL = "qwen3.8:27b"
BASH_PACKAGE = "just-bash==0.2.1"
WORKSPACE = "/project"
MAX_STEPS = 16
MAX_MODEL_CALLS = 24
MAX_OBSERVATION_CHARACTERS = 12000
WORKER_START_TIMEOUT_SECONDS = 180
MODEL_TIMEOUT_SECONDS = 180
TOOL_TIMEOUT_SECONDS = 600
SYSTEM = f"""You are a software engineering agent. Inspect, act, observe, verify.
Your workspace is {WORKSPACE}. An attached folder is at {WORKSPACE}/shared.
Bash and Python use the same files. Use ordinary open() and pathlib in Python.
Bash is just-bash: an emulated shell, without native subprocesses or executables.
Python variables persist between actions. Top-level await works.
Use print() for observations and await bash(command) for shell work from Python.
Finish with an ordinary answer, or set FINAL to a string in Python.
Describe what you changed and what you actually checked.
"""
RLM_INSTRUCTIONS = """
The supplied context is external, in the Python variable context.
Inspect and divide it with code. Use await llm_query(question) for useful
subproblems, including selected context in the question. Subcalls return text
and have no tools. Keep them sequential. Store and combine results in Python.
"""
TOOLS = []
for name, description in (
("bash", "Run a just-bash command in the shared workspace."),
("python", "Run persistent Python with top-level await. Print observations."),
):
TOOLS.append({
"type": "function",
"function": {
"name": name,
"description": description,
"parameters": {
"type": "object",
"properties": {"code": {"type": "string"}},
"required": ["code"],
"additionalProperties": False,
},
},
})
def element(identifier):
return document.getElementById(identifier)
def js_object(value):
return to_js(value, dict_converter=Object.fromEntries)
def show(text):
element("transcript").value += str(text)
worker = None
current_task = None
subcall_active = False
controller = None
settings = {}
model_calls = 0
folder_access = None
folder_mode = None
folder_name = None
async def get_worker():
global worker
if worker is None:
element("status").textContent = "Loading execution worker…"
source = element("worker-source").textContent
blob = Blob.new(to_js([source]), js_object({"type": "text/plain"}))
address = URL.createObjectURL(blob)
worker = PyWorker(address, type="pyodide", config={
"packages": [BASH_PACKAGE],
"sync_main_only": True,
"workspace": WORKSPACE,
})
worker.sync.llm_query = create_proxy(model_subcall)
try:
await asyncio.wait_for(worker.ready, WORKER_START_TIMEOUT_SECONDS)
except BaseException:
if worker is not None:
worker.terminate()
worker = None
raise
finally:
URL.revokeObjectURL(address)
return worker
async def stream_events(response):
reader = response.body.getReader()
decoder = TextDecoder.new()
buffer = ""
while True:
chunk = await reader.read()
buffer += decoder.decode() if chunk.done else decoder.decode(
chunk.value, js_object({"stream": True})
)
if chunk.done:
buffer += "\n\n"
buffer = buffer.replace("\r\n", "\n")
while "\n\n" in buffer:
event, buffer = buffer.split("\n\n", 1)
data = "\n".join(
line.removeprefix("data:").lstrip() for line in event.splitlines()
if line.startswith("data:")
)
if data == "[DONE]":
return
if data:
yield json.loads(data)
if chunk.done:
return
async def completion(messages, tools=None, label="Assistant"):
global model_calls
if model_calls >= MAX_MODEL_CALLS:
raise RuntimeError("Model-call limit reached.")
model_calls += 1
payload = {
**settings["parameters"],
"model": settings["model"],
"messages": messages,
"stream": True,
}
if tools is not None:
payload["tools"] = tools
element("request").value = json.dumps(payload, indent=2)
show(f"\n{label} (model call {model_calls})\n")
content = ""
reasoning = ""
calls = {}
finish_reason = None
channel = None
async with asyncio.timeout(MODEL_TIMEOUT_SECONDS):
response = await fetch(settings["endpoint"] + "/chat/completions", js_object({
"method": "POST",
"headers": settings["headers"],
"body": json.dumps(payload),
"signal": controller.signal,
}))
if not response.ok:
raise RuntimeError(f"HTTP {response.status}: {await response.text()}")
async for event in stream_events(response):
if event.get("error"):
raise RuntimeError(str(event["error"]))
if not event.get("choices"):
continue
choice = event["choices"][0]
delta = choice.get("delta", {})
finish_reason = choice.get("finish_reason") or finish_reason
thought = delta.get("reasoning_content") or delta.get("reasoning") or ""
text = delta.get("content") or ""
for name, fragment in (("Reasoning", thought), ("Text", text)):
if fragment:
if channel != name:
show(f"\n{name}\n")
channel = name
show(fragment)
reasoning += thought
content += text
for change in delta.get("tool_calls") or []:
index = change["index"]
if index not in calls:
calls[index] = {
"id": f"call_{model_calls}_{index}",
"type": "function",
"function": {"name": "", "arguments": ""},
}
call = calls[index]
if change.get("id"):
call["id"] = change["id"]
function = change.get("function", {})
call["function"]["name"] += function.get("name") or ""
call["function"]["arguments"] += function.get("arguments") or ""
if finish_reason not in ("stop", "tool_calls"):
raise RuntimeError(f"Incomplete model response: {finish_reason}")
message = {"role": "assistant", "content": content or None}
if reasoning:
message["reasoning_content"] = reasoning
if calls:
message["tool_calls"] = [calls[index] for index in sorted(calls)]
return message
async def model_subcall(question):
global subcall_active
if not settings.get("rlm") or current_task is None or subcall_active:
raise RuntimeError("Subcalls require an active RLM run and must be sequential.")
subcall_active = True
show(f"\nRLM question\n{question}\n")
try:
message = await completion(
[{"role": "user", "content": question}],
label="RLM subcall",
)
return message["content"] or ""
finally:
subcall_active = False
async def agent():
global settings, model_calls, controller
parameters = json.loads(element("parameters").value)
headers = json.loads(element("headers").value)
if not isinstance(parameters, dict) or not isinstance(headers, dict):
raise ValueError("Headers and parameters must be JSON objects.")
if any(name in parameters for name in ("model", "messages", "stream", "tools", "tool_choice")):
raise ValueError("Parameter overrides are for generation settings, not protocol fields.")
settings = {
"endpoint": element("endpoint").value.rstrip("/"),
"model": element("model").value,
"headers": {"Content-Type": "application/json", **headers},
"parameters": parameters,
"rlm": element("rlm").checked,
}
question = element("task").value.strip()
context = element("context").value
if not question:
raise ValueError("Enter a task.")
controller = AbortController.new()
model_calls = 0
executive = await get_worker()
await executive.sync.set_context(context)
prompt = SYSTEM + (RLM_INSTRUCTIONS if settings["rlm"] else "")
supplied_context = (
f"\nExternal context: {len(context)} characters."
if settings["rlm"] else "\nContext:\n" + context
)
messages = [
{"role": "system", "content": prompt},
{"role": "user", "content": question + supplied_context},
]
show(f"\nUser\n{question}\n")
for step in range(MAX_STEPS):
element("status").textContent = f"Step {step + 1}"
message = await completion(messages, TOOLS)
messages.append(message)
if not message.get("tool_calls"):
return "Finished."
for call in message["tool_calls"]:
function = call["function"]
name = function["name"]
arguments = json.loads(function["arguments"])
show(f"\n{name}\n{arguments.get('code', arguments)}\n")
raw = await asyncio.wait_for(
executive.sync.run_tool(name, function["arguments"]),
TOOL_TIMEOUT_SECONDS,
)
observation = json.loads(raw)
text = json.dumps(observation, ensure_ascii=False)
show("\nObservation\n")
for field, value in observation.items():
if value is not None and value != "":
show(f"{field}: {value}\n")
if observation.get("final") is not None:
show(f"\nFinal\n{observation['final']}\n")
return "Finished."
if len(text) > MAX_OBSERVATION_CHARACTERS:
text = text[:MAX_OBSERVATION_CHARACTERS] + "\n[Clipped; inspect a smaller slice.]"
messages.append({"role": "tool", "tool_call_id": call["id"], "content": text})
return "Step limit reached."
def native_folder_picker_available():
return hasattr(window, "showDirectoryPicker")
def update_folder_ui():
if native_folder_picker_available():
element("attach").textContent = "Attach folder"
element("save").textContent = "Save folder changes"
element("access").options.item(1).textContent = "Read and write access"
element("folder-help").innerHTML = (
'Attached files appear under <code>/project/shared</code>. '
'Save writes changes back to that folder. Read access allows edits in memory only.'
)
else:
element("attach").textContent = "Import folder copy"
element("save").textContent = "Download folder changes (.zip)"
element("access").options.item(1).textContent = "Read and download changes"
element("folder-help").innerHTML = (
'This browser cannot grant direct folder access. Import copies the selected files into '
'<code>/project/shared</code>. With read and write access, Save downloads the changed copy as a ZIP.'
)
def controls(busy):
for identifier in ("run", "attach", "access", "settings", "task", "context", "rlm"):
element(identifier).disabled = busy
element("attach").disabled = busy or worker is None or folder_access is not None
element("save").disabled = busy or folder_access != "readwrite"
element("stop").disabled = not busy
async def perform(operation):
global current_task
controls(True)
result = (await asyncio.gather(operation, return_exceptions=True))[0]
if isinstance(result, BaseException):
if controller is not None:
controller.abort()
show(f"\nError\n{result}\n")
if isinstance(result, TimeoutError):
stop()
return
element("status").textContent = "Stopped with an error."
else:
element("status").textContent = str(result)
current_task = None
controls(False)
def launch(operation):
global current_task
current_task = asyncio.create_task(perform(operation))
def stop(event=None):
global worker, current_task, folder_access, folder_mode, folder_name
if controller is not None:
controller.abort()
if current_task is not None:
current_task.cancel()
current_task = None
if worker is not None:
worker.terminate()
worker = None
folder_access = None
folder_mode = None
folder_name = None
element("folder-input").value = ""
element("folder").textContent = "No folder attached."
element("status").textContent = "Stopped. Worker reset; unsaved changes discarded."
controls(False)
def choose_folder(event):
access = element("access").value
if native_folder_picker_available():
selection = window.showDirectoryPicker(js_object({"mode": access}))
launch(attach(selection, access))
else:
element("folder-input").click()
def choose_imported_folder(event):
file_list = event.target.files
if not file_list.length:
return
files = [file_list.item(index) for index in range(file_list.length)]
launch(import_folder(files, element("access").value))
def imported_path(file):
relative_path = str(file.webkitRelativePath or file.name).replace("\\", "/")
parts = [part for part in relative_path.split("/") if part]
if not parts:
raise ValueError("Selected file has no usable path.")
root = parts[0] if len(parts) > 1 else None
path = "/".join(parts[1:] if root else parts)
return root, path
async def attach(selection, access):
global folder_access, folder_mode, folder_name
handle = await selection
result = await worker.sync.mount_folder(handle, access)
folder_access = access
folder_mode = "native"
folder_name = str(handle.name)
element("folder").textContent = f"{result} ({access})"
return result
async def import_folder(files, access):
global folder_access, folder_mode, folder_name
await worker.sync.reset_shared_folder()
roots = set()
imported = 0
for index, file in enumerate(files, start=1):
root, relative_path = imported_path(file)
if root:
roots.add(root)
data = Uint8Array.new(await file.arrayBuffer())
encoded_data = base64.b64encode(bytes(data.to_py())).decode("ascii")
await worker.sync.import_file(relative_path, encoded_data)
imported += 1
if index % 25 == 0:
element("status").textContent = f"Importing files… {index}/{len(files)}"
folder_access = access
folder_mode = "copy"
folder_name = next(iter(roots)) if len(roots) == 1 else "shared"
element("folder").textContent = (
f"Imported copy of {folder_name}: {imported} files at {WORKSPACE}/shared ({access})"
)
return f"Imported {imported} files."
def download_name():
name = folder_name or "shared"
safe = "".join(character if character.isalnum() or character in "._-" else "_" for character in name)
safe = safe.strip("._") or "shared"
return safe + "-glass-agent.zip"
async def download_imported_folder():
encoded = await worker.sync.export_folder_zip()
binary = window.atob(encoded)
data = Uint8Array.new(len(binary))
for index, character in enumerate(binary):
data[index] = ord(character)
blob = Blob.new(to_js([data]), js_object({"type": "application/zip"}))
address = URL.createObjectURL(blob)
link = document.createElement("a")
link.href = address
link.download = download_name()
document.body.appendChild(link)
link.click()
link.remove()
await asyncio.sleep(0)
URL.revokeObjectURL(address)
return "Folder changes downloaded as " + link.download
async def save():
if folder_mode == "native":
return await worker.sync.save_folder()
if folder_mode == "copy":
return await download_imported_folder()
raise RuntimeError("No folder is attached or imported.")
async def prepare():
await get_worker()
update_folder_ui()
return "Ready."
element("endpoint").value = DEFAULT_ENDPOINT
element("model").value = DEFAULT_MODEL
element("run").onclick = create_proxy(lambda event: launch(agent()))
element("stop").onclick = create_proxy(stop)
element("attach").onclick = create_proxy(choose_folder)
element("folder-input").onchange = create_proxy(choose_imported_folder)
element("save").onclick = create_proxy(lambda event: launch(save()))
launch(prepare())
</script>
</body>
</html>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment