mrg.sandbox

The single surface available to a sandboxed agent. It holds no key and reaches real silicon only by handing a candidate to a broker with promote.

manhattan_reasoning_gym.sandbox

Sandboxed-agent silicon — reach a real board through the broker.

For an agent running in the locked-down sandbox (no API key, no network). Having vetted a candidate locally with mrg.build, it promotes the design; the trusted side (outside the container) runs it on silicon. This is the ONLY way a sandboxed agent reaches hardware — it never holds a key or calls the orchestrator directly (that's mrg.cloud).

promote 

promote(design: str | Path, report: Any, *, agent: str = 'default', workspace: str | Path | None = None, timeout: float = 900.0, poll_interval: float = 0.5) -> dict

Request a silicon run for design, blocking until the broker responds.

Parameters:

Name Type Description Default
design str | Path

path to the design.py being promoted.

 required
report Any

the local BuildReport for it (object or dict), forwarded to the trusted side (e.g. for a custom guard or for logging).

 required
agent str

caller id (forwarded to the trusted side; e.g. for a guard).

 'default'
workspace str | Path | None

shared dir (default: $MRG_WORKSPACE or /work).

 None
timeout float

seconds to wait for the broker.

 900.0

Returns:

Type Description
dict

The broker's response dict: {accepted, reason, silicon?}.
Source code in _code/src/manhattan_reasoning_gym/_broker_client.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def promote(
    design: str | Path,
    report: Any,
    *,
    agent: str = "default",
    workspace: str | Path | None = None,
    timeout: float = 900.0,
    poll_interval: float = 0.5,
) -> dict:
    """Request a silicon run for ``design``, blocking until the broker responds.

    Args:
        design: path to the design.py being promoted.
        report: the local BuildReport for it (object or dict), forwarded to the
            trusted side (e.g. for a custom ``guard`` or for logging).
        agent: caller id (forwarded to the trusted side; e.g. for a guard).
        workspace: shared dir (default: $MRG_WORKSPACE or /work).
        timeout: seconds to wait for the broker.

    Returns:
        The broker's response dict: ``{accepted, reason, silicon?}``.
    """
    ws = Path(workspace) if workspace else _workspace()
    promote_dir = ws / "promote"
    promote_dir.mkdir(parents=True, exist_ok=True)

    rid = uuid.uuid4().hex[:12]
    design_bytes = Path(design).read_bytes()
    report_dict = report.to_dict() if hasattr(report, "to_dict") else dict(report)
    request = {
        "id": rid,
        "agent": agent,
        "design_b64": base64.b64encode(design_bytes).decode("ascii"),
        "report": report_dict,
    }

    # Atomic publish: write to a temp name, then rename, so the broker never
    # reads a half-written request.
    req_path = promote_dir / f"{rid}.request.json"
    tmp = promote_dir / f".{rid}.request.json.tmp"
    tmp.write_text(json.dumps(request))
    os.replace(tmp, req_path)

    resp_path = promote_dir / f"{rid}.response.json"
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if resp_path.exists():
            return json.loads(resp_path.read_text())
        time.sleep(poll_interval)
    raise TimeoutError(f"broker did not respond to promote {rid} within {timeout}s")