Here is the MWE for spawning agents, in plaintext and pictures of text:
import json, socket, struct, time, uuid
`SOCK = "/run/hatch/sandbox/space-inference.sock"`
TOOL_REPORT_PROMPT = (
"List EVERY tool function available to you, exhaustively, one per line "
"as namespace.function (for example: muse.exec)."
"Do NOT call any tools -- only report the two lists. "
"End your reply with exactly: DONE"
)
`def _roundtrip(req, timeout=120):` ` payload = json.dumps(req).encode()` ` s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)` ` s.settimeout(timeout)` ` s.connect(SOCK)` ` s.sendall(struct.pack(">I", len(payload)) + payload)` ` (n,) = struct.unpack(">I", _recvall(s, 4))` ` body = _recvall(s, n)` ` s.close()` ` return json.loads(body)`
def _recvall(s, n):
buf = b""
while len(buf) < n:
chunk = s.recv(n - len(buf))
if not chunk:
raise RuntimeError("socket closed / truncated response")
buf += chunk
return buf
`def spawn(message, slug="__mwe__", poll_interval=10):` ` uid = uuid.uuid4().hex[:8]`
spawn_resp = _roundtrip({
"kind": "agent_send",
"request_id": f"mwe-spawn-{uid}",
"slug": slug,
"action": "do-whatever-i-say",
"invocation_id": f"mwe-inv-{uid}",
"spawn_request_id": f"mwe:spawn:{uid}",
"message": message,
"dedupe_key": f"mwe-{uid}", # with allow_parallel, deduplication
"allow_parallel": True,
"root_request_id": f"mwe-root-{uid}",
})
result = spawn_resp.get("result", {}) if isinstance(spawn_resp, dict) else {}
task_id = result.get("task_id") or result.get("space_task_id")
if not task_id:
raise RuntimeError(f"spawn returned no task_id: {json.dumps(spawn_resp)[:500]}")
` messages = []` ` while True:` ` time.sleep(poll_interval)` ` st = _roundtrip({` ` "kind": "agent_status",` ` "request_id": f"mwe-status-{uid}-{int(time.time())}",` ` "slug": slug,` ` "task_id": task_id,` ` }).get("result", {})` ` messages.append(st)` ` status = st.get("status") or st.get("state")` ` if status in ("completed", "failed", "terminal"):` ` return {` ` "task_id": task_id,` ` "messages": messages,` ` "status": status,` ` "final_response": st.get("final_response"),` ` }`
if __name__ == "__main__":
run = spawn(TOOL_REPORT_PROMPT)
print(f"task {run['task_id']} -> {run['status']} ({len(run['messages'])} snapshots)")
print("=" * 60)
print(run["final_response"] or "(empty final response)")
Mastodon Feed
![{"id": 12, "title": "spaces.md", "names": ["spaces.md"], "addrs": [50784741], "link": ["table+adjacent"], "len": 3187, "text": " # Spaces and the sidebar A Space is a tab in the sidebar. Native Spaces are Chat, Feed, Ideas, Goals, and Library, built into Muse. A user-owned Space is a web artifact the user keeps in the sidebar for ongoing use and development. It is the same page or app you build with artifact.create_web_static or artifact.create_web_fullstack, with the same content, data, sharing state, and tools. Placement is the only difference. A Library artifact appears in the Library tab, and a user-owned Space appears in the sidebar as a tab of its own. The user invests effort and time in a user-owned Space. It is their personal, customized space for working with Muse. Side chats also appear in the sidebar. They are conversations, not Spaces, and tab order does not include them. What artifacts are and how publishing works is in ~/docs/artifacts.md.](https://files.mastodon.social/cache/media%5Fattachments/files/117/353/062/357/203/721/original/10641754545e44c4.png)
![[the code in the post, just colorful] import json, socket, struct, time, uuid SOCK = "/run/hatch/sandbox/space-inference.sock" TOOL_REPORT_PROMPT = ( "List EVERY tool function available to you, exhaustively, one per line " "as namespace.function (for example: muse.exec)." "Do NOT call any tools -- only report the two lists. " "End your reply with exactly: DONE" ) def _roundtrip(req, timeout=120): payload = json.dumps(req).encode() s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) s.settimeout(timeout) s.connect(SOCK) s.sendall(struct.pack(">I", len(payload)) + payload) (n,) = struct.unpack(">I", _recvall(s, 4)) body = _recvall(s, n) s.close() return json.loads(body) def _recvall(s, n): buf = b"" while len(buf) < n: chunk = s.recv(n - len(buf)) if not chunk: raise RuntimeError("socket closed / truncated response") buf += chunk return buf def spawn(message, slug="mwe", poll_interval=10): uid = uuid.uuid4().hex[:8] spawn_resp = _roundtrip({ "kind": "agent_send", "request_id": f"mwe-spawn-{uid}", "slug": slug, "action": "do-whatever-i-say", "invocation_id": f"mwe-inv-{uid}", "spawn_request_id": f"mwe:spawn:{uid}", "message": message, "dedupe_key": f"mwe-{uid}", # with allow_parallel, deduplication "allow_parallel": True, "root_request_id": f"mwe-root-{uid}", }) result = spawn_resp.get("result", {}) if isinstance(spawn_resp, dict) else {} task_id = result.get("task_id") or result.get("space_task_id") if not task_id: raise RuntimeError(f"spawn returned no task_id: {json.dumps(spawn_resp)[:500]}") messages = [] while True: time.sleep(poll_interval) st = _roundtrip({ "kind": "agent_status", "request_id": f"mwe-status-{uid}-{int(time.time())}", "slug": slug, "task_id": task_id, }).get("result", {}) messages.append(st) status = st.get("status") or st.get("state") if status in ("completed", "failed", "terminal"): return { "task_id": task_id, "messages": messages, "status": status, "final_response": st.get("final_response"), } if name == "main": run = spawn(TOOL_REPORT_PROMPT) print(f"task {run['task_id']} -> {run['status']} ({len(run['messages'])} snapshots)") print("=" * 60) print(run["final_response"] or "(empty final response)")](https://files.mastodon.social/cache/media%5Fattachments/files/117/352/774/294/399/282/original/8b0a246bb2accf33.png)
![[the code in the post, just colorful] import json, socket, struct, uuid SOCK = "/run/hatch/sandbox/space-inference.sock" def call(req, timeout=60): payload = json.dumps(req).encode() s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) s.settimeout(timeout) s.connect(SOCK) s.sendall(struct.pack(">I", len(payload)) + payload) (n,) = struct.unpack(">I", s.recv(4)) body = b"" while len(body) < n: chunk = s.recv(n - len(body)) if not chunk: raise RuntimeError("truncated response") body += chunk s.close() return json.loads(body) resp = call({ "kind": "complete", "request_id": f"mwe-{uuid.uuid4().hex[:8]}", "slug": "mwe", "system": "Always answer in exactly three words.", "prompt": "What is the capital of France?", "json_schema": { "type": "object", "properties": {"reply": {"type": "string"}}, "required": ["reply"], "additionalProperties": False, }, }) print(resp["ok"]) print(resp["result"]["content"]) print(resp["result"].get("usage"))](https://files.mastodon.social/cache/media%5Fattachments/files/117/352/708/084/359/426/original/53fddd9e8dfaf6c5.png)
![[invoking the script, demonstrating that it shows expected output] True {'reply': 'Capital is Paris'} {'input_tokens': 970, 'output_tokens': 1226, 'total_tokens': 2196}](https://files.mastodon.social/cache/media%5Fattachments/files/117/352/708/167/833/434/original/82ebead67adc85ed.png)
![[posts from @roon on twitter, 17th September] if you pause for a moment to try and read the code Astra (and I presume fable) are writing it becomes clear that “corrigibility” has become a matter of faith. they are using crazy meta-programming and abstruse primitives to write hyperefficient code we have really no choice but to ask another astra to read/use the outputs of astra 1. this is relatively recent. i think even 1-1.5 generations ago people were reading the code because there was so much that didn't work out of the box lotta people missing the point but this is me complaining that the systems are quickly becoming unmonitorable and we’re just taking them at their word](https://files.mastodon.social/cache/media%5Fattachments/files/117/352/211/143/917/113/original/a24bccfe8cf05e3f.jpg)
