Something went wrong. Try again.
small gleam coding and persistent agent daemon with a detachable cli and a canvas ui forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805"""Human shell execution stays durable, bounded, and separate from model turns."""
from concurrent.futures import ThreadPoolExecutorimport jsonimport osfrom pathlib import Pathimport shleximport shutilimport sqlite3import sysimport threadingimport unittestimport urllib.error
from provider_events import responses_events
from harness import ( Albedo, Provider, Reply, exclusive, error, operation_id, python, release_fifo, text, wait_until,)
def python_command(source): return shlex.join([sys.executable, "-c", source])
def started_response(value, started, release): """Flush response.created, then hold the rest of a successful stream.""" reply = text(value)
def events(): stream = responses_events(reply, "", "fixture-call", 20) yield next(stream) started.set() if not release.wait(20): raise AssertionError("fixture never released the started response") yield from stream
return Reply("text", events=events())
def process_running(pid): try: os.kill(pid, 0) stat = Path(f"/proc/{pid}/stat") return not stat.exists() or stat.read_text().rsplit(")", 1)[1].split()[0] != "Z" except (ProcessLookupError, FileNotFoundError): return False
@exclusiveclass ShellCommandTests(unittest.TestCase): # Shell startup reads HOME and SHELL: every daemon needs a private environment. def setUp(self): self.script = lambda _: text("reported") self.provider = Provider(lambda request: self.script(request)) self.addCleanup(self.provider.close)
def prepare(app): app.daemon.env["HOME"] = str(app.home) app.daemon.env["SHELL"] = shutil.which("bash") for key in ("BASH_ENV", "ENV", "ZDOTDIR"): app.daemon.env.pop(key, None) (app.home / ".bash_profile").write_text( "export BANG_LOGIN_PROFILE=loaded\n" )
self.app = Albedo(self.provider, protocol="responses", prepare=prepare) self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None) self.session = self.app.session()
def url(self, command_id): return f"/sessions/{self.session}/shell-commands/{command_id}"
def put(self, command_id, command): with self.app.api( self.url(command_id), {"command": command}, method="PUT" ) as response: return json.load(response)["shell_command"]
def read(self, command_id): with self.app.api(self.url(command_id)) as response: return json.load(response)["shell_command"]
def done(self, command_id, *, committed=True): observed = []
def ready(): observed[:] = [self.read(command_id)] command = observed[0] return command["status"] != "running" and ( not committed or command["history_seq"] is not None )
wait_until(ready, 15, f"shell command {command_id} did not finish") return observed[0]
def shell_history(self): return [ item for item in self.app.history(self.session)["items"] if item["kind"] == "shell_command" ]
def output_page(self, command_id, offset=0, limit=65536): with self.app.api( self.url(command_id) + f"/output?offset={offset}&limit={limit}" ) as response: return json.load(response)
def test_login_shell_cwd_stdin_and_exit_status_without_model_generation(self): command_id = operation_id() command = ( 'printf "%s\\n" "$PWD" "$BANG_LOGIN_PROFILE"; ' + python_command( 'import os,sys; print("tty",os.isatty(0)); ' 'print("stdin",len(sys.stdin.buffer.read()))' ) + '; printf "stderr-marker\\n" >&2; exit 7' ) self.put(command_id, command) result = self.done(command_id) self.assertEqual(result["status"], "completed") self.assertEqual(result["exit_code"], 7) self.assertIn(str(self.app.workspace), result["output"]) self.assertIn("loaded", result["output"]) self.assertIn("tty False", result["output"]) self.assertIn("stdin 0", result["output"]) self.assertIn("stderr-marker", result["output"]) self.assertEqual(self.provider.requests, []) self.assertEqual(len(self.shell_history()), 1) self.assertEqual(self.app.idle(self.session)["status"]["phase"], "idle")
def test_result_is_in_next_model_request_without_fabricated_tool_calls(self): command_id = operation_id() command = "printf 'user-shell-foobar\\n'" self.put(command_id, command) result = self.done(command_id) history = self.shell_history() self.assertEqual(len(history), 1) self.assertEqual(history[0]["position"], result["history_seq"]) metadata = next( part["value"] for part in history[0]["content"] if part["kind"] == "json" and part["field"] == "shell_command" ) self.assertEqual(metadata["id"], command_id) self.assertEqual(metadata["command"], command) self.assertEqual(self.provider.requests, []) self.app.prompt(self.session, "What did the user command print?").close() self.app.idle(self.session) sent = json.dumps(self.provider.requests[-1]["request"]["input"]) self.assertIn("user-shell-foobar", sent) self.assertIn("printf", sent) self.assertNotIn("function_call_output", sent)
def test_model_reads_output_beyond_its_context_excerpt_through_read_only_binding( self, ): command_id = operation_id() self.put( command_id, python_command( 'import sys; sys.stdout.write("h"*65536+"read-through-daemon")' ), ) self.done(command_id)
def script(request): if any( item.get("type") == "function_call_output" for item in request["input"] ): return text("read") return python( f"page = await output.read_shell({command_id!r}, offset=65536); " "assert page.next_offset is None; print(page.text)" )
self.script = script self.app.prompt(self.session, "Read the rest of that command output.").close() self.app.idle(self.session) sent = json.dumps(self.provider.requests[-1]["request"]["input"]) self.assertIn("read-through-daemon", sent) self.assertEqual(len(self.provider.requests), 2)
def test_same_identity_retries_execute_once_and_changed_intent_conflicts(self): command_id = operation_id() command = python_command( 'from pathlib import Path; p=Path("effect"); p.open("a").write("x")' ) with ThreadPoolExecutor(max_workers=6) as executor: results = list( executor.map(lambda _: self.put(command_id, command), range(6)) ) self.done(command_id) self.assertTrue(all(item["id"] == command_id for item in results)) self.assertEqual((self.app.workspace / "effect").read_text(), "x") with self.assertRaises(urllib.error.HTTPError) as failure: self.put(command_id, command + "; printf changed") self.assertEqual(failure.exception.code, 409) self.assertEqual(len(self.shell_history()), 1)
def test_completion_during_final_provider_response_gets_a_followup_step(self): entered, release = threading.Event(), threading.Event() self.addCleanup(release.set)
def script(_request): entered.set() if not release.wait(15): raise AssertionError("fixture never released model turn") return text("current-agent-turn")
self.script = script self.app.prompt(self.session, "Hold this response.").close() self.assertTrue(entered.wait(5)) try: command_id = operation_id() self.put(command_id, "printf 'parallel-shell-result\\n'") result = self.done(command_id, committed=False) self.assertIsNone(result["history_seq"]) self.assertEqual(result["delivery"], "pending") self.assertEqual(self.shell_history(), []) with self.app.api(f"/sessions/{self.session}?tail=0") as response: snapshot = json.load(response) self.assertIn( command_id, [item["id"] for item in snapshot["shell_commands"]] ) reset = self.app.stream_page(self.session, tail=0) self.assertIn(command_id, json.dumps(reset)) self.assertEqual(len(self.provider.requests), 1) finally: release.set() self.app.idle(self.session) result = self.done(command_id) items = self.app.history(self.session)["items"] assistant = next(item for item in items if item["kind"] == "assistant") self.assertLess(assistant["position"], result["history_seq"]) self.assertEqual(result["delivery"], "delivered") self.assertEqual(len(self.provider.requests), 2) followup = json.dumps(self.provider.requests[1]["request"]["input"]) self.assertIn("parallel-shell-result", followup)
def test_second_running_command_is_refused_and_stop_reaps_descendants(self): command_id = operation_id() source = ( "import os,subprocess,sys,time; from pathlib import Path; " 'child=subprocess.Popen([sys.executable,"-c","import time;time.sleep(120)"]); ' 'Path("pids").write_text(f"{os.getpid()} {child.pid}"); ' 'print("started",flush=True); time.sleep(120)' ) self.put(command_id, python_command(source)) marker = self.app.workspace / "pids" wait_until(marker.exists, 10, "shell process did not start") pids = list(map(int, marker.read_text().split())) with self.assertRaises(urllib.error.HTTPError) as failure: self.put(operation_id(), "printf should-not-run") self.assertEqual(failure.exception.code, 409) with self.app.api(self.url(command_id) + "/stop", {}) as response: json.load(response) result = self.done(command_id) self.assertEqual(result["status"], "cancelled") wait_until( lambda: not any(map(process_running, pids)), 10, "shell descendants survived stop", ) self.assertEqual(self.provider.requests, [])
def test_utf8_output_pages_are_complete_and_terminal_controls_are_inert(self): command_id = operation_id() visible = "α🌙z" * 12 source = f'import sys; sys.stdout.write({visible!r}); sys.stdout.flush(); sys.stdout.buffer.write(b"\\x1b]52;c;Zm9v\\x07")' self.put(command_id, python_command(source)) result = self.done(command_id) pieces, offset = [], 0 while True: page = self.output_page(command_id, offset, 7) pieces.append(page["text"]) if page["next_offset"] is None: break self.assertGreater(page["next_offset"], offset) offset = page["next_offset"] full = "".join(pieces) self.assertIn(visible, full) self.assertNotIn("\x1b", full) self.assertNotIn("\x07", full) self.assertEqual(result["output_bytes"], len(full.encode())) self.assertFalse(result["truncated"])
def test_large_capture_is_bounded_but_retained_end_is_reachable(self): command_id = operation_id() command = python_command('import sys; sys.stdout.write("x"*(17*1024*1024))') self.put(command_id, command) result = self.done(command_id) self.assertTrue(result["truncated"]) self.assertEqual(result["output_bytes"], 16 * 1024 * 1024) self.assertLessEqual(len(result["output"].encode()), 65536) tail = self.output_page(command_id, result["output_bytes"] - 32) self.assertEqual(tail, {"text": "x" * 32, "next_offset": None}) self.app.prompt(self.session, "Inspect the command output.").close() self.app.idle(self.session) sent = json.dumps(self.provider.requests[-1]["request"]["input"]) self.assertNotIn("x" * 20000, sent)
def test_fork_preserves_capture_and_rewind_keeps_retry_receipt_without_reexecution( self, ): self.app.prompt(self.session, "before shell").close() self.app.idle(self.session) first = next( item for item in self.app.history(self.session)["items"] if item["kind"] == "user" ) command_id = operation_id() command = "printf x >> fork-effect; printf forked-shell-output" self.put(command_id, command) self.done(command_id) self.app.prompt(self.session, "after shell").close() self.app.idle(self.session) last = next( item for item in reversed(self.app.history(self.session)["items"]) if item["kind"] == "user" ) branch = operation_id() with self.app.api( f"/sessions/{branch}", { "kind": "fork", "source_session_id": self.session, "checkpoint_id": last["checkpoint_id"], }, method="PUT", headers={"If-None-Match": "*"}, ) as response: json.load(response) copied = [ item for item in self.app.history(branch)["items"] if item["kind"] == "shell_command" ] self.assertEqual(len(copied), 1) with self.app.api( f"/sessions/{branch}/shell-commands/{command_id}/output" ) as response: self.assertEqual(json.load(response)["text"], "forked-shell-output") self.app.prompt(branch, "What did the shell print?").close() self.app.idle(branch) self.assertIn( "forked-shell-output", json.dumps(self.provider.requests[-1]["request"]["input"]), ) with self.app.api( f"/sessions/{self.session}/rewind", { "checkpoint_id": first["checkpoint_id"], }, ) as response: json.load(response) self.assertEqual(self.shell_history(), []) recovered = self.put(command_id, command) self.assertEqual(recovered["status"], "completed") self.assertIsNone(recovered["history_seq"]) self.assertEqual(recovered["delivery"], "detached") self.assertEqual(self.shell_history(), []) self.assertEqual((self.app.workspace / "fork-effect").read_text(), "x") self.assertEqual(len(self.provider.requests), 3)
def test_plain_message_or_assistant_bang_text_is_not_executable(self): self.script = lambda _: text("!touch assistant-must-not-run") self.app.prompt(self.session, "!touch user-message-must-not-run").close() self.app.idle(self.session) self.assertFalse((self.app.workspace / "assistant-must-not-run").exists()) self.assertFalse((self.app.workspace / "user-message-must-not-run").exists()) self.assertEqual(self.shell_history(), []) self.assertEqual(len(self.provider.requests), 1)
def test_capture_storage_failure_stops_effects_and_records_failed_outcome(self): def fail_capture(_app): with sqlite3.connect(self.app.home / "albedo.sqlite") as db: db.execute( "CREATE TRIGGER bang_shell_capture_failure BEFORE INSERT ON shell_command_output " "BEGIN SELECT RAISE(ABORT, 'simulated capture failure'); END" )
self.app.restart(prepare=fail_capture) command_id = operation_id() self.put( command_id, python_command( "import os,time; from pathlib import Path; " 'Path("pid").write_text(str(os.getpid())); print("capture",flush=True); ' "time.sleep(120)" ), ) marker = self.app.workspace / "pid" wait_until(marker.exists, 10, "shell process did not start") pid = int(marker.read_text()) result = self.done(command_id) self.assertEqual(result["status"], "failed") self.assertTrue(result["error"]) wait_until( lambda: not process_running(pid), 10, "command survived capture persistence failure", ) self.assertEqual(self.provider.requests, [])
def test_session_deletion_reaps_command_and_removes_output(self): command_id = operation_id() self.put( command_id, python_command( "import os,time; from pathlib import Path; " 'Path("pid").write_text(str(os.getpid())); print("retained",flush=True); ' "time.sleep(120)" ), ) marker = self.app.workspace / "pid" wait_until(marker.exists, 10, "shell process did not start") pid = int(marker.read_text()) resource = f"/sessions/{self.session}?view=configuration" with self.app.api(resource) as response: validator = response.getheader("ETag") response.read() with self.app.api( resource, method="DELETE", headers={"If-Match": validator} ) as response: self.assertEqual(json.load(response)["state"], "complete") wait_until( lambda: not process_running(pid), 10, "shell process survived session deletion", ) with self.assertRaises(urllib.error.HTTPError) as failure: self.read(command_id) self.assertEqual(failure.exception.code, 404)
def test_graceful_restart_stops_command_without_rerunning_it(self): command_id = operation_id() command = python_command( "import os,time; from pathlib import Path; " 'Path("effect").open("a").write("x"); ' 'Path("pid").write_text(str(os.getpid())); time.sleep(120)' ) self.put(command_id, command) marker = self.app.workspace / "pid" wait_until(marker.exists, 10, "shell process did not start") pid = int(marker.read_text()) self.app.restart() recovered = self.done(command_id, committed=False) self.assertIn(recovered["status"], ("cancelled", "interrupted")) self.app.idle(self.session) # reconnect materializes the idle actor result = self.done(command_id) self.assertIn(result["status"], ("cancelled", "interrupted")) self.assertEqual(self.put(command_id, command)["status"], result["status"]) self.assertEqual((self.app.workspace / "effect").read_text(), "x") wait_until( lambda: not process_running(pid), 10, "shell process survived daemon shutdown", ) self.assertEqual(self.provider.requests, [])
def test_daemon_crash_interrupts_command_without_replaying_its_effect(self): command_id = operation_id() command = python_command( "import os,time; from pathlib import Path; " 'Path("effect").open("a").write("x"); ' 'Path("pid").write_text(str(os.getpid())); time.sleep(120)' ) self.put(command_id, command) marker = self.app.workspace / "pid" wait_until(marker.exists, 10, "shell process did not start") pid = int(marker.read_text()) self.app.restart(crash=True) self.assertEqual( self.done(command_id, committed=False)["status"], "interrupted" ) self.app.idle(self.session) # reconnect materializes the idle actor result = self.done(command_id) self.assertEqual(result["status"], "interrupted") self.assertEqual(self.put(command_id, command)["status"], "interrupted") self.assertEqual((self.app.workspace / "effect").read_text(), "x") wait_until( lambda: not process_running(pid), 10, "shell process survived daemon crash" ) self.assertEqual(len(self.shell_history()), 1) self.assertEqual(self.provider.requests, [])
def test_pending_receipt_survives_restart_then_delivers_without_rerunning(self): command_id = operation_id() command = "printf x >> receipt-effect; printf '%s%s' restart-receipt- 416" self.put(command_id, command) self.assertEqual(self.done(command_id)["delivery"], "pending") self.app.restart(crash=True) self.app.idle(self.session) self.assertEqual(self.read(command_id)["delivery"], "pending") self.assertEqual(self.provider.requests, []) self.app.prompt(self.session, "Read the recovered shell receipt.").close() self.app.idle(self.session) self.assertEqual(self.read(command_id)["delivery"], "delivered") sent = json.dumps(self.provider.requests[-1]["request"]["input"]) self.assertEqual(sent.count("restart-receipt-416"), 1) self.app.restart(crash=True) self.app.idle(self.session) self.assertEqual(self.put(command_id, command)["delivery"], "delivered") self.assertEqual((self.app.workspace / "receipt-effect").read_text(), "x") self.assertEqual(len(self.shell_history()), 1) self.assertEqual(len(self.provider.requests), 1)
class ShellDeliveryTests(unittest.TestCase): """Shell receipts join model context at a safe point, not at turn end."""
url = ShellCommandTests.url put = ShellCommandTests.put read = ShellCommandTests.read done = ShellCommandTests.done shell_history = ShellCommandTests.shell_history
def setUp(self): self.script = lambda _: text("done") self.provider = Provider(lambda request: self.script(request)) self.addCleanup(self.provider.close) self.app = Albedo(self.provider, protocol="responses") self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None) self.session = self.app.session()
def delivery(self, command_id, expected): wait_until( lambda: self.read(command_id)["delivery"] == expected, 20, f"shell {command_id} was not {expected}", ) return self.read(command_id)
def gate_cell(self): fifo = self.app.workspace / "cell-gate" os.mkfifo(fifo) self.addCleanup(release_fifo, fifo) started = self.app.workspace / "cell-started" child = f"open({str(fifo)!r}, 'rb').read(1)" code = ( f"from pathlib import Path\nPath({str(started)!r}).touch()\n" f"import sys\nj = run(sys.executable, '-c', {child!r})\n" "await j\nprint('cell finished')" ) return fifo, started, code
def test_completed_shell_is_delivered_after_cell_before_final_answer(self): fifo, started, code = self.gate_cell() second_request, finish = threading.Event(), threading.Event() self.addCleanup(finish.set) seen = []
def script(request): seen.append(request) if len(seen) == 1: return python(code) return started_response("answer after shell", second_request, finish)
self.script = script self.app.prompt(self.session, "Start a gated cell.").close() wait_until(started.exists, 20, "model cell never started") cursor = self.app.stream_page(self.session, tail=0) command_id = operation_id() self.put(command_id, "printf '%s%s' shell-receipt- 7351; printf '%020000d' 0") result = self.done(command_id, committed=False) self.assertEqual(result["delivery"], "pending") self.assertEqual(len(self.provider.requests), 1) with self.app.api(f"/sessions/{self.session}?tail=0") as response: snapshot = json.load(response) record = next(c for c in snapshot["shell_commands"] if c["id"] == command_id) self.assertEqual(record["delivery"], "pending") self.app.prompt(self.session, "steering-between-receipts").close() later_id = operation_id() self.put(later_id, "printf '%s%s' later-fifo-receipt- 481") self.assertEqual(self.done(later_id, committed=False)["delivery"], "pending") wait_until(lambda: release_fifo(fifo), 20, "cell never opened its gate") self.assertTrue(second_request.wait(20), "next model step never started") sent = json.dumps(seen[1]["input"]) self.assertEqual(sent.count("shell-receipt-7351"), 1, sent) self.assertEqual(sent.count("steering-between-receipts"), 1, sent) self.assertEqual(sent.count("later-fifo-receipt-481"), 1, sent) self.assertLess( sent.index("shell-receipt-7351"), sent.index("steering-between-receipts") ) self.assertLess( sent.index("steering-between-receipts"), sent.index("later-fifo-receipt-481"), ) self.assertIn("Human shell command", sent) self.assertNotIn("0" * 20000, sent) delivered = self.delivery(command_id, "delivered") self.assertIsNotNone(delivered["history_seq"]) events = self.app.stream_page(self.session, after=cursor) shell_events = [ event["data"]["shell_command"] for event in events["events"] if event.get("type") == "shell_command" and event.get("data", {}).get("shell_command", {}).get("id") == command_id ] self.assertIn("pending", [item["delivery"] for item in shell_events]) self.assertIn("delivered", [item["delivery"] for item in shell_events]) finish.set() self.app.idle(self.session) history = self.app.history(self.session)["items"] receipt = next(item for item in history if item["kind"] == "shell_command") answer = next( item for item in history if item["kind"] == "assistant" and "answer after shell" in json.dumps(item) ) self.assertLess(receipt["position"], answer["position"]) self.assertEqual(len(self.shell_history()), 2) steering = next( item for item in history if "steering-between-receipts" in json.dumps(item) ) later_receipt = next( item for item in self.shell_history() if item["position"] != receipt["position"] ) self.assertLess(receipt["position"], steering["position"]) self.assertLess(steering["position"], later_receipt["position"]) self.delivery(later_id, "delivered") metadata = next( part["value"] for part in receipt["content"] if part.get("field") == "shell_command" ) self.assertEqual(metadata["delivery"], "delivered") self.assertTrue(receipt.get("turn_id"))
def test_interrupt_keeps_pending_shell_in_history_without_waking_model(self): fifo, started, code = self.gate_cell() self.script = lambda _: python(code) self.app.prompt(self.session, "Start a gated cell to interrupt.").close() wait_until(started.exists, 20, "model cell never started") command_id = operation_id() self.put(command_id, "printf '%s%s' interrupted-receipt- 774") self.assertEqual(self.done(command_id, committed=False)["delivery"], "pending") with self.app.api(f"/sessions/{self.session}?tail=0") as response: snapshot = json.load(response) with self.app.api( f"/sessions/{self.session}/interrupt", { "run_id": snapshot["status"]["run_id"], "through_input_order": snapshot["input_order"], }, ) as response: response.read() self.app.idle(self.session) result = self.done(command_id) self.assertEqual(result["delivery"], "pending") self.assertEqual(len(self.shell_history()), 1) self.assertEqual(len(self.provider.requests), 1) release_fifo(fifo) self.assertEqual(len(self.provider.requests), 1)
def test_fork_preserves_pending_receipt_and_delivers_it_once(self): command_id = operation_id() self.put(command_id, "printf '%s%s' fork-pending-receipt- 328") self.assertEqual(self.done(command_id)["delivery"], "pending") self.script = lambda _: error(400, "keep the receipt pending") self.app.prompt(self.session, "Create a checkpoint after the receipt.").close() self.app.idle(self.session) self.assertEqual(self.read(command_id)["delivery"], "pending") checkpoint = next( item["checkpoint_id"] for item in reversed(self.app.history(self.session)["items"]) if item["kind"] == "user" ) branch = operation_id() with self.app.api( f"/sessions/{branch}", { "kind": "fork", "source_session_id": self.session, "checkpoint_id": checkpoint, }, method="PUT", headers={"If-None-Match": "*"}, ) as response: json.load(response) with self.app.api( f"/sessions/{branch}/shell-commands/{command_id}" ) as response: self.assertEqual( json.load(response)["shell_command"]["delivery"], "pending" ) self.script = lambda _: text("fork accepted receipt") self.app.prompt(branch, "Read the forked shell receipt.").close() self.app.idle(branch) sent = json.dumps(self.provider.requests[-1]["request"]["input"]) self.assertEqual(sent.count("fork-pending-receipt-328"), 1) with self.app.api( f"/sessions/{branch}/shell-commands/{command_id}" ) as response: self.assertEqual( json.load(response)["shell_command"]["delivery"], "delivered" ) self.assertEqual(self.read(command_id)["delivery"], "pending") branch_history = [ item for item in self.app.history(branch)["items"] if item["kind"] == "shell_command" ] self.assertEqual(len(branch_history), 1)
def test_idle_receipts_are_seen_once_and_all_are_acknowledged(self): commands = [] for number in range(33): command_id = operation_id() marker = f"idle-receipt-{number:02d}" self.put(command_id, f"printf '%s%s' idle-receipt- {number:02d}") result = self.done(command_id) self.assertEqual(result["delivery"], "pending") commands.append((command_id, marker)) self.assertEqual(self.provider.requests, []) self.app.prompt(self.session, "Read the idle shell receipts.").close() self.app.idle(self.session) self.assertEqual(len(self.provider.requests), 1) sent = json.dumps(self.provider.requests[0]["request"]["input"]) for command_id, marker in commands: self.assertEqual(sent.count(marker), 1, marker) self.assertEqual( self.delivery(command_id, "delivered")["delivery"], "delivered" ) self.assertEqual(len(self.shell_history()), 33)
def test_rejected_initial_request_does_not_consume_pending_receipt(self): command_id = operation_id() self.put(command_id, "printf '%s%s' rejected-receipt- 239") self.assertEqual(self.done(command_id)["delivery"], "pending") self.script = lambda _: error(400, "fixture deliberately rejected request") self.app.prompt(self.session, "Read that receipt.").close() self.app.idle(self.session) self.assertEqual(self.read(command_id)["delivery"], "pending") self.assertEqual(len(self.shell_history()), 1) self.script = lambda _: text("accepted") self.app.prompt(self.session, "Try the receipt again.").close() self.app.idle(self.session) sent = json.dumps(self.provider.requests[-1]["request"]["input"]) self.assertEqual(sent.count("rejected-receipt-239"), 1) self.delivery(command_id, "delivered") self.assertEqual(len(self.shell_history()), 1)
def test_shell_finishing_during_request_waits_for_following_safe_point(self): fifo, started, code = self.gate_cell() entered, release, second = ( threading.Event(), threading.Event(), threading.Event(), ) finish = threading.Event() self.addCleanup(release.set) self.addCleanup(finish.set) seen = []
def script(request): seen.append(request) if len(seen) == 1: entered.set() if not release.wait(20): raise AssertionError("first request was not released") return python(code) return started_response("finished", second, finish)
self.script = script first, later = operation_id(), operation_id() self.put(first, "printf '%s%s' first-receipt- 901") self.assertEqual(self.done(first)["delivery"], "pending") self.app.prompt(self.session, "Read the first command.").close() self.assertTrue(entered.wait(20), "provider request did not start") self.put(later, "printf '%s%s' later-receipt- 902") self.assertEqual(self.done(later, committed=False)["delivery"], "pending") self.assertNotIn("later-receipt-902", json.dumps(seen[0]["input"])) self.assertEqual(self.read(first)["delivery"], "pending") release.set() wait_until(started.exists, 20, "first response tool did not start") self.delivery(first, "delivered") self.assertEqual(self.read(later)["delivery"], "pending") wait_until(lambda: release_fifo(fifo), 20, "cell never opened its gate") self.assertTrue(second.wait(20), "subsequent request did not start") sent = json.dumps(seen[1]["input"]) self.assertEqual(sent.count("first-receipt-901"), 1) self.assertEqual(sent.count("later-receipt-902"), 1) self.delivery(first, "delivered") self.delivery(later, "delivered") finish.set() self.app.idle(self.session)