"""A session's name: the model may choose one, the person's own always wins. The model's name is the session's title, kept apart from the person's so that a refusal, an etag and the per-message title all behave as the docs say; only a daemon, its storage and a kernel's `commands` object show all of that at once. """ import json import unittest from harness import Albedo, Provider, latest_user, python, text, wait_until def tool_output(request): """What the cell the model just ran printed.""" last = request["messages"][-1] if last.get("role") != "tool": return None return json.loads(last["content"])["output"] class SessionNameTests(unittest.TestCase): def setUp(self): self.outputs = [] def script(request): output = tool_output(request) if output is not None: self.outputs.append(output) return text("done") prompt = latest_user(request) if "name it" in prompt: return python( "print(await commands.rename('apple pie recipe'))\n" "print(await commands.rename('apple pie recipe'))" ) if "start the agents" in prompt: return python( "models = await agents.models()\n" "await agents.self.spawn('child task rename', name='scout', model=models[0])\n" "await agents.start('peer task rename', 'helper', models[0])\n" ) if "task rename" in prompt: return python( "print(await commands.rename())\n" "try:\n" " await commands.rename('mine')\n" "except CommandsError as error:\n" " print('REFUSED', error)" ) if "who named" in prompt: return python("print(await commands.rename())") if "long name" in prompt or "blank name" in prompt: name = "'x' * 81" if "long" in prompt else "' '" return python( "try:\n" f" await commands.rename({name})\n" "except CommandsError as error:\n" " print('REFUSED', error)" ) if "rename again" in prompt: return python( "try:\n" " await commands.rename('model again')\n" "except CommandsError as error:\n" " print('REFUSED', error)" ) return text("ok") self.provider = Provider(script) self.addCleanup(self.provider.close) def shown(self, app, session): with app.api(f"/sessions/{session}?tail=0") as response: snapshot = json.load(response) return snapshot["name"], snapshot["automatic_name"] def turn(self, app, session, prompt): app.prompt(session, prompt).close() app.idle(session) def configuration(self, app, session): with app.api(f"/sessions/{session}?view=configuration") as response: json.load(response) return response.headers["ETag"] def set_name(self, app, session, name, etag): app.api( f"/sessions/{session}?view=configuration", {"name": name}, method="PATCH", headers={"If-Match": etag}, ).close() def test_the_model_names_a_session_and_the_person_outranks_it(self): with Albedo(self.provider) as app: session = app.session() etag = self.configuration(app, session) self.turn(app, session, "tell me about apples") self.assertEqual(self.shown(app, session)[0], "tell me about apples") self.assertIn("commands.rename", json.dumps(self.provider.requests[0])) # The model names it; the same name again changes nothing. self.turn(app, session, "name it please") named, again = self.outputs[-1].split("\n")[:2] self.assertIn("'source': 'model'", named) self.assertIn("apple pie recipe", again) self.assertEqual(self.shown(app, session), ("apple pie recipe",) * 2) # Later messages do not take the name back. self.turn(app, session, "now bananas") self.assertEqual(self.shown(app, session)[0], "apple pie recipe") # It is no configuration edit: the person's validator still holds. self.assertEqual(self.configuration(app, session), etag) self.turn(app, session, "who named this") self.assertIn("'source': 'model'", self.outputs[-1]) # The person's name wins, and the model is told to leave it. self.set_name(app, session, "My name", etag) self.assertEqual(self.shown(app, session)[0], "My name") self.turn(app, session, "rename again") self.assertIn("REFUSED", self.outputs[-1]) self.assertIn("My name", self.outputs[-1]) self.assertEqual(self.shown(app, session)[0], "My name") self.turn(app, session, "who named this") self.assertIn("'source': 'user'", self.outputs[-1]) # Clearing theirs shows the model's again. self.set_name(app, session, None, self.configuration(app, session)) self.assertEqual(self.shown(app, session)[0], "apple pie recipe") def test_clients_are_offered_rename_as_a_read_and_a_change_of_the_name(self): with Albedo(self.provider) as app: session = app.session() self.turn(app, session, "hello") with app.api(f"/sessions/{session}/catalog") as response: commands = json.load(response)["loaded"]["commands"] offered = { entry["delivery"]: entry for entry in commands if entry["slash_name"] == "/rename" } self.assertEqual(sorted(offered), ["mutation", "read"]) change = offered["mutation"]["operation"] self.assertEqual(change["method"], "PATCH") self.assertEqual( change["body"], {"/name": {"source": "form", "pointer": "/name"}} ) self.assertEqual( [argument["name"] for argument in offered["mutation"]["arguments"]], ["name"], ) def test_an_agent_is_named_by_whoever_started_it_not_by_the_user(self): # Creating one fills `name` too, so only who started it tells them apart. with Albedo(self.provider) as app: session = app.session() self.turn(app, session, "start the agents") wait_until( lambda: sum("REFUSED" in output for output in self.outputs) >= 2, 30, f"the child and the peer never answered: {self.outputs}", ) for output in (o for o in self.outputs if "REFUSED" in o): self.assertIn("'source': 'agent'", output) self.assertIn("whoever started it", output) self.assertNotIn("the user named", output) def test_a_name_must_be_a_few_words(self): with Albedo(self.provider) as app: session = app.session() self.turn(app, session, "tell me about pears") self.turn(app, session, "try a long name") self.assertIn("REFUSED", self.outputs[-1]) self.assertIn("under 80 characters", self.outputs[-1]) self.turn(app, session, "try a blank name") self.assertIn("REFUSED", self.outputs[-1]) self.assertIn("needs some words", self.outputs[-1]) # Refused names leave the automatic title following the messages. self.assertEqual(self.shown(app, session)[0], "try a blank name")