Athena/NexusOS
1
0
Fork 0
forked from enderofwings/NexusOS
Code Pull Requests Activity

feat(chat): add run_snippet, an execution track beside the render track

Browse Source 
render_preview validates markup and hands it to the browser, which renders it
in an opaque-origin sandboxed iframe. Nothing executes server-side. That model
fits HTML/SVG/JSX and cannot fit C, Rust or Erlang, which need a real
toolchain - so those get a second tool instead of a widened first one.

The split is the feature: the model picks a track by picking a tool, rather
than picking a `lang` value from an enum where half the entries run
server-side and half do not.

synapse/code_run.py compiles and runs one file in a throwaway directory and
returns a ```nexus-run fence carrying the source and its captured output
together, so a model cannot paste output without the code that produced it.
Backticks in the source are re-encoded as ` - still valid JSON, and it
cannot close the fence early.

It is not a sandbox, and the module docstring says so up front. What it gives
is containment by layers: consent (an action tool, gated by
action_tool_policy, per-call Approve/Deny on "ask"), static screening, a
scrubbed environment in a temp dir, wall-clock and POSIX rlimits, and a
network namespace on Linux where unprivileged userns are available. Screening
is a tripwire against a model reaching for `requests` out of habit, not a
boundary against an adversary; layers 1 and 3-5 are the load-bearing ones.

Backend RUN_LANGS and frontend run-langs.js are separate registries because
the two sides need different things - one executes, one labels - and neither
should depend on the other at runtime. tests/test_tools.py asserts the key
sets and the fence tag stay equal, so drift fails the gate instead of
rendering a run result under the wrong language.

