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.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061"""Loopback-only development server for the iOS app.
Serves the real WebDashboard routes against projected private data with athrowaway auth store. /app/login skips the PDS hop and returns an exchange codefor the configured DID, so the simulator exercises the real exchange, bearerand data paths. Never expose this beyond 127.0.0.1."""from pathlib import Pathimport osimport secretsimport sysimport tempfileimport time
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))import uvicornfrom starlette.applications import Starlettefrom starlette.responses import JSONResponsefrom starlette.routing import Route
from public_snapshot import read_snapshotfrom web_dashboard import CHALLENGE, EXCHANGE_TTL, WebDashboard
PORT = int(os.environ.get('AGENTS_DEV_PORT', '8769'))DID = os.environ.get('AGENTS_DEV_DID', 'did:plc:xbtmt2zjwlrfegqvch7fboei')MODE = Path(tempfile.gettempdir()) / 'agents-dev-mode'
class DevDashboard(WebDashboard): async def app_login(self, request): challenge = request.query_params.get('challenge', '') if not CHALLENGE.fullmatch(challenge): return self.app_return(error='state') if MODE.exists() and MODE.read_text().strip() in ('denied', 'account', 'server'): return self.app_return(error=MODE.read_text().strip()) code = secrets.token_urlsafe(32) self.store.put('exchange', code, {'did': DID, 'challenge': challenge}, time.time() + EXCHANGE_TTL) return self.app_return(code=code)
def fault(handler): async def wrapped(request): mode = MODE.read_text().strip() if MODE.exists() else '' if mode in ('401', '503'): return JSONResponse({'error': 'injected'}, status_code=int(mode)) return await handler(request) return wrapped
def main(): directory = tempfile.mkdtemp(prefix='agents-dev-') web = DevDashboard(f'http://127.0.0.1:{PORT}', [DID], read_snapshot, auth_path=Path(directory) / 'auth.sqlite', zones_path=Path(directory) / 'zones.json') routes = [Route('/app/login', web.app_login), Route('/app/session', web.app_session, methods=['GET', 'POST']), Route('/app/logout', web.app_logout, methods=['POST'])] routes += [Route(path, fault(web.data)) for path in ('/api/sessions', '/api/sessions/item', '/api/attention', '/api/attention/digest')] print(f'agents dev server on http://127.0.0.1:{PORT} (fault injection: write 401/503/denied/account/server to {MODE})', flush=True) uvicorn.run(Starlette(routes=routes), host='127.0.0.1', port=PORT, log_level='warning')
if __name__ == '__main__': main()