Something went wrong. Try again.
Where attention appears to go across agent threads: collector, web explorer, and native iOS client
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208import base64import hashlibimport jsonimport secretsimport tempfileimport timeimport unittestfrom pathlib import Pathfrom types import SimpleNamespacefrom unittest import mockfrom urllib.parse import parse_qs, urlsplit
import httpxfrom starlette.applications import Starlettefrom starlette.routing import Route
from web_dashboard import APP_RETURN, COOKIE, WebDashboard
DID = 'did:plc:xbtmt2zjwlrfegqvch7fboei'BASE = 'https://agents.example.test'SNAPSHOT = {'sessions': [ {'provider': 'Codex', 'id': 'abc', 'title': 'One', 'updated': 1, 'status': 'quiet', 'events': [{'at': 1, 'label': 'You', 'text': 'hi'}]}, {'provider': 'Claude', 'id': 'def', 'title': 'Two', 'updated': 2, 'status': 'recent', 'events': []},], 'errors': [], 'at': 5}
def pair(): verifier = secrets.token_urlsafe(48) challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).rstrip(b'=').decode() return verifier, challenge
class NativeAuthTests(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self): self.directory = tempfile.TemporaryDirectory() self.root = Path(self.directory.name) self.snapshot = json.loads(json.dumps(SNAPSHOT)) self.digests = [] self.web = self.dashboard() self.client = self.http(self.web)
def dashboard(self): def digest(zone): self.digests.append(zone) return {'at': 1, 'timezone': zone, 'days': {}, 'threads': {}, 'errors': []} return WebDashboard(BASE, [DID], lambda: self.snapshot, auth_path=self.root / 'auth.sqlite', zones_path=self.root / 'zones.json', digest=digest, attention=lambda: {'events': [], 'threads': {}, 'at': 1})
def http(self, web): routes = [Route('/web/callback', web.callback), Route('/app/login', web.app_login), Route('/app/session', web.app_session, methods=['GET', 'POST']), Route('/app/logout', web.app_logout, methods=['POST']), Route('/api/sessions', web.data), Route('/api/sessions/item', web.data), Route('/api/attention', web.data), Route('/api/attention/digest', web.data)] return httpx.AsyncClient(transport=httpx.ASGITransport(app=Starlette(routes=routes)), base_url=BASE)
async def asyncTearDown(self): await self.client.aclose() self.web.store.db.close() self.directory.cleanup()
async def complete(self, challenge, outcome='ok'): state, binding = secrets.token_urlsafe(32), secrets.token_urlsafe(32) self.web.store.put('pending', state, {'binding': binding, 'app': challenge}, time.time() + 600) self.client.cookies.set(self.web.binding(state), binding) session = SimpleNamespace(did=DID if outcome == 'ok' else 'did:plc:someoneelse') with mock.patch.object(self.web.client, 'handle_callback', mock.AsyncMock(return_value=session)), mock.patch.object(self.web.client, 'revoke_session', mock.AsyncMock()): query = f'state={state}&error=access_denied' if outcome == 'denied' else f'state={state}&code=c&iss=https://pds.example.test' response = await self.client.get('/web/callback?' + query) self.assertEqual(response.status_code, 303) location = response.headers['location'] self.assertTrue(location.startswith(APP_RETURN + '?')) self.assertNotIn(COOKIE, response.headers.get('set-cookie', '')) return {k: v[0] for k, v in parse_qs(urlsplit(location).query).items()}
async def exchange(self, code, verifier): return await self.client.post('/app/session', content=json.dumps({'code': code, 'verifier': verifier}))
async def test_full_flow_issues_opaque_bearer_that_never_appears_in_a_url(self): verifier, challenge = pair() params = await self.complete(challenge) self.assertEqual(set(params), {'code'}) response = await self.exchange(params['code'], verifier) self.assertEqual(response.status_code, 200) token = response.json()['token'] self.assertNotEqual(token, params['code']) self.assertGreater(response.json()['expires_at'], time.time() + 29 * 86400) headers = {'authorization': f'Bearer {token}'} self.assertEqual((await self.client.get('/app/session', headers=headers)).json(), {'did': DID}) self.assertEqual((await self.client.get('/api/sessions', headers=headers)).status_code, 200) self.assertEqual((await self.client.get('/api/attention', headers=headers)).status_code, 200) self.assertIsNone(self.web.store.db.execute('SELECT 1 FROM auth WHERE key=?', (token,)).fetchone())
async def test_code_is_single_use_and_bound_to_the_verifier(self): verifier, challenge = pair() code = (await self.complete(challenge))['code'] wrong, _ = pair() self.assertEqual((await self.exchange(code, wrong)).json(), {'error': 'grant'}) self.assertEqual((await self.exchange(code, verifier)).json(), {'error': 'state'}) code = (await self.complete(challenge))['code'] self.assertEqual((await self.exchange(code, verifier)).status_code, 200) self.assertEqual((await self.exchange(code, verifier)).json(), {'error': 'state'}) for body in ['not json', json.dumps({'code': code}), json.dumps({'code': 1, 'verifier': verifier}), json.dumps({'code': code, 'verifier': 'short'})]: self.assertEqual((await self.client.post('/app/session', content=body)).status_code, 400)
async def test_expired_code_is_rejected(self): verifier, challenge = pair() code = (await self.complete(challenge))['code'] self.web.store.put('exchange', code, {'did': DID, 'challenge': challenge}, time.time() - 1) self.assertEqual((await self.exchange(code, verifier)).json(), {'error': 'state'})
async def test_cancel_wrong_account_and_forged_state_return_to_the_app(self): _, challenge = pair() self.assertEqual(await self.complete(challenge, 'denied'), {'error': 'denied'}) self.assertEqual(await self.complete(challenge, 'other'), {'error': 'account'}) self.assertEqual(self.web.store.count('exchange'), 0) response = await self.client.get('/web/callback?state=forged&code=c') self.assertEqual(response.headers['location'], '/?error=state') for bad in ['', 'short', 'x' * 42 + '!', 'x' * 44]: location = (await self.client.get('/app/login', params={'challenge': bad})).headers['location'] self.assertEqual(location, APP_RETURN + '?error=state')
async def test_login_start_binds_the_browser_and_redirects_to_the_pds(self): _, challenge = pair() with mock.patch.object(self.web.client, 'start_authorization', mock.AsyncMock(return_value=('https://pds.example.test/authorize?x=1', 'state-1'))): response = await self.client.get('/app/login', params={'challenge': challenge}) self.assertEqual(response.headers['location'], 'https://pds.example.test/authorize?x=1') self.assertIn(self.web.binding('state-1'), response.headers['set-cookie']) self.assertEqual(self.web.store.get('pending', 'state-1')['app'], challenge) with mock.patch.object(self.web.client, 'start_authorization', mock.AsyncMock(side_effect=RuntimeError)): self.web.last_start = 0 self.assertEqual((await self.client.get('/app/login', params={'challenge': challenge})).headers['location'], APP_RETURN + '?error=server')
async def test_revocation_restart_expiry_allowlist_and_namespace_separation(self): token = self.web.issue(DID, 'app') headers = {'authorization': f'Bearer {token}'} restarted = self.dashboard() client = self.http(restarted) self.assertEqual((await client.get('/api/sessions', headers=headers)).status_code, 200) self.assertEqual((await client.get('/api/sessions', headers={'authorization': 'Bearer ' + self.web.issue(DID)})).status_code, 401) client.cookies.set(COOKIE, token) self.assertEqual((await client.get('/api/sessions')).status_code, 401) self.assertEqual((await client.post('/app/logout', headers=headers)).status_code, 204) self.assertEqual((await self.client.get('/api/sessions', headers=headers)).status_code, 401) self.assertEqual((await self.client.get('/app/session', headers=headers)).status_code, 401) await client.aclose() restarted.store.db.close() token = self.web.issue(DID, 'app') self.web.store.put('app', token, {'did': DID}, time.time() - 1) self.assertEqual((await self.client.get('/api/sessions', headers={'authorization': f'Bearer {token}'})).status_code, 401) token = self.web.issue(DID, 'app') self.web.allowed = frozenset() self.assertEqual((await self.client.get('/api/sessions', headers={'authorization': f'Bearer {token}'})).status_code, 401)
async def test_sessions_list_omits_excerpts_and_supports_conditional_requests(self): headers = {'authorization': 'Bearer ' + self.web.issue(DID, 'app')} response = await self.client.get('/api/sessions?view=list', headers=headers) self.assertEqual(response.status_code, 200) self.assertTrue(all('events' not in s for s in response.json()['sessions'])) tag = response.headers['etag'] self.snapshot['at'] = 99 again = await self.client.get('/api/sessions?view=list', headers={**headers, 'if-none-match': tag}) self.assertEqual(again.status_code, 304) self.snapshot['sessions'][0]['title'] = 'Changed' self.assertEqual((await self.client.get('/api/sessions?view=list', headers={**headers, 'if-none-match': tag})).status_code, 200) item = await self.client.get('/api/sessions/item', params={'id': 'Codex:abc'}, headers=headers) self.assertEqual(item.json()['session']['events'][0]['text'], 'hi') self.assertEqual((await self.client.get('/api/sessions/item', params={'id': 'Codex:abc'}, headers={**headers, 'if-none-match': item.headers['etag']})).status_code, 304) self.assertIsNone((await self.client.get('/api/sessions/item', params={'id': 'Codex:missing'}, headers=headers)).json()['session']) self.assertIn('events', (await self.client.get('/api/sessions', headers=headers)).json()['sessions'][0])
async def test_universal_links_claim_only_dashboard_pages(self): body = (await self.web.app_site_association(None)).body details = json.loads(body)['applinks']['details'][0] self.assertEqual(details['appIDs'], ['65M396B5CL.io.zzstoatzz.agents']) paths = [c['/'] for c in details['components']] self.assertEqual(paths, ['/', '/attention', '/attention/timeline']) self.assertFalse(any(p.startswith(('/web', '/app', '/api', '/mcp')) for p in paths))
async def test_digest_uses_and_records_the_requested_zone(self): headers = {'authorization': 'Bearer ' + self.web.issue(DID, 'app')} self.assertEqual((await self.client.get('/api/attention/digest', params={'tz': 'Not/AZone'}, headers=headers)).status_code, 400) response = await self.client.get('/api/attention/digest', params={'tz': 'America/Vancouver'}, headers=headers) self.assertEqual(response.json()['timezone'], 'America/Vancouver') self.assertIn('America/Vancouver', json.loads((self.root / 'zones.json').read_text())) await self.client.get('/api/attention/digest', headers=headers) self.assertEqual(len(self.digests), 2) self.assertEqual(oct((self.root / 'zones.json').stat().st_mode & 0o777), '0o600')
class SnapshotShapeTests(unittest.TestCase): def test_full_view_stays_short_while_items_carry_full_text(self): from web_dashboard import compact long = 'line one\n| a | b |\n' + 'x' * 5000 snap = {'sessions': [{'provider': 'Codex', 'id': 'a', 'events': [{'at': i, 'label': 'Assistant', 'text': long} for i in range(40)] + [{'at': 50, 'label': 'You', 'text': 'spoken', 'voice': True}]}], 'errors': [], 'at': 1} out = compact(snap)['sessions'][0]['events'] self.assertEqual(len(out), 16) self.assertTrue(all(len(e['text']) <= 240 and '\n' not in e['text'] and 'voice' not in e for e in out)) self.assertEqual(snap['sessions'][0]['events'][0]['text'], long)
def test_projection_marks_cuts_and_voice(self): from public_snapshot import event self.assertTrue(event({'at': 1, 'label': 'Assistant', 'text': 'y' * 4001})['cut']) self.assertNotIn('cut', event({'at': 1, 'label': 'Assistant', 'text': 'short\nlines'})) self.assertEqual(event({'at': 1, 'label': 'Assistant', 'text': 'short\nlines'})['text'], 'short\nlines') self.assertTrue(event({'at': 1, 'label': 'You', 'text': 'hi', 'voice': True})['voice']) self.assertIn('[redacted]', event({'at': 1, 'label': 'You', 'text': 'key sk-proj-abcdefghijklmnop'})['text'])
def test_body_keeps_markdown_lines(self): from server import body self.assertEqual(body(' | a | b |\n|---|---|\n\n\n\n| 1 | 2 | '), '| a | b |\n|---|---|\n\n| 1 | 2 |') self.assertEqual(len(body('z' * 9000)), 4001)