Modal

from rebuno import CheckpointPolicy, resource
from workspace_resource import ModalResource

workspace = await resource(
    "workspace",
    driver=ModalResource(REPO, token),
    checkpoints=CheckpointPolicy(every_steps=5),
)

from modal.exception import SandboxFilesystemNotFoundError
from rebuno import tool


@tool("shell", resources=["workspace"])
async def shell(command: str) -> str:
    """Run a shell command in the repository and return its exit code and output."""
    process = await workspace.exec.aio(
        "bash", "-c", command, workdir=WORKDIR, timeout=120
    )
    stdout = await process.stdout.read.aio()
    stderr = await process.stderr.read.aio()
    exit_code = await process.wait.aio()
    return f"exit code {exit_code}\n{stdout}{stderr}"[-10000:]


@tool("read_file")
async def read_file(path: str) -> str:
    """Return a file's contents. The path is relative to the repository root."""
    try:
        return await workspace.filesystem.read_text.aio(f"{WORKDIR}/{path}")
    except SandboxFilesystemNotFoundError:
        return f"{path} does not exist"


@tool("write_file", resources=["workspace"])
async def write_file(path: str, content: str) -> str:
    """Replace a file's contents, creating it if needed. The path is relative to the repository root."""
    await workspace.filesystem.write_text.aio(content, f"{WORKDIR}/{path}")
    return f"wrote {path}"

credentials = base64.b64encode(f"x-access-token:{token}".encode()).decode()
name = f"rebuno-github-{hashlib.sha256(credentials.encode()).hexdigest()[:16]}"
await modal.Secret.objects.create.aio(
    name, {"GITHUB_BASIC": credentials}, allow_existing=True
)
policy = modal.experimental.OutboundPolicy().with_header_replacement(
    domain="github.com",
    secret=modal.Secret.from_name(name),
    headers={"Authorization": "Basic $GITHUB_BASIC"},
)
sandbox = await modal.Sandbox.create.aio(
    app=app, image=image, timeout=3600, _experimental_outbound_policy=policy
)

@tool("open_pr", idempotency="at_most_once")
async def open_pr(title: str, body: str) -> str:
    """Open a pull request from the pushed branch."""
    ...

from langchain_core.messages import messages_to_dict
from rebuno import Result, previous

prior = await previous() or {}
...
return Result(
    output={"answer": result["messages"][-1].text},
    state={
        "messages": messages_to_dict(result["messages"]),
    },
)

rebuno exec fork <execution-id> --at <event-seq> --session fix-tests-fork

default_action: deny
rules:
  - id: allow-llm
    when:
      step_kind: llm_call
    then:
      decision: allow

  - id: allow-sandbox
    when:
      targets: [shell, read_file, write_file]
    then:
      decision: allow

  - id: open-pr
    when:
      target: open_pr
    then:
      decision: require_approval
      reason: opening a pull request needs approval

cd examples/integrations/sandbox/modal
rebuno dev --config rebuno.yaml

pip install rebuno modal httpx2 langchain langchain-openai
python agent.py

rebuno exec create modal '{"task": "The tests are failing. Fix the bug and open a pull request."}' --session fix-tests

rebuno exec create modal '{"task": "Also add a test for the edge case."}' --session fix-tests