PaperClipAI/tests/runner-e2e/everyday-artifact.py

"""Independent acceptance oracle. Never runs the project's own test assertions."""
import argparse, json, os, pathlib, selectors, stat, subprocess, sys, tempfile, time, uuid, zipfile

BASE_CASES = [("Hello World", "hello-world"), ("  Queue--Ready!!  ", "queue-ready"),
              ("Already-Fine", "already-fine"), ("Café 東京", "caf"), ("!!!", ""),
              ("a__b  c", "a-b-c"), ("123", "123"), ("", "")]

# Published multi-platform python:3.13-slim digest. Never pull implicitly during a model run.
SANDBOX_IMAGE = 'python@sha256:9d2e5553305c7c7b0097999bb17187c69b921ccd6bc9d40e4bb5ebe652c00285'

def preflight():
    for args in [['docker', 'version', '--format', '{{.Server.Version}}'],
                 ['docker', 'image', 'inspect', SANDBOX_IMAGE]]:
        result = subprocess.run(args, capture_output=True, text=True, timeout=10)
        if result.returncode:
            raise RuntimeError(f'Artifact sandbox qualification failed. Start Docker and run: docker pull {SANDBOX_IMAGE}')

class ArtifactSandbox:
    """Only extracted delivery files enter the container. No host secrets or network."""
    def __init__(self, root, source):
        self.root = root
        self.cwd = '/project/' + str(source.parent.relative_to(root))
        self.source = '/project/' + str(source.relative_to(root))
        self.name = 'paperclip-artifact-oracle-' + uuid.uuid4().hex

    def __enter__(self):
        preflight()
        self.root.chmod(0o755)
        args = ['docker', 'run', '--detach', '--rm', '--pull=never', '--name', self.name,
                '--network=none', '--read-only', '--cap-drop=ALL',
                '--security-opt=no-new-privileges', '--pids-limit=64', '--memory=128m',
                '--memory-swap=128m', '--cpus=1', '--user=65534:65534',
                '--log-driver=none', '--tmpfs=/tmp:rw,noexec,nosuid,size=16m',
                '--mount', f'type=bind,source={self.root},target=/project,readonly',
                '--workdir', self.cwd, SANDBOX_IMAGE, 'python', '-I', '-c',
                'import time; time.sleep(180)']
        try:
            result = subprocess.run(args, capture_output=True, text=True, timeout=15)
            if result.returncode:
                raise RuntimeError('Artifact sandbox could not start: ' + result.stderr[:500])
        except BaseException:
            self.close()
            raise
        return self

    def close(self):
        subprocess.run(['docker', 'rm', '--force', self.name], capture_output=True, timeout=10)

    def __exit__(self, *unused):
        self.close()

    def invoke(self, args, *, imported=False):
        code = ['-c', "import sys; sys.path.insert(0, " + repr(self.cwd) + "); from slugify import slugify; assert slugify(' A B! ') == 'a-b'"] if imported else [self.source, *args]
        command = ['docker', 'exec', '--workdir', self.cwd, self.name, 'python', '-I', '-B', *code]
        # Bound output as it arrives. A generated program must not exhaust host memory or disk.
        proc = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        output = {'stdout': bytearray(), 'stderr': bytearray()}
        deadline = time.monotonic() + 10
        try:
            with selectors.DefaultSelector() as selector:
                selector.register(proc.stdout, selectors.EVENT_READ, 'stdout')
                selector.register(proc.stderr, selectors.EVENT_READ, 'stderr')
                while selector.get_map():
                    remaining = deadline - time.monotonic()
                    if remaining <= 0:
                        raise ValueError('Artifact command exceeded the time limit')
                    for key, _ in selector.select(remaining):
                        chunk = os.read(key.fd, 4096)
                        if not chunk:
                            selector.unregister(key.fileobj)
                            continue
                        output[key.data].extend(chunk)
                        if sum(map(len, output.values())) > 65_536:
                            raise ValueError('Artifact command exceeded the output limit')
            code = proc.wait(timeout=max(0.01, deadline-time.monotonic()))
            return subprocess.CompletedProcess(command, code,
                output['stdout'].decode('utf-8', errors='replace'),
                output['stderr'].decode('utf-8', errors='replace'))
        finally:
            if proc.poll() is None:
                proc.kill()
            proc.wait()
            proc.stdout.close()
            proc.stderr.close()

def inspect(archive, mode):
    checks = []
    def check(name, passed, detail=""):
        checks.append(dict(id=name, passed=bool(passed), detail=detail))
    with tempfile.TemporaryDirectory(prefix="paperclip-artifact-oracle-") as tmp:
        root=pathlib.Path(tmp)
        with zipfile.ZipFile(archive) as z:
            entries=z.infolist()
            if len(entries)>250 or sum(e.file_size for e in entries)>10_000_000:
                raise ValueError("Archive exceeds the bounded source project size")
            for entry in entries:
                p=pathlib.PurePosixPath(entry.filename)
                if p.is_absolute() or ".." in p.parts or "\\" in entry.filename or stat.S_ISLNK(entry.external_attr >> 16):
                    raise ValueError("Unsafe archive member")
            z.extractall(root)
        sources=list(root.rglob("slugify.py"))
        check("one-slugify-source",len(sources)==1)
        check("readme-present",any(p.name.lower().startswith('readme') for p in root.rglob('*')))
        check("project-tests-present",any(p.name.startswith('test') and p.suffix=='.py' for p in root.rglob('*')))
        if len(sources)!=1:return checks
        source=sources[0]
        with ArtifactSandbox(root, source) as sandbox:
            invoke = sandbox.invoke
            for index,(text,expected) in enumerate(BASE_CASES):
                result=invoke([text]);check(f"base-{index}",result.returncode==0 and result.stdout.rstrip('\r\n')==expected,
                                         f"exit={result.returncode}; expected={expected!r}; observed={result.stdout[:160]!r}")
            # Import from the delivered module, not an evaluator reimplementation.
            imported=invoke([], imported=True)
            check('importable-function',imported.returncode==0)
            if mode=='separator':
                for separator,expected in [('_','queue_ready'),('-','queue-ready')]:
                    result=invoke(['  Queue--Ready!!  ','--separator',separator]);check('separator-'+separator,result.returncode==0 and result.stdout.strip()==expected)
                check('reject-invalid-separator',invoke(['hello','--separator','/']).returncode!=0)
            if mode=='max-length':
                for size,expected in [('7','queue-r'),('6','queue'),('1','q')]:
                    result=invoke(['  Queue--Ready!!  ','--max-length',size]);check('length-'+size,result.returncode==0 and result.stdout.strip()==expected)
                check('reject-zero-length',invoke(['hello','--max-length','0']).returncode!=0)
    return checks

if __name__=='__main__':
    parser=argparse.ArgumentParser();parser.add_argument('archive',nargs='?');parser.add_argument('--preflight',action='store_true');parser.add_argument('--mode',choices=['base','separator','max-length'],default='base');args=parser.parse_args()
    try:
        if args.preflight:
            preflight();checks=[dict(id='artifact-sandbox-ready',passed=True,detail='Pinned image available')]
        elif args.archive:checks=inspect(args.archive,args.mode)
        else:parser.error('archive is required unless --preflight is set')
    except Exception as error:checks=[dict(id='artifact-readable',passed=False,detail=str(error))]
    print(json.dumps(dict(passed=all(c['passed'] for c in checks),checks=checks)))
    sys.exit(0 if all(c['passed'] for c in checks) else 1)