guest/prefix.py

"""Append-only source intake; only leading, complete, literal/input tool reads.

No eval/exec here. Final execution and Future ownership remain in the existing path.
"""
import ast
import json


class Prefix:
    def __init__(self, inputs, prepare):
        self.inputs = inputs
        self.prepare = prepare
        self.source = ""
        self.seen = 0
        self.ready = []
        self.claimed = 0

    def feed(self, chunk):
        self.source += chunk
        # A trailing partial line is not a complete received statement.
        complete = self.source[:self.source.rfind("\n") + 1]
        try:
            tree = ast.parse(complete, filename="<spine>")
        except SyntaxError:
            return  # Incomplete/invalid source is diagnosed at final compile.
        for index, statement in enumerate(tree.body):
            if not self.candidate(statement):
                break  # Never pass a branch, dependency, mutation, import, or other code.
            if index < self.seen:
                continue
            self.seen = index + 1
            call = statement.value
            try:
                args = {k.arg: self.argument(k.value) for k in call.keywords}
                request = json.dumps({"tool": call.args[0].value, "args": args})
            except Exception:
                continue  # Argument errors are still raised at the actual Python call.
            handle = self.prepare(request)
            if handle:
                self.ready.append((request, handle))

    def candidate(self, statement):
        if not (isinstance(statement, ast.Assign) and len(statement.targets) == 1
                and isinstance(statement.targets[0], ast.Name)
                and statement.targets[0].id not in ("inputs", "tool")):
            return False
        call = statement.value
        return (isinstance(call, ast.Call) and isinstance(call.func, ast.Name) and call.func.id == "tool"
                and len(call.args) == 1 and isinstance(call.args[0], ast.Constant)
                and isinstance(call.args[0].value, str)
                and all(k.arg is not None and self.literal_or_input(k.value) for k in call.keywords))

    @staticmethod
    def literal_or_input(node):
        if isinstance(node, ast.Constant):
            return isinstance(node.value, (str, int, float, bool, type(None)))
        return (isinstance(node, ast.Subscript) and isinstance(node.value, ast.Name)
                and node.value.id == "inputs" and isinstance(node.slice, ast.Constant)
                and isinstance(node.slice.value, (str, int)))

    def argument(self, node):
        if isinstance(node, ast.Constant):
            return node.value
        return self.inputs[node.slice.value]

    def claim(self, request):
        # Leading calls execute in source order even if the full program is not optimized.
        # Match the actual request too; never search ahead or reuse a different read.
        if self.claimed < len(self.ready) and self.ready[self.claimed][0] == request:
            handle = self.ready[self.claimed][1]
            self.claimed += 1
            return handle
        return None