saphid--frame-control/ui/frame_agent.py

"""Agent actions and one-use human approvals. No model SDK or network calls here."""
import hashlib
import os
from pathlib import Path
import secrets
import shlex
import shutil
import subprocess
import threading
import time


class Approvals:
    def __init__(self):
        self.pending = {}
        self.lock = threading.Lock()

    def request(self, action):
        with self.lock:
            now = time.monotonic()
            self.pending = {k: v for k, v in self.pending.items() if v['expires'] > now}
            if len(self.pending) >= 100:
                raise ValueError('Too many pending approvals; wait five minutes')
            token = secrets.token_urlsafe(24)
            self.pending[token] = {'action': action, 'approved': False, 'expires': now + 300}
        return {'confirmation': token, 'action': action, 'approvalPath': '/assistant#confirm=' + token,
                'message': 'Ask the user to review and approve this action in Frame Control, then retry with confirmation. Expires in five minutes.'}

    def entry(self, token):
        entry = self.pending.get(token)
        if not entry or entry['expires'] <= time.monotonic():
            raise ValueError('Approval expired or unknown; request a new one')
        return entry

    def inspect(self, token):
        with self.lock:
            entry = self.entry(token)
            return {'action': entry['action'], 'approved': entry['approved']}

    def decide(self, token, accept):
        with self.lock:
            entry = self.entry(token)
            if accept is True:
                entry['approved'] = True
            else:
                del self.pending[token]
        return {'message': 'Approved for one use' if accept is True else 'Rejected'}

    def consume(self, token, action):
        with self.lock:
            entry = self.entry(token)
            if entry['action'] != action or not entry['approved']:
                raise ValueError('This exact action needs approval in Frame Control')
            del self.pending[token]  # consume before starting, including on failure


approvals = Approvals()


def validate(name, args):
    fields = {
        'launch': {'appid'}, 'install': {'id'}, 'uninstall': {'id'},
        'send_text': {'text'}, 'send_file': {'path'}, 'panel': {'id'},
        'power': {'action'}, 'keep_awake': {'action'},
    }
    if name not in fields or not isinstance(args, dict) or set(args) != fields[name]:
        raise ValueError('Unknown action or arguments')
    if any(not isinstance(v, str) or not v or len(v) > 65536 for v in args.values()):
        raise ValueError('Arguments must be nonempty strings (maximum 65536 characters)')
    if name == 'power' and args['action'] not in ('suspend', 'reboot', 'poweroff'):
        raise ValueError('Unknown power action')
    if name == 'keep_awake' and args['action'] not in ('on', 'off', 'status'):
        raise ValueError('Expected on, off or status')
    action = {'name': name, 'arguments': dict(args)}
    if name == 'send_file':
        path = Path(args['path']).expanduser().resolve(strict=True)
        if not path.is_file() or path.stat().st_size > 16 * 1024**2:
            raise ValueError('Choose a regular file of at most 16 MiB')
        # Bind approval to bytes, not just a mutable filename.
        with path.open('rb') as stream:
            data = stream.read(16 * 1024**2 + 1)
        if len(data) > 16 * 1024**2:
            raise ValueError('File grew beyond 16 MiB')
        action['arguments']['path'] = str(path)
        action['sha256'] = hashlib.sha256(data).hexdigest()
        action['bytes'] = len(data)
    return action


def call(server, body):
    name, args = body.get('name'), body.get('arguments', {})
    action = validate(name, args)
    if name in ('install', 'uninstall', 'panel') and not server.FLATPAK_ID.fullmatch(args['id']):
        raise ValueError('Expected a Flatpak application ID')
    if name == 'launch' and not server.APPID.fullmatch(args['appid']):
        raise ValueError('Expected a Steam app ID')
    if name == 'keep_awake' and args['action'] == 'status':
        return keep_awake(server, 'status')
    token = body.get('confirmation')
    if not token:
        return approvals.request(action)
    approvals.consume(token, action)
    if name == 'launch':
        return server.launch(args)
    if name in ('install', 'uninstall'):
        return server.flatpak({**args, 'action': name})
    if name == 'send_text':
        return server.clipboard(args)
    if name == 'send_file':
        # Stage the reviewed bytes before the existing transfer helper reads them.
        import tempfile
        with tempfile.TemporaryDirectory(prefix='frame-agent-') as tmp:
            source = Path(action['arguments']['path'])
            with source.open('rb') as stream:
                data = stream.read(16 * 1024**2 + 1)
            if hashlib.sha256(data).hexdigest() != action['sha256']:
                raise ValueError('File changed after approval')
            staged = Path(tmp) / source.name
            staged.write_bytes(data)
            return {'message': server.push_file(staged)}
    if name == 'power':
        if server.LOCAL:
            raise ValueError('Use the Frame Control power controls to enter the password; MCP never takes passwords')
        return server.open_thing({'what': args['action']})
    if name == 'keep_awake':
        return keep_awake(server, args['action'])
    return run_script(server, 'panel-on-frame.sh', [args['id']])


def run_script(server, name, args):
    script = server.HERE.parent / 'scripts' / name
    if not script.exists() or not shutil.which('zsh') or server.LOCAL:
        raise ValueError(name + ' requires a computer with zsh and the matching script installed')
    # The headset the server is routed to, not whatever `frame` means in ~/.ssh/config.
    env = {**os.environ, 'FRAME_ALIAS': server.FRAME,
           'FRAME_SSH_OPTS': shlex.join(server.SSH[1:])}
    result = subprocess.run(['zsh', str(script), *args], capture_output=True, text=True, timeout=60, env=env)
    if result.returncode:
        raise ValueError(result.stderr.strip() or 'Script failed')
    return {'message': result.stdout.strip()}


def keep_awake(server, action):
    # PR #16 owns this interface. Never silently change timers or claim a lease.
    return run_script(server, 'keep-awake.sh', [action])