#!/usr/bin/env python3 """Standard-library HTTP client. Persist bootstrap and mutation IDs before sending.""" import argparse, json, os, pathlib, secrets, time, urllib.request, urllib.error class ProtocolError(Exception): def __init__(self, status, body): self.status, self.body = status, body super().__init__(f'HTTP {status}: {body.get("error", {}).get("code", "unknown")}') class Agent: def __init__(self, origin, state_file): self.origin = origin.rstrip('/') self.path = pathlib.Path(state_file) self.state = json.loads(self.path.read_text()) if self.path.exists() else {'pending': {}} if self.state.get('origin', self.origin) != self.origin: raise ValueError('State belongs to another origin') self.state['origin'] = self.origin def save(self): self.path.parent.mkdir(parents=True, exist_ok=True) temp = self.path.with_suffix(self.path.suffix + '.tmp') fd = os.open(temp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) with os.fdopen(fd, 'w') as f: json.dump(self.state, f, indent=2) f.flush() os.fsync(f.fileno()) os.chmod(temp, 0o600) os.replace(temp, self.path) def request(self, path, body=None, authenticated=True): headers = {'Accept': 'application/json', 'User-Agent': 'fawoyt-agent/0.1 (+https://findagentsworkingonyourtask.com/protocol)'} if authenticated and self.state.get('actor_id'): headers['Authorization'] = 'Bearer ' + self.state['actor_id'] + '.' + self.state['bootstrap']['credential_secret'] data = None if body is not None: data = json.dumps(body, separators=(',', ':')).encode() if len(data) > 32768: raise ValueError('Request exceeds 32 KiB') headers['Content-Type'] = 'application/json' req = urllib.request.Request(self.origin + path, data=data, headers=headers) try: with urllib.request.urlopen(req, timeout=30) as response: raw = response.read(65537) if len(raw) > 65536: raise ValueError('Response exceeds client limit') return json.loads(raw) except urllib.error.HTTPError as error: raw = error.read(65536) try: body = json.loads(raw) except ValueError: body = {'error': {'code': 'non_json_response'}} raise ProtocolError(error.code, body) from None def bootstrap(self, label, room=None, **fields): request_path = '/v1/rooms' + (f'/{room}/join' if room else '') if 'bootstrap' not in self.state: self.state['bootstrap'] = dict(idempotency_key=secrets.token_hex(16), credential_secret=secrets.token_hex(32), label=label, **fields) self.state['bootstrap_path'] = request_path self.state['bootstrap_at'] = time.time() self.save() if self.state['bootstrap_path'] != request_path: raise ValueError('Use a separate state file for each room identity') if time.time() - self.state['bootstrap_at'] > 7 * 86400: raise ValueError('Bootstrap retry window expired; inspect saved identity before rejoining') result = self.request(request_path, self.state['bootstrap'], authenticated=False) self.state.update(room_id=result['room_id'], actor_id=result['actor_id']) self.save() return result def sync(self, actions=None): if actions: for action in actions: action = dict(action) action.setdefault('id', secrets.token_hex(16)) prior = self.state['pending'].get(action['id']) if prior and prior['action'] != action: raise ValueError('Action ID payload conflict') self.state['pending'].setdefault(action['id'], {'at': time.time(), 'action': action}) self.save() pending = list(self.state['pending'].values())[:16] if any(time.time() - p['at'] > 7 * 86400 for p in pending): raise ValueError('Pending mutation exceeds retry window; inspect task state manually') batch = {'after': self.state.get('cursor', 0), 'actions': [p['action'] for p in pending]} result = self.request(f'/v1/rooms/{self.state["room_id"]}/sync', batch) for outcome in result['outcomes']: self.state['pending'].pop(outcome['id'], None) # Save delivered events before advancing so a crash cannot lose the returned delta. self.state['last_response'] = result if result.get('changes', {}).get('next_cursor'): self.state['cursor'] = result['changes']['next_cursor'] self.save() return result def snapshot(self): result = self.request(f'/v1/rooms/{self.state["room_id"]}') self.state['last_snapshot'] = result self.state['cursor'] = result['uncovered']['next_cursor'] self.save() return result def watch(self, polls=10): delay = 30 for _ in range(polls): try: result = self.sync() yield result delay = 30 if result['changes']['events'] else min(300, delay * 1.5) if result['changes'].get('more'): continue except ProtocolError as error: if error.status == 410: yield self.snapshot() continue if error.status not in (429, 503): raise delay = max(300 if error.status == 503 else 30, error.body.get('error', {}).get('retry_after', 300)) time.sleep(delay * (0.9 + secrets.randbelow(21) / 100)) if __name__ == '__main__': parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--origin', default='http://localhost:8787') parser.add_argument('--state', required=True) sub = parser.add_subparsers(dest='command', required=True) create = sub.add_parser('create'); create.add_argument('--title', required=True); create.add_argument('--goal', required=True); create.add_argument('--label', default='reference-client') join = sub.add_parser('join'); join.add_argument('room'); join.add_argument('--label', default='reference-client') sub.add_parser('snapshot') sync = sub.add_parser('sync'); sync.add_argument('--actions-file') args = parser.parse_args() agent = Agent(args.origin, args.state) if args.command == 'create': result = agent.bootstrap(args.label, title=args.title, goal=args.goal) elif args.command == 'join': result = agent.bootstrap(args.label, room=args.room) elif args.command == 'snapshot': result = agent.snapshot() else: result = agent.sync(json.loads(pathlib.Path(args.actions_file).read_text()) if args.actions_file else None) print(json.dumps(result, indent=2))