tests/snippet_probes/ is a data catalog rather than inlined cases, so adding a
language is a data change and the meta-tests can assert every RUN_LANGS key
has both a smoke probe and a screening probe. Probes skip cleanly on hosts
without the toolchain.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Athena Kaminsky
2026-08-20 14:30:03 -05:00
co-authored by Claude Opus 5
parent 425184a30b
commit affba1805c
14 changed files with 2016 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_snippet_probes.py
+176
View File
@@ -0,0 +1,176 @@
"""Automatic code-snippet probe suite.
Walks the catalog in tests/snippet_probes/catalog.py and, for every probe:
* screen — asserts critique rejects it (never executed)
* run — asserts critique is clean, then runs via synapse.code_run
(skipped cleanly when the host has no toolchain)
* tool — same as run, plus tools.dispatch("run_snippet") envelope checks
Adding a language to RUN_LANGS without a smoke probe fails
test_every_run_lang_has_a_smoke_probe. That is the point of the catalog:
coverage is automatic and visible.
"""
from __future__ import annotations
import asyncio
import json
import pytest
from synapse import code_run, tools
from tests.snippet_probes.catalog import PROBES, Probe
def _has_toolchain(lang: str) -> bool:
key = code_run.resolve_lang(lang)
entry = code_run.RUN_LANGS.get(key)
return bool(entry and entry["tool"]())
def _ids(probes=PROBES):
return [p.id for p in probes]
# ---------------------------------------------------------------------------
# Catalog integrity (runs even when every compiled toolchain is missing)
# ---------------------------------------------------------------------------
def test_probe_ids_are_unique():
ids = [p.id for p in PROBES]
assert len(ids) == len(set(ids)), "duplicate probe ids in catalog"
def test_every_run_lang_has_a_smoke_probe():
"""New RUN_LANGS entry without a smoke probe = silent blind spot."""
smoke = {
code_run.resolve_lang(p.lang)
for p in PROBES
if "smoke" in p.tags and p.kind in ("run", "tool")
}
missing = set(code_run.RUN_LANGS) - smoke
assert not missing, (
f"RUN_LANGS without a smoke probe: {sorted(missing)}. "
"Add a Probe(..., tags=('smoke', ...)) to tests/snippet_probes/catalog.py."
)
def test_every_run_lang_has_a_screen_probe():
screened = {
code_run.resolve_lang(p.lang)
for p in PROBES
if p.kind == "screen"
}
missing = set(code_run.RUN_LANGS) - screened
assert not missing, (
f"RUN_LANGS without a screening probe: {sorted(missing)}."
)
def test_catalog_langs_resolve_into_run_langs_or_aliases():
for probe in PROBES:
key = code_run.resolve_lang(probe.lang)
assert key in code_run.RUN_LANGS, (
f"probe {probe.id!r} lang={probe.lang!r} resolves to unknown {key!r}"
)
# ---------------------------------------------------------------------------
# Per-probe execution
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("probe", PROBES, ids=_ids())
def test_snippet_probe(probe: Probe):
if probe.kind == "screen":
_assert_screen(probe)
return
issues = code_run.critique(probe.lang, probe.source)
assert issues == [], f"{probe.id}: unexpected critique issues: {issues}"
if not _has_toolchain(probe.lang):
pytest.skip(f"no {code_run.resolve_lang(probe.lang)} toolchain on this machine")
result = code_run.run(probe.lang, probe.source, stdin=probe.stdin)
_assert_run_result(probe, result)
if probe.kind == "tool":
_assert_tool_envelope(probe, result)
def _assert_screen(probe: Probe) -> None:
assert probe.screen_needle, f"{probe.id}: screen probe needs screen_needle"
issues = code_run.critique(probe.lang, probe.source)
assert issues, f"{probe.id}: expected screening to reject the snippet"
assert any(probe.screen_needle in i for i in issues), (
f"{probe.id}: needle {probe.screen_needle!r} not in {issues}"
)
# Screening is the whole point — do not execute a rejected snippet.
# (A future change that runs despite issues would be a security regression.)
def _assert_run_result(probe: Probe, result: dict) -> None:
assert result.get("ok") is probe.expect_ok, (
f"{probe.id}: ok={result.get('ok')} expected {probe.expect_ok}; full={result}"
)
if probe.expect_stage is not None:
assert result.get("stage") == probe.expect_stage, result
if not probe.expect_ok:
# Failure path: still check optional stderr/stdout breadcrumbs.
for needle in probe.expect_stderr_contains:
assert needle in (result.get("stderr") or ""), result
for needle in probe.expect_stdout_contains:
assert needle in (result.get("stdout") or ""), result
return
stdout = (result.get("stdout") or "").strip()
stderr = (result.get("stderr") or "")
if probe.expect_stdout is not None:
assert stdout == probe.expect_stdout, (
f"{probe.id}: stdout {stdout!r} != {probe.expect_stdout!r}"
)
for needle in probe.expect_stdout_contains:
assert needle in stdout, result
for needle in probe.expect_stderr_contains:
assert needle in stderr, result
if probe.expect_exit is not None:
assert result.get("exit_code") == probe.expect_exit, result
else:
# Explicit "don't care" still requires that a run happened.
assert "exit_code" in result, result
def _assert_tool_envelope(probe: Probe, direct: dict) -> None:
out = json.loads(asyncio.run(tools.dispatch("run_snippet", {
"lang": probe.lang,
"source": probe.source,
"stdin": probe.stdin,
})))
assert out.get("ok") is probe.expect_ok, out
assert "fence" in out and out["fence"].startswith("```nexus-run\n"), out
body = out["fence"].split("\n", 1)[1].rsplit("\n", 1)[0]
envelope = json.loads(body)
assert envelope.get("lang") == code_run.resolve_lang(probe.lang)
if probe.expect_stdout is not None:
assert (envelope.get("stdout") or "").strip() == probe.expect_stdout
# Direct driver and tool path must agree on exit for successful runs.
if probe.expect_ok and probe.expect_exit is not None:
assert out.get("exit_code") == direct.get("exit_code") == probe.expect_exit
# ---------------------------------------------------------------------------
# One-shot inventory (useful when running the file directly)
# ---------------------------------------------------------------------------
def test_probe_inventory_lists_toolchain_readiness():
"""Not an assertion about readiness — just fails if the inventory shape
breaks, so `pytest -k inventory -s` is a quick host capability dump."""
rows = []
for name, entry in code_run.RUN_LANGS.items():
tool = entry["tool"]()
n = sum(1 for p in PROBES if code_run.resolve_lang(p.lang) == name)
rows.append({"lang": name, "ready": bool(tool), "probes": n, "tool": tool})
assert rows and all(r["probes"] >= 1 for r in rows)
# Printed only under -s; kept as a structured object for debuggability.
print("snippet-probe inventory:", json.dumps(rows, indent=2))