diff --git a/CLAUDE.md b/CLAUDE.md
index 1e3f7ef..65f8619 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -109,6 +109,28 @@ A bundled Ollama binary lives at `ollama/bin/ollama`. `OllamaManager` in `synaps
### Frontend (`interface/web/`)
React 19 + Vite. No routing library — `App.jsx` manages page state in a single `currentPage` useState. All API calls hit `http://localhost:8000` (configured in `src/config.js`). Built to `dist/` (gitignored) via `npm run build` and served by the backend at `:8000` — the mount is in `synapse/main.py` (`_DIST` at `/`, guarded by `is_dir()`), so `dist/` must be built for the UI to appear. Pages: Chatbot, Playbook editor, Conversation History, Models, Memory, Settings, Logs.
+### Code Tracks (`synapse/tools.py` + `synapse/code_run.py`)
+
+Two separate tools, split by *where the code runs*:
+
+- **`render_preview`** — validates markup and returns a fence the chat renders in
+ an opaque-origin `sandbox="allow-scripts"` iframe. Nothing executes
+ server-side. Languages: `PREVIEW_LANGS` in `synapse/tools.py`, mirrored by
+ `interface/web/src/preview/languages.js`.
+- **`run_snippet`** — compiles and runs a single file on the host via
+ `synapse/code_run.py`, and returns a ```nexus-run fence carrying the source and
+ its captured output. Languages: `RUN_LANGS` in `synapse/code_run.py`, mirrored
+ by `interface/web/src/preview/run-langs.js`.
+
+Each pair of registries is asserted equal by `tests/test_tools.py` — nothing
+couples them at runtime, so drift fails the check gate instead of silently
+degrading in the chat.
+
+`run_snippet` is an **action tool**: `action_tool_policy` gates it (`off` by
+default, `ask` = per-call Approve/Deny in chat). Read the `code_run.py` module
+docstring before touching it — it runs code as the current user and is explicit
+about which of its five layers are load-bearing and which are only a tripwire.
+
### Persistent Storage
Most data lands in `synapse/memory/memory.db` (SQLite, WAL mode). Tables: memory facts, conversations, messages, app settings. `synapse/memory/store.py` (`PersistentMemoryStore`) owns the schema and all queries. Playbooks are the exception — they live as YAML files in `data/playbooks/` (see Playbook System). `nexus_config.py` defines all paths; it also ensures all required directories exist on import.
diff --git a/interface/web/package.json b/interface/web/package.json
index d6fa21e..dbd0929 100644
--- a/interface/web/package.json
+++ b/interface/web/package.json
@@ -10,7 +10,7 @@
"dev": "vite",
"build": "vite build",
"lint": "eslint .",
- "test": "node --test src/preview/jsx-transform.test.js",
+ "test": "node --test \"src/preview/*.test.js\"",
"preview": "vite preview"
},
"dependencies": {
diff --git a/interface/web/src/Markdown.jsx b/interface/web/src/Markdown.jsx
index 4aeefd8..e12fb3a 100644
--- a/interface/web/src/Markdown.jsx
+++ b/interface/web/src/Markdown.jsx
@@ -6,6 +6,9 @@ import { useEffect, useRef, useState } from "react";
// auto-executing bare script isn't this feature's job (see RenderBlock's doc
// comment for the sandboxing model).
import { PREVIEW_LANGS, RENDERABLE_LANGS } from "./preview/languages.js";
+// The execution track's display half: a ```nexus-run fence is a run result
+// (source + captured output), not a program to execute here. See run-langs.js.
+import { RUN_FENCE_LANG, parseRunResult } from "./preview/run-langs.js";
// Parse content into an array of {type, value, lang, streaming} blocks.
// Handles:
@@ -29,9 +32,11 @@ function parseBlocks(content) {
let j = fenceStart + 3;
let lang = "";
- // Language specifier is valid only when word-chars are followed by a newline.
+ // Language specifier is valid only when tag-chars are followed by a newline.
// If there's no newline (e.g. ```pythonprint(...)) treat everything as code.
- const langMatch = content.slice(j).match(/^(\w+)(\r?\n)/);
+ // Hyphens count: `nexus-run` is a tag this file dispatches on, and real
+ // languages spell themselves that way too (objective-c, c-sharp).
+ const langMatch = content.slice(j).match(/^([\w-]+)(\r?\n)/);
if (langMatch) {
lang = langMatch[1];
j += langMatch[0].length;
@@ -61,6 +66,12 @@ export function Markdown({ content }) {
{blocks.map((block, i) => {
if (block.type !== "code") return ;
const lang = (block.lang || "").toLowerCase();
+ if (lang === RUN_FENCE_LANG) {
+ // A half-streamed envelope is not parseable JSON, so the block shows
+ // as code until the fence closes and then becomes the run panel.
+ const run = block.streaming ? null : parseRunResult(block.value);
+ if (run) return ;
+ }
return RENDERABLE_LANGS.has(lang)
?
: ;
@@ -126,6 +137,121 @@ function CodeBlock({ lang, value, streaming }) {
);
}
+// A finished run: the source that was executed and what it printed, in one
+// block with an Output/Code toggle.
+//
+// Nothing executes in the browser here, which is the whole difference from
+// RenderBlock below. The program already ran on the host (synapse/code_run.py)
+// under the user's per-call approval; by the time this renders, the result is
+// history. So there is no iframe, no CSP and no sandbox in this component - the
+// only untrusted thing present is *text*, and React escapes it.
+//
+// Output and stderr are shown together rather than on separate tabs: a program
+// that printed three lines and then panicked is telling one story, and splitting
+// it hides which half the reader needs. Exit code sits in the header because a
+// silent non-zero exit is otherwise invisible.
+function RunBlock({ run }) {
+ const [tab, setTab] = useState("output");
+ const [copied, setCopied] = useState(false);
+
+ const copy = () => {
+ navigator.clipboard.writeText(run.source.trimEnd()).then(() => {
+ setCopied(true);
+ setTimeout(() => setCopied(false), 1500);
+ });
+ };
+
+ const failed = run.exitCode !== null && run.exitCode !== 0;
+ const empty = !run.stdout.trim() && !run.stderr.trim();
+
+ return (
+
+ {empty && (
+
+ (the program printed nothing)
+
+ )}
+ {run.stdout && (
+
+ {run.stdout.replace(/\n$/, "")}
+
+ )}
+ {run.stderr && (
+
+ {run.stderr.replace(/\n$/, "")}
+
+ )}
+
+ ) : (
+
+ {run.source.trimEnd()}
+
+ )}
+
+ );
+}
+
// Content-Security-Policy for the rendered preview. Together with the iframe's
// `sandbox` attribute below, this is the entire trust boundary for model-
// authored HTML/SVG, so it stays conservative rather than convenient:
diff --git a/interface/web/src/preview/run-langs.js b/interface/web/src/preview/run-langs.js
new file mode 100644
index 0000000..8ea5e3e
--- /dev/null
+++ b/interface/web/src/preview/run-langs.js
@@ -0,0 +1,60 @@
+/*
+ * run-langs.js — what the chat can display a *run result* for, one entry per
+ * language.
+ *
+ * This is the display half of the execution track, and the counterpart to
+ * languages.js. The distinction between the two is where the code runs:
+ *
+ * languages.js the fence IS the program; the browser runs it in a sandboxed
+ * iframe, and this side has to build a document for it.
+ * run-langs.js the program already ran, on the host, in synapse/code_run.py.
+ * Nothing executes here — the fence carries source and captured
+ * output as JSON, and this side only labels and lays it out.
+ *
+ * So an entry needs far less than a preview entry does: no toBody, no line
+ * offsets, no runtime to inline. Just how to name the language to the reader.
+ *
+ * The backend keeps a matching registry (RUN_LANGS in synapse/code_run.py) that
+ * says how each language is *executed*. Neither depends on the other at runtime;
+ * tests/test_tools.py asserts the key sets stay equal.
+ */
+
+export const RUN_LANGS = {
+ python: { label: "Python" },
+ c: { label: "C" },
+ cpp: { label: "C++" },
+ rust: { label: "Rust" },
+ erlang: { label: "Erlang" },
+};
+
+// The fence tag run_snippet emits. Not a real language: the block's body is the
+// JSON envelope { lang, source, stdout, stderr, exit_code }, which keeps a run's
+// output attached to the source that produced it. A model pasting the fence
+// cannot paste output without the code, or code with output it never produced.
+export const RUN_FENCE_LANG = "nexus-run";
+
+/**
+ * Parse a nexus-run fence body. Returns null for anything that isn't a
+ * well-formed envelope naming a known language — the caller then falls back to
+ * showing the block as plain code, which is the honest thing to do with a
+ * result we can't vouch for.
+ */
+export function parseRunResult(text) {
+ let data;
+ try {
+ data = JSON.parse(text);
+ } catch {
+ return null;
+ }
+ if (!data || typeof data !== "object") return null;
+ if (!Object.prototype.hasOwnProperty.call(RUN_LANGS, data.lang)) return null;
+ if (typeof data.source !== "string") return null;
+ return {
+ lang: data.lang,
+ label: RUN_LANGS[data.lang].label,
+ source: data.source,
+ stdout: typeof data.stdout === "string" ? data.stdout : "",
+ stderr: typeof data.stderr === "string" ? data.stderr : "",
+ exitCode: Number.isInteger(data.exit_code) ? data.exit_code : null,
+ };
+}
diff --git a/interface/web/src/preview/run-langs.test.js b/interface/web/src/preview/run-langs.test.js
new file mode 100644
index 0000000..68d3557
--- /dev/null
+++ b/interface/web/src/preview/run-langs.test.js
@@ -0,0 +1,83 @@
+/*
+ * The run-result envelope: what parseRunResult will and won't accept.
+ *
+ * Everything this parses arrived as text a language model chose to paste into
+ * its reply, so the interesting cases are all the malformed ones. A bad envelope
+ * has to return null - the caller then shows the raw block as code, which is
+ * ugly but honest - rather than yield a half-built object that renders as a run
+ * that never happened.
+ */
+import { test } from "node:test";
+import assert from "node:assert/strict";
+
+import { RUN_LANGS, RUN_FENCE_LANG, parseRunResult } from "./run-langs.js";
+
+const envelope = (over = {}) => JSON.stringify({
+ lang: "python",
+ source: "print(1)",
+ stdout: "1\n",
+ stderr: "",
+ exit_code: 0,
+ ...over,
+});
+
+test("the fence tag is the one the backend emits", () => {
+ assert.equal(RUN_FENCE_LANG, "nexus-run");
+});
+
+test("every language has a display label", () => {
+ for (const [name, spec] of Object.entries(RUN_LANGS)) {
+ assert.equal(typeof spec.label, "string", name);
+ assert.ok(spec.label.length, name);
+ }
+});
+
+test("a well-formed envelope parses into display fields", () => {
+ const run = parseRunResult(envelope());
+ assert.equal(run.lang, "python");
+ assert.equal(run.label, "Python");
+ assert.equal(run.source, "print(1)");
+ assert.equal(run.stdout, "1\n");
+ assert.equal(run.exitCode, 0);
+});
+
+test("a backtick-escaped source round-trips through JSON", () => {
+ // run_snippet re-encodes ` as ` so the source cannot close the fence.
+ const run = parseRunResult('{"lang":"python","source":"x = \\u0060a\\u0060","exit_code":0}');
+ assert.equal(run.source, "x = `a`");
+});
+
+test("a non-zero exit code is preserved, not coerced away", () => {
+ // `exitCode || null` would turn 0 into null and hide a clean exit; a plain
+ // falsy check on the other side would call a failing program successful.
+ assert.equal(parseRunResult(envelope({ exit_code: 2 })).exitCode, 2);
+ assert.equal(parseRunResult(envelope({ exit_code: 0 })).exitCode, 0);
+});
+
+test("a missing exit code becomes null rather than a guess", () => {
+ assert.equal(parseRunResult(envelope({ exit_code: undefined })).exitCode, null);
+ assert.equal(parseRunResult(envelope({ exit_code: "0" })).exitCode, null);
+});
+
+test("absent streams read as empty, never undefined", () => {
+ const run = parseRunResult('{"lang":"c","source":"int main(){}","exit_code":0}');
+ assert.equal(run.stdout, "");
+ assert.equal(run.stderr, "");
+});
+
+test("malformed or foreign envelopes are rejected", () => {
+ assert.equal(parseRunResult("not json at all"), null);
+ assert.equal(parseRunResult("null"), null);
+ assert.equal(parseRunResult("[1,2,3]"), null);
+ assert.equal(parseRunResult('"a string"'), null);
+ assert.equal(parseRunResult(envelope({ lang: "haskell" })), null);
+ assert.equal(parseRunResult(envelope({ source: undefined })), null);
+ assert.equal(parseRunResult(envelope({ source: 42 })), null);
+});
+
+test("a prototype key is not mistaken for a supported language", () => {
+ // `data.lang in RUN_LANGS` would be true for "toString" and read the label
+ // off Object.prototype - a run panel titled with a function body.
+ assert.equal(parseRunResult(envelope({ lang: "toString" })), null);
+ assert.equal(parseRunResult(envelope({ lang: "constructor" })), null);
+});
diff --git a/synapse/chat.py b/synapse/chat.py
index 5abe2fa..cc200e0 100644
--- a/synapse/chat.py
+++ b/synapse/chat.py
@@ -139,6 +139,10 @@ async def _normalize_to_async_generator(maybe_iterable) -> AsyncGenerator[str, N
pending_approvals: Dict[str, Dict[str, Any]] = {}
_APPROVAL_TIMEOUT = 300 # seconds; a timeout is treated as "deny all"
+# Tools whose success is a fenced block the model must paste unchanged, and whose
+# rejections are worth retrying (with `_attempt`) rather than abandoning.
+_FENCE_TOOLS = frozenset({"render_preview", "run_snippet"})
+
def _as_tool_calls(obj) -> list:
"""Normalize a parsed JSON value into Ollama-style tool_calls entries."""
@@ -385,13 +389,16 @@ async def _run_tool_loop(manager, messages, model, tool_schemas, temperature, nu
# complete, styled demo handed to a struggling model gets pasted
# rather than adapted, and then persists into the conversation as a
# template for later requests.
- if name == "render_preview" and isinstance(call_args, dict):
+ if name in _FENCE_TOOLS and isinstance(call_args, dict):
call_args = {**call_args, "_attempt": render_rejects}
result = await _tools.dispatch(name, call_args)
messages.append({"role": "tool", "content": result})
- # Cap render_preview reject loops — each retry is another full
- # non-stream generation and looks like the UI is "stuck thinking".
- if name == "render_preview":
+ # Cap reject loops — each retry is another full non-stream
+ # generation and looks like the UI is "stuck thinking". The counter
+ # is shared across both fence tools on purpose: two failed attempts
+ # in a turn is two too many whether they were previews, runs, or one
+ # of each.
+ if name in _FENCE_TOOLS:
try:
body = _json.loads(result)
except Exception:
@@ -408,7 +415,13 @@ async def _run_tool_loop(manager, messages, model, tool_schemas, temperature, nu
def _last_ok_render_fence(messages: list) -> tuple[str | None, dict]:
- """Return (fence, tool_payload) from the latest successful render_preview."""
+ """Return (fence, tool_payload) from the latest successful fence tool.
+
+ Covers render_preview and run_snippet alike — both return {ok, fence, title}
+ and both are pasted verbatim rather than reconstructed. Small models rewrite
+ a fence they were told to copy, which for a preview means a demo that no
+ longer runs and for a run means output the program never actually produced.
+ """
for m in reversed(messages or []):
if m.get("role") != "tool":
continue
diff --git a/synapse/code_run.py b/synapse/code_run.py
new file mode 100644
index 0000000..c224098
--- /dev/null
+++ b/synapse/code_run.py
@@ -0,0 +1,548 @@
+"""Run a short code snippet in an ephemeral working directory.
+
+This is the *second* track of the code-preview feature and deliberately not the
+first. `render_preview` (synapse/tools.py) executes nothing server-side: it
+validates markup and the chat UI renders it inside an opaque-origin iframe.
+That model fits HTML/SVG/JSX and cannot fit C, Rust or Erlang, which need a real
+toolchain. So those go through here instead, and the result is shown as terminal
+output rather than a rendered document.
+
+WHAT THIS IS NOT
+----------------
+Not a security sandbox. Snippets run as the current user on the host. What this
+module actually provides is *containment by limits*, layered:
+
+ 1. Consent run_snippet is an ACTION tool, so it is withheld entirely
+ unless `action_tool_policy` is "ask"/"allow" — and on "ask"
+ every call waits for the user's Approve/Deny in chat.
+ 2. Screening `critique` rejects the obvious-abuse shapes (sockets, process
+ spawning, absolute paths) before anything is written to disk.
+ 3. Isolation cwd is a fresh temp dir that is deleted afterwards; HOME and
+ TMPDIR point at it; the environment is scrubbed to a small
+ allowlist.
+ 4. Limits wall-clock timeout, RLIMIT_CPU/AS/FSIZE/NPROC on POSIX,
+ truncated output.
+ 5. Network on Linux, `unshare -rn` when unprivileged user namespaces are
+ available (probed once, see `_net_isolation`). Nowhere else —
+ macOS and Windows get no network isolation at all.
+
+Layer 2 is a tripwire, not a boundary: arbitrary Python can evade any regex
+trivially. It exists to catch a model that reaches for `requests` by habit, not
+an adversary. The load-bearing layers are 1 and 3–5. Anyone wanting a real
+boundary should run this under a container or a VM; that is a deployment
+decision this module does not make for them.
+"""
+from __future__ import annotations
+
+import os
+import re
+import shutil
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+
+# Wall clock. Compilation gets its own, larger budget: rustc on a cold cache
+# routinely spends longer than any snippet is allowed to *run*, and killing a
+# compile at the run timeout would look like the code hung when it never started.
+RUN_TIMEOUT = 5.0
+COMPILE_TIMEOUT = 25.0
+
+MAX_SOURCE = 100_000 # chars of source accepted
+MAX_OUTPUT = 20_000 # chars of stdout/stderr returned, per stream
+MAX_STDIN = 10_000
+
+# Applied to the run step only. A compiler legitimately needs more address space
+# than a snippet does and forks a linker, so imposing these on the compile step
+# breaks the toolchain rather than containing the snippet.
+_MEM_BYTES = 512 * 1024 * 1024
+_MAX_PROCS = 64
+_MAX_FILE_BYTES = 8 * 1024 * 1024
+
+# RLIMIT_NPROC is counted per real UID, not per run. An absolute "512" is
+# therefore 512 minus whatever the user already has — enough on a quiet host,
+# zero on a busy macOS desktop. Erlang needs relative headroom for BEAM's
+# boot process tree; other languages keep the absolute _MAX_PROCS (blocking
+# fork is the intent there).
+_ERLANG_PROC_HEADROOM = 256
+
+
+# ---------------------------------------------------------------------------
+# Static screening
+# ---------------------------------------------------------------------------
+
+def _shared_issues(source: str) -> list[str]:
+ stripped = source.strip()
+ if len(stripped) < 8:
+ return ["source is empty or too short to run."]
+ issues: list[str] = []
+ # Only flag URLs inside string literals. A citation in a comment
+ # (`/* see https://… */`) is not a network reach and bouncing it costs a
+ # useless retry round — this layer is a tripwire, not a parser.
+ if re.search(r"""['"][^'"]*https?://[^'"]*['"]""", source):
+ issues.append(
+ "source contains an http(s) URL string — the runner has no network "
+ "access. Inline the data you need."
+ )
+ if re.search(r"\b(TODO|FIXME|your code here|implement this)\b", source, re.I):
+ issues.append("source still contains a placeholder — send the finished code.")
+ # An absolute path is either reaching outside the ephemeral dir or is a
+ # machine-specific guess that will not exist. Relative paths are fine: cwd
+ # is the temp dir and goes away with it.
+ # Windows: match `C:\…` and `C:\\…` (raw / escaped). Unix: common roots.
+ if re.search(
+ r"""['"](?:/(?:etc|home|root|usr|var|proc|sys)/|[A-Za-z]:[/\\])""",
+ source,
+ ):
+ issues.append(
+ "source references an absolute filesystem path. The snippet runs in a "
+ "throwaway directory — use relative paths, or inline the data."
+ )
+ return issues
+
+
+def _deny(source: str, rules: list[tuple[str, str]]) -> list[str]:
+ return [msg for pattern, msg in rules if re.search(pattern, source)]
+
+
+_NO_NET = "the runner has no network access"
+_NO_SPAWN = "the runner does not allow spawning other processes"
+
+_PY_RULES = [
+ (r"\b(?:import|from)\s+(?:socket|ssl|ftplib|smtplib|telnetlib|urllib|http)\b",
+ f"networking module imported — {_NO_NET}."),
+ (r"\bimport\s+(?:requests|httpx|aiohttp|urllib3)\b",
+ f"HTTP client imported — {_NO_NET}."),
+ (r"\b(?:import|from)\s+(?:subprocess|multiprocessing)\b",
+ f"subprocess/multiprocessing imported — {_NO_SPAWN}."),
+ (r"\bos\.(?:system|popen|exec[lv]|fork|spawn|kill)\b",
+ f"os process call — {_NO_SPAWN}."),
+ (r"\b(?:import|from)\s+ctypes\b",
+ "ctypes imported — the runner does not allow native calls."),
+]
+
+_C_RULES = [
+ (r"#\s*include\s*<(?:sys/socket|netinet/|arpa/|netdb)",
+ f"socket header included — {_NO_NET}."),
+ (r"\b(?:system|popen|fork|execv?[lpe]*)\s*\(",
+ f"process call — {_NO_SPAWN}."),
+]
+
+_RUST_RULES = [
+ (r"\bstd::net\b", f"std::net used — {_NO_NET}."),
+ (r"\bstd::process::(?:Command|abort)\b", f"std::process::Command used — {_NO_SPAWN}."),
+]
+
+_ERL_RULES = [
+ (r"\b(?:gen_tcp|gen_udp|httpc|inets|ssl)\b", f"networking module used — {_NO_NET}."),
+ (r"\bos:cmd\b", f"os:cmd/1 used — {_NO_SPAWN}."),
+]
+
+
+def _entry_point(pattern: str, message: str):
+ """Most of these languages fail with a linker/loader error rather than a
+ useful one when the entry point is missing, so name it up front."""
+ def check(source: str) -> list[str]:
+ return [] if re.search(pattern, source) else [message]
+ return check
+
+
+def _critique(rules: list[tuple[str, str]], entry=None):
+ def check(source: str) -> list[str]:
+ issues = _shared_issues(source)
+ if issues and issues[0].startswith("source is empty"):
+ return issues # nothing else is worth saying about an empty body
+ issues += _deny(source, rules)
+ if entry:
+ issues += entry(source)
+ return issues
+ return check
+
+
+# ---------------------------------------------------------------------------
+# Drivers
+# ---------------------------------------------------------------------------
+
+def _source_name(default: str):
+ return lambda _source: default
+
+
+def _erlang_source_name(source: str) -> str:
+ """escript reads the same file two different ways, chosen by extension: a
+ `.erl` file is compiled as a module (needs -module/-export), anything else
+ is a plain script (needs only main/1). Models write both, so let the source
+ pick its own filename instead of forcing one dialect."""
+ return "main.erl" if re.search(r"^\s*-module\s*\(", source, re.M) else "main.escript"
+
+
+def _erlang_write_source(source: str) -> str:
+ """OTP 28+ escript rejects a shebang-less `.escript` with 'Premature end of
+ file'. Module-form `.erl` files do not need one. Inject only when missing so
+ a model that already wrote `#!/usr/bin/env escript` is left alone."""
+ if _erlang_source_name(source).endswith(".escript"):
+ stripped = source.lstrip()
+ if not stripped.startswith("#!"):
+ return "#!/usr/bin/env escript\n" + source
+ return source
+
+
+# The one place that says which languages can be executed. Each entry owns that
+# language's screening, toolchain probe and argv. The tool schema's `lang` enum,
+# the capability line in the system prompt and the dispatch below are all derived
+# from these keys rather than repeating them.
+#
+# The frontend keeps a matching registry (RUN_LANGS in
+# interface/web/src/preview/run-langs.js) because the two sides need different
+# things per language — this side executes, that side labels and displays — and
+# neither should depend on the other at runtime. tests/test_tools.py asserts the
+# key sets stay equal, so drift fails the check gate instead of silently showing
+# a run result with the wrong language on it.
+RUN_LANGS: dict[str, dict] = {
+ "python": {
+ "summary": "Python script (stdlib only)",
+ "tool": lambda: sys.executable,
+ "install": "Python is bundled with NexusOS; this should not happen.",
+ "source_name": _source_name("main.py"),
+ # -I is isolated mode: ignores PYTHON* env vars, the user site-packages
+ # dir and the script's own directory on sys.path.
+ "compile": None,
+ "run": lambda tool, src, _exe: [tool, "-I", src],
+ "critique": _critique(_PY_RULES),
+ },
+ "c": {
+ "summary": "single-file C program (C11, libm linked)",
+ "tool": lambda: shutil.which("cc") or shutil.which("gcc") or shutil.which("clang"),
+ "install": "install a C compiler (clang or gcc)",
+ "source_name": _source_name("main.c"),
+ "compile": lambda cc, src, exe: [cc, "-std=c11", "-O0", "-Wall", "-o", exe, src, "-lm"],
+ "run": lambda _tool, _src, exe: [exe],
+ "critique": _critique(_C_RULES, _entry_point(
+ r"\bmain\s*\(", "no main() — a C program needs `int main(void)`.")),
+ },
+ "cpp": {
+ "summary": "single-file C++ program (C++17)",
+ "tool": lambda: shutil.which("c++") or shutil.which("g++") or shutil.which("clang++"),
+ "install": "install a C++ compiler (clang++ or g++)",
+ "source_name": _source_name("main.cpp"),
+ "compile": lambda cc, src, exe: [cc, "-std=c++17", "-O0", "-Wall", "-o", exe, src],
+ "run": lambda _tool, _src, exe: [exe],
+ "critique": _critique(_C_RULES, _entry_point(
+ r"\bmain\s*\(", "no main() — a C++ program needs `int main()`.")),
+ },
+ "rust": {
+ "summary": "single-file Rust program (2021 edition, std only)",
+ "tool": lambda: shutil.which("rustc"),
+ "install": "install Rust (https://rustup.rs)",
+ "source_name": _source_name("main.rs"),
+ # Debug build: -O roughly triples compile time for snippets that run for
+ # milliseconds either way.
+ "compile": lambda cc, src, exe: [cc, "--edition", "2021", "-o", exe, src],
+ "run": lambda _tool, _src, exe: [exe],
+ "critique": _critique(_RUST_RULES, _entry_point(
+ r"\bfn\s+main\s*\(", "no main() — a Rust program needs `fn main()`.")),
+ },
+ "erlang": {
+ "summary": "escript program with a main/1 entry point",
+ "tool": lambda: shutil.which("escript"),
+ "install": "install Erlang/OTP (provides escript)",
+ "source_name": _erlang_source_name,
+ "compile": None,
+ "run": lambda tool, src, _exe: [tool, src],
+ "critique": _critique(_ERL_RULES, _entry_point(
+ r"\bmain\s*\(", "no main/1 — escript calls `main(Args)`.")),
+ },
+}
+
+# Spellings a model reaches for that aren't the registry key. Resolved before
+# lookup so `c++` and `py` work without doubling the tool schema's enum.
+ALIASES = {
+ "c++": "cpp", "cc": "c", "py": "python", "python3": "python",
+ "rs": "rust", "erl": "erlang", "escript": "erlang",
+}
+
+
+def resolve_lang(lang: str) -> str:
+ key = (lang or "").strip().lower()
+ return ALIASES.get(key, key)
+
+
+def lang_prose() -> str:
+ """'python, c or rust' — the runnable languages as a phrase for prompts."""
+ names = list(RUN_LANGS)
+ if len(names) < 2:
+ return names[0] if names else ""
+ return f"{', '.join(names[:-1])} or {names[-1]}"
+
+
+# ---------------------------------------------------------------------------
+# Execution
+# ---------------------------------------------------------------------------
+
+# Environment variables the child keeps. Everything else is dropped: the snippet
+# has no business seeing API keys, proxy settings or the user's shell config, and
+# PYTHON*/LD_* in particular would let ambient config change how it runs.
+_ENV_KEEP = ("PATH", "LANG", "LC_ALL", "TERM", "SYSTEMROOT", "COMSPEC")
+
+# rustc is usually a rustup shim, and a shim with HOME rewritten cannot find its
+# own toolchain. Passing these through is what makes Rust work at all here; they
+# point at read-only toolchain data, not at anything the snippet should write.
+_ENV_KEEP_BY_LANG = {"rust": ("RUSTUP_HOME", "CARGO_HOME", "RUSTUP_TOOLCHAIN")}
+
+
+def _child_env(lang: str, workdir: Path) -> dict:
+ env = {k: os.environ[k] for k in _ENV_KEEP if k in os.environ}
+ for k in _ENV_KEEP_BY_LANG.get(lang, ()):
+ if k in os.environ:
+ env[k] = os.environ[k]
+ if lang == "rust" and "RUSTUP_HOME" not in env:
+ # HOME is about to be rewritten, so resolve rustup's default location
+ # against the real home while we still know it.
+ default = Path.home() / ".rustup"
+ if default.is_dir():
+ env["RUSTUP_HOME"] = str(default)
+ env.setdefault("CARGO_HOME", str(Path.home() / ".cargo"))
+ env["HOME"] = str(workdir)
+ env["TMPDIR"] = str(workdir)
+ env.setdefault("LC_ALL", "C.UTF-8")
+ return env
+
+
+def _user_process_count() -> int | None:
+ """How many processes this UID already owns. None when we cannot count
+ (Windows, or psutil missing) — callers must not invent an absolute cap."""
+ if os.name == "nt" or not hasattr(os, "getuid"):
+ return None
+ try:
+ import psutil # optional; process extra
+ except ImportError:
+ return None
+ uid = os.getuid()
+ n = 0
+ for proc in psutil.process_iter(["uids"]):
+ try:
+ uids = proc.info.get("uids")
+ if uids is not None and getattr(uids, "real", None) == uid:
+ n += 1
+ except (psutil.Error, TypeError, AttributeError):
+ continue
+ return n
+
+
+def _max_procs_for(lang: str) -> int | None:
+ """Soft RLIMIT_NPROC for this run, or None to leave the limit unset.
+
+ Erlang: current per-UID count + headroom (RLIMIT_NPROC is UID-scoped).
+ Everything else: the absolute _MAX_PROCS tripwire against fork bombs.
+ """
+ if lang == "erlang":
+ current = _user_process_count()
+ if current is None:
+ return None
+ return current + _ERLANG_PROC_HEADROOM
+ return _MAX_PROCS
+
+
+def _limits(constrain_memory: bool, max_procs: int | None = None):
+ """preexec_fn applying POSIX rlimits, or None where they don't exist.
+
+ RLIMIT_CPU is a backstop for the wall-clock timeout: a snippet that ignores
+ SIGTERM still loses the CPU. The memory and process caps are skipped for
+ compilation — see _MEM_BYTES.
+
+ `max_procs=None` with constrain_memory means "do not set RLIMIT_NPROC"
+ (used when we cannot compute a relative Erlang ceiling). Passing an int
+ always sets it.
+ """
+ if sys.platform == "win32":
+ return None
+ try:
+ import resource
+ except ImportError: # pragma: no cover - POSIX only
+ return None
+
+ cpu = int(COMPILE_TIMEOUT if not constrain_memory else RUN_TIMEOUT) + 1
+ wanted = [("RLIMIT_CPU", cpu), ("RLIMIT_FSIZE", _MAX_FILE_BYTES), ("RLIMIT_CORE", 0)]
+ if constrain_memory:
+ wanted.append(("RLIMIT_AS", _MEM_BYTES))
+ # Distinguish "caller omitted" (use default) from "explicitly skip"
+ # by requiring the kw to be passed — see _spawn.
+ if max_procs is not None:
+ wanted.append(("RLIMIT_NPROC", max_procs))
+
+ def apply(): # runs in the forked child, between fork and exec
+ # Every limit is set independently and failure is swallowed. Which of
+ # these exist, and which can be lowered, varies by platform (macOS has
+ # no usable RLIMIT_AS, RLIMIT_NPROC is absent on some POSIX systems) —
+ # and an exception raised here does not "skip a limit", it aborts the
+ # spawn entirely. Partial limits are the right failure mode; no run at
+ # all is not.
+ for name, soft in wanted:
+ which = getattr(resource, name, None)
+ if which is None:
+ continue
+ try:
+ _, hard = resource.getrlimit(which)
+ if hard != resource.RLIM_INFINITY:
+ soft = min(soft, hard)
+ resource.setrlimit(which, (soft, hard))
+ except (ValueError, OSError):
+ continue
+
+ return apply
+
+
+_net_isolation_cache: list | None = None
+
+
+def _net_isolation() -> list[str]:
+ """argv prefix that drops the child into an empty network namespace, or [].
+
+ Linux only, and only where unprivileged user namespaces are enabled — which
+ is a kernel/distro setting we can't change and shouldn't fail over. Probed
+ once and cached; an empty list means the run simply has host networking, and
+ callers must not treat this as a guarantee either way.
+ """
+ global _net_isolation_cache
+ if _net_isolation_cache is not None:
+ return _net_isolation_cache
+ _net_isolation_cache = []
+ if sys.platform.startswith("linux") and shutil.which("unshare"):
+ try:
+ probe = subprocess.run(
+ ["unshare", "-rn", "true"],
+ capture_output=True, timeout=5,
+ )
+ if probe.returncode == 0:
+ _net_isolation_cache = ["unshare", "-rn"]
+ except (OSError, subprocess.SubprocessError):
+ pass
+ return _net_isolation_cache
+
+
+def _clip(raw: bytes) -> str:
+ text = raw.decode("utf-8", errors="replace")
+ if len(text) <= MAX_OUTPUT:
+ return text
+ return text[:MAX_OUTPUT] + f"\n... [truncated at {MAX_OUTPUT} characters]"
+
+
+def _spawn(argv: list[str], workdir: Path, env: dict, timeout: float,
+ stdin: str = "", constrain_memory: bool = True,
+ max_procs: int | None = _MAX_PROCS):
+ """Spawn a child. `max_procs` defaults to `_MAX_PROCS`; pass `None` to skip
+ RLIMIT_NPROC entirely (Erlang, when a relative ceiling cannot be computed)."""
+ return subprocess.run(
+ argv,
+ cwd=str(workdir),
+ env=env,
+ input=stdin.encode("utf-8"),
+ capture_output=True,
+ timeout=timeout,
+ preexec_fn=_limits(constrain_memory, max_procs=max_procs),
+ )
+
+
+def run(lang: str, source: str, stdin: str = "") -> dict:
+ """Compile (if needed) and run `source`. Blocking — call from a thread.
+
+ Returns {ok, lang, stage, exit_code, stdout, stderr, error}. `ok` is False
+ only when the snippet could not be run at all (missing toolchain, compile
+ error, timeout); a program that runs and exits non-zero is a successful run
+ with a non-zero exit_code, because its stderr is the answer the user wants.
+ """
+ key = resolve_lang(lang)
+ entry = RUN_LANGS.get(key)
+ if entry is None:
+ return {"ok": False, "lang": lang, "stage": "lang",
+ "error": f"cannot run {lang!r} — use {lang_prose()}."}
+
+ tool = entry["tool"]()
+ if not tool:
+ return {"ok": False, "lang": key, "stage": "toolchain",
+ "error": f"no toolchain for {key} on this machine — {entry['install']}. "
+ "Show the code instead of running it."}
+
+ with tempfile.TemporaryDirectory(prefix="nexus-run-") as tmp:
+ workdir = Path(tmp)
+ body = _erlang_write_source(source) if key == "erlang" else source
+ src = workdir / entry["source_name"](source)
+ src.write_text(body, encoding="utf-8")
+ exe = str(workdir / ("program.exe" if sys.platform == "win32" else "program"))
+ env = _child_env(key, workdir)
+ max_procs = _max_procs_for(key)
+
+ if entry["compile"]:
+ try:
+ built = _spawn(entry["compile"](tool, str(src), exe), workdir, env,
+ COMPILE_TIMEOUT, constrain_memory=False)
+ except subprocess.TimeoutExpired:
+ return {"ok": False, "lang": key, "stage": "compile",
+ "error": f"compilation timed out after {COMPILE_TIMEOUT:g}s."}
+ except OSError as e:
+ return {"ok": False, "lang": key, "stage": "compile",
+ "error": f"could not start the compiler: {e}"}
+ if built.returncode != 0:
+ return {"ok": False, "lang": key, "stage": "compile",
+ "exit_code": built.returncode,
+ "stdout": _clip(built.stdout), "stderr": _clip(built.stderr),
+ "error": "compilation failed — read stderr, fix the source, "
+ "and call run_snippet again."}
+
+ argv = _net_isolation() + entry["run"](tool, str(src), exe)
+ try:
+ done = _spawn(argv, workdir, env, RUN_TIMEOUT, stdin=stdin[:MAX_STDIN],
+ max_procs=max_procs)
+ except subprocess.TimeoutExpired as e:
+ return {"ok": False, "lang": key, "stage": "run",
+ "stdout": _clip(e.stdout or b""), "stderr": _clip(e.stderr or b""),
+ "error": f"the program did not finish within {RUN_TIMEOUT:g}s — "
+ "it is probably looping. Bound the work and try again."}
+ except OSError as e:
+ return {"ok": False, "lang": key, "stage": "run",
+ "error": f"could not start the program: {e}"}
+
+ stdout = _clip(done.stdout)
+ stderr = _clip(done.stderr)
+ # BEAM failing to fork under RLIMIT_NPROC looks like a snippet bug if we
+ # report ok=True. Call it out as a runner limit so the model does not keep
+ # rewriting a correct program.
+ if (
+ key == "erlang"
+ and done.returncode != 0
+ and "Resource temporarily unavailable" in stderr
+ ):
+ return {
+ "ok": False,
+ "lang": key,
+ "stage": "run",
+ "exit_code": done.returncode,
+ "stdout": stdout,
+ "stderr": stderr,
+ "error": (
+ "Erlang could not start under the process limit (host already has "
+ "many processes for this user). Free some processes and retry, or "
+ "show the code instead of running it."
+ ),
+ }
+
+ return {
+ "ok": True,
+ "lang": key,
+ "stage": "run",
+ "exit_code": done.returncode,
+ "stdout": stdout,
+ "stderr": stderr,
+ }
+
+
+def critique(lang: str, source: str) -> list[str]:
+ """Static screening for one snippet. See the module docstring on what this
+ is worth: a tripwire against habitual network/process reaches, not a
+ boundary."""
+ key = resolve_lang(lang)
+ entry = RUN_LANGS.get(key)
+ if entry is None:
+ return [f"cannot run {lang!r} — use {lang_prose()}."]
+ if len(source) > MAX_SOURCE:
+ return [f"source is over {MAX_SOURCE} characters — send a single focused snippet."]
+ return entry["critique"](source)
diff --git a/synapse/main.py b/synapse/main.py
index 88ae72f..cf32dd3 100644
--- a/synapse/main.py
+++ b/synapse/main.py
@@ -79,6 +79,19 @@ _RENDER_PREAMBLE = (
"CSS/JS, no network.\n"
)
+# The execution track's hint, on the same terms as the render one: offered only
+# when the tool behind it is, for the same contamination reason. The distinction
+# it has to carry is which track a request belongs to — a model that reaches for
+# run_snippet to "preview" an HTML page gets a compile error, and one that
+# reaches for render_preview to run a C program gets a plain code block.
+_RUN_PREAMBLE = (
+ "\n\n---\nCode runner: when the answer depends on what code actually does, call "
+ f"the `run_snippet` tool with a complete {_tools.code_run.lang_prose()} program, "
+ "then paste the returned `fence` into your reply. It really runs, in a throwaway "
+ "directory with no network and a few seconds of CPU. Describe only the output it "
+ "returned.\n"
+)
+
_CODING_KEYWORDS = frozenset({
"code", "coding", "function", "class", "method", "variable", "bug", "error",
@@ -416,8 +429,16 @@ async def chat_stream_endpoint(payload: Dict[str, Any]):
# is implemented using a tool called render_preview... renders it live in
# a sandbox" — this text, recited as fact. A hint for a tool that isn't
# being offered is pure contamination.
+ # Read once here rather than at the tools block below: the run-track
+ # hint and the run-track schema have to agree about whether the tool is
+ # on offer, and a second lookup is a second thing to keep in step.
+ _policy = app_settings.get("action_tool_policy", "off")
+ allow_actions = _policy != "off"
+
if _tools.wants_render_preview(message):
system_prompt = (system_prompt + _RENDER_PREAMBLE) if system_prompt else _RENDER_PREAMBLE.lstrip()
+ if allow_actions and _tools.wants_code_run(message):
+ system_prompt = (system_prompt + _RUN_PREAMBLE) if system_prompt else _RUN_PREAMBLE.lstrip()
# ── MindTrace pre-flight ──────────────────────────────────────────
_trace_intent = _detect_intent(message) if message else "chat"
@@ -479,13 +500,17 @@ async def chat_stream_endpoint(payload: Dict[str, Any]):
# Tools: playbook allowlist, plus render_preview only when this turn
# looks like a visual ask (always advertising it forced a non-stream
# tool round on every chat and felt like "stuck thinking").
- _policy = app_settings.get("action_tool_policy", "off")
- allow_actions = _policy != "off"
_pb_tools = list(getattr(_main_pb, "tools", None) or []) if _main_pb else []
schemas_by_name: dict = {}
if _tools.wants_render_preview(message) or "render_preview" in _pb_tools:
for s in _tools.standing_schemas():
schemas_by_name[s["function"]["name"]] = s
+ # run_snippet rides the same cue mechanism but stays behind the action
+ # gate: it executes code on this machine. With the policy "off", "run
+ # this" gets an explanation and a code block, never a subprocess.
+ if allow_actions and _tools.wants_code_run(message):
+ for s in _tools.run_schemas():
+ schemas_by_name[s["function"]["name"]] = s
for s in _tools.schemas_for(_pb_tools, allow_actions):
schemas_by_name[s["function"]["name"]] = s
schemas = list(schemas_by_name.values())
diff --git a/synapse/tools.py b/synapse/tools.py
index 62f7396..03806c9 100644
--- a/synapse/tools.py
+++ b/synapse/tools.py
@@ -13,6 +13,7 @@ from __future__ import annotations
import json
from typing import Awaitable, Callable
+from . import code_run
from .memory.store import store, MemoryItem
from .ollama_manager import get_ollama_manager
@@ -742,6 +743,152 @@ async def _render_preview(
})
+# ---------------------------------------------------------------------------
+# run_snippet — the execution track
+# ---------------------------------------------------------------------------
+#
+# render_preview and run_snippet are deliberately separate tools over separate
+# registries. A preview is validated here and *rendered* by the browser inside a
+# sandboxed frame; a snippet is *executed* on the host by synapse/code_run.py and
+# comes back as terminal output. Stretching one tool across both would have meant
+# a `lang` enum where half the values run server-side and half don't, and a
+# single description that could only be vague about which. The split is the
+# feature: the model picks a track by picking a tool.
+#
+# Result fence: one ```nexus-run block whose body is JSON (source + streams +
+# exit code), so the code and the output it produced cannot be separated by a
+# model pasting only half of it. Backticks in the source are re-encoded as \u0060
+# — still valid JSON, and it cannot terminate the fence early.
+_RUN_FENCE_LANG = "nexus-run"
+
+
+def _run_fence(payload: dict) -> str:
+ body = json.dumps(payload, ensure_ascii=False).replace("`", "\\u0060")
+ return f"```{_RUN_FENCE_LANG}\n{body}\n```"
+
+
+def _run_lang_prose() -> str:
+ return code_run.lang_prose()
+
+
+def _run_scaffold_for(lang: str) -> str:
+ """Minimal entry-point patterns for a struggling model. Shown from the
+ second rejection on — same discipline as render_preview's scaffold."""
+ return {
+ "python": "print(sum(range(10)))\n",
+ "c": (
+ "#include \n"
+ "int main(void) {\n"
+ ' printf("%d\\n", 42);\n'
+ " return 0;\n"
+ "}\n"
+ ),
+ "cpp": (
+ "#include \n"
+ "int main() {\n"
+ ' std::cout << 42 << "\\n";\n'
+ " return 0;\n"
+ "}\n"
+ ),
+ "rust": 'fn main() { println!("{}", 42); }\n',
+ "erlang": 'main(_) -> io:format("~p~n", [42]).\n',
+ }.get(lang, f"(a short self-contained {lang} program that prints to stdout)\n")
+
+
+def _with_run_scaffold(payload: dict, lang: str, attempt: int) -> dict:
+ """Attach a starting pattern from the second rejection on. First rejection
+ stays issues-only so a pasteable scaffold does not become the answer."""
+ if attempt < 1:
+ return payload
+ return {
+ **payload,
+ "scaffold": _run_scaffold_for(lang),
+ "scaffold_note": (
+ "A pattern to adapt, not an answer to paste. Keep the entry point, "
+ "print results to stdout, and drop anything you do not use."
+ ),
+ }
+
+
+async def _run_snippet(
+ lang: str = "",
+ source: str = "",
+ stdin: str = "",
+ title: str = "",
+ _attempt: int = 0,
+ **_,
+) -> str:
+ """Compile and run a snippet, returning a fence that carries both the source
+ and what it printed.
+
+ ACTION tool: it executes code on this machine. `action_tool_policy` gates it
+ (withheld on "off", per-call Approve/Deny on "ask"), and synapse/code_run.py
+ documents exactly how much containment the run itself gets — which is less
+ than the word "sandbox" would imply.
+
+ `_attempt` is supplied by the tool loop, not by the model — it is how many
+ times this call has already been rejected in the current turn."""
+ import asyncio as _a
+
+ key = code_run.resolve_lang(lang)
+ source = (source or "").strip()
+ title = (title or "").strip()
+
+ if key not in code_run.RUN_LANGS:
+ return json.dumps({
+ "ok": False,
+ "error": f"lang must be {_run_lang_prose()} (got {lang!r}). "
+ "For HTML, SVG or JSX use render_preview instead — those are "
+ "rendered in the browser, not executed here.",
+ })
+ if not source:
+ return json.dumps(_with_run_scaffold({
+ "ok": False,
+ "error": f"source is required — send the complete {key} program, "
+ "entry point included.",
+ }, key, _attempt))
+
+ issues = code_run.critique(key, source)
+ if issues:
+ return json.dumps(_with_run_scaffold({
+ "ok": False,
+ "issues": issues,
+ "hint": (
+ "Fix these and call run_snippet again. The program runs in a throwaway "
+ "directory with no network and a few seconds of CPU: no downloads, no "
+ "absolute paths, no unbounded loops. Print your results to stdout."
+ ),
+ }, key, _attempt))
+
+ result = await _a.to_thread(code_run.run, key, source, stdin or "")
+ if not result.get("ok"):
+ # A failed compile hands back the compiler's own diagnostics: they name
+ # the line and the fix, and paraphrasing them here would only lose that.
+ return json.dumps({k: v for k, v in result.items() if v not in ("", None)})
+
+ return json.dumps({
+ "ok": True,
+ "lang": key,
+ "exit_code": result["exit_code"],
+ "stdout": result["stdout"],
+ "stderr": result["stderr"],
+ "title": title or f"{key} output",
+ "instruction": (
+ "This ran for real — the output below is what it printed. Write a short "
+ "intro, then paste the fenced block exactly as it is. Do not wrap it in a "
+ "second fence, retype the output, or claim results it does not show."
+ ),
+ "fence": _run_fence({
+ "lang": key,
+ "source": source,
+ "stdout": result["stdout"],
+ "stderr": result["stderr"],
+ "exit_code": result["exit_code"],
+ }),
+ })
+
+
+
# name -> (schema, callable). Schema is the OpenAI/Ollama function-tool format.
REGISTRY: dict[str, tuple[dict, Callable[..., Awaitable[str]]]] = {
"search_memory": (
@@ -870,6 +1017,62 @@ REGISTRY: dict[str, tuple[dict, Callable[..., Awaitable[str]]]] = {
},
_render_preview,
),
+ "run_snippet": (
+ {
+ "type": "function",
+ "function": {
+ "name": "run_snippet",
+ # Same house style as render_preview: imperative, second person,
+ # nothing the model can recite back in place of acting.
+ "description": (
+ f"Compile and run a short {code_run.lang_prose()} program on this "
+ "machine and get its real output back. Use this when the answer "
+ "depends on what the code actually does — output, a computed "
+ "result, whether it compiles. Write one self-contained file with "
+ "its entry point; there is no package manager, no network, a "
+ f"throwaway working directory and {code_run.RUN_TIMEOUT:g}s of "
+ "runtime, so bound your loops and print results to stdout. "
+ "For HTML, SVG or JSX use render_preview instead. "
+ "Rejected: fix what `issues` or `stderr` says and send it again. "
+ "Accepted: paste the returned `fence` into your reply unchanged, "
+ "and describe only the output it actually contains."
+ ),
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "lang": {
+ "type": "string",
+ "enum": list(code_run.RUN_LANGS),
+ "description": (
+ "Language to run: "
+ + "; ".join(
+ f"{name} ({spec['summary']})"
+ for name, spec in code_run.RUN_LANGS.items()
+ )
+ ),
+ },
+ "title": {
+ "type": "string",
+ "description": "Short label for what this program does.",
+ },
+ "source": {
+ "type": "string",
+ "description": (
+ "The complete single-file program, entry point included. "
+ "Standard library only."
+ ),
+ },
+ "stdin": {
+ "type": "string",
+ "description": "Optional text piped to the program's stdin.",
+ },
+ },
+ "required": ["lang", "source"],
+ },
+ },
+ },
+ _run_snippet,
+ ),
"web_search": (
{
"type": "function",
@@ -924,7 +1127,13 @@ REGISTRY: dict[str, tuple[dict, Callable[..., Awaitable[str]]]] = {
# Tools that act (write local state or reach the network). These require an
# explicit consent gate (settings.allow_action_tools) on top of the per-playbook
# allowlist — a playbook granting one isn't enough on its own.
-ACTION_TOOLS = frozenset({"web_search", "fetch_url", "remember"})
+ACTION_TOOLS = frozenset({"web_search", "fetch_url", "remember", "run_snippet"})
+
+# Action tools offered on a *cue* rather than only via a playbook allowlist —
+# the run track is a standing UI capability like the render window, but unlike
+# render_preview it executes code, so it stays behind the action gate. Listed
+# here so main.py can advertise it on a "run this" without a playbook edit.
+CUED_ACTION_TOOLS = frozenset({"run_snippet"})
# Always advertised when the user asks for a visual (see wants_render_preview).
# Not playbook-gated — the render window is a standing UI capability.
@@ -949,16 +1158,51 @@ _RENDER_HINTS = (
) + tuple(PREVIEW_LANGS)
-def wants_render_preview(message: str) -> bool:
- """True when this turn should advertise render_preview / enter the tool loop."""
+# Cues for run_snippet. Unlike _RENDER_HINTS these are verb phrases, not topic
+# words, and deliberately do not include the language names: "write me a python
+# function" is not a request to execute anything, and putting `python` in here
+# would drag every mention of the language into a slow non-stream tool round.
+# Asking for the *output* is the signal, so that is what these match.
+_RUN_HINTS = (
+ "run this", "run it", "run that", "run the code", "run this code",
+ "run my code", "run the program", "run and show", "actually run",
+ "run snippet", "run_snippet", "execute this", "execute it", "execute the code",
+ "compile", "compiles", "does it compile", "does this compile",
+ "show the output", "show me the output", "what does it print",
+ "what does this print", "what's the output", "whats the output",
+ "what is the output", "actual output", "real output",
+)
+
+
+def _mentions(message: str, hints: tuple[str, ...]) -> bool:
+ """True if any hint appears in `message` as a whole word/phrase.
+
+ The lookarounds rather than \\b: several hints end in a non-word character
+ ("what's the output"), where \\b would anchor to the apostrophe instead of
+ the phrase and match inside longer words.
+ """
import re
lower = (message or "").lower()
return any(
re.search(rf"(? bool:
+ """True when this turn should advertise render_preview / enter the tool loop."""
+ return _mentions(message, _RENDER_HINTS)
+
+
+def wants_code_run(message: str) -> bool:
+ """True when this turn is asking for code to actually be executed.
+
+ Only a hint: run_snippet is an action tool, so this can never be what makes
+ it available — `action_tool_policy` still has to be off "off" first.
+ """
+ return _mentions(message, _RUN_HINTS)
+
+
def is_action(name: str) -> bool:
return name in ACTION_TOOLS
@@ -978,6 +1222,15 @@ def standing_schemas() -> list[dict]:
return schemas_for(sorted(STANDING_TOOLS), allow_actions=True)
+def run_schemas() -> list[dict]:
+ """Schemas that ship with "run this" turns (currently just run_snippet).
+
+ Callers must have already established that actions are permitted — these are
+ action tools and schemas_for would happily hand them over regardless.
+ """
+ return schemas_for(sorted(CUED_ACTION_TOOLS), allow_actions=True)
+
+
async def dispatch(name: str, args: dict | None) -> str:
"""Run a tool by name. Never raises — returns an error string on failure."""
entry = REGISTRY.get(name)
diff --git a/tests/snippet_probes/__init__.py b/tests/snippet_probes/__init__.py
new file mode 100644
index 0000000..a3cfc20
--- /dev/null
+++ b/tests/snippet_probes/__init__.py
@@ -0,0 +1 @@
+"""Data-driven snippet probes exercised by tests/test_snippet_probes.py."""
diff --git a/tests/snippet_probes/catalog.py b/tests/snippet_probes/catalog.py
new file mode 100644
index 0000000..eeb1fe2
--- /dev/null
+++ b/tests/snippet_probes/catalog.py
@@ -0,0 +1,274 @@
+"""Catalog of code-snippet probes for the execution track.
+
+Each probe is a small, self-contained program (or deliberate reject) that the
+suite in tests/test_snippet_probes.py runs automatically. Keeping the corpus
+here — not inlined in the test file — means adding a language or a case is a
+data change, and the meta-tests can assert every RUN_LANGS key is covered.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+
+@dataclass(frozen=True)
+class Probe:
+ """One automatic snippet probe.
+
+ kind:
+ run — critique must be clean, then code_run.run (skip if no toolchain)
+ screen — critique must report at least one issue matching needle;
+ never executed
+ tool — same as run, but also through tools.dispatch("run_snippet")
+ """
+
+ id: str
+ lang: str
+ source: str
+ kind: str = "run"
+ expect_stdout: str | None = None # exact strip() match when set
+ expect_stdout_contains: tuple[str, ...] = ()
+ expect_stderr_contains: tuple[str, ...] = ()
+ expect_exit: int | None = 0 # None = don't care; run-ok can be nonzero
+ expect_ok: bool = True # False => stage failure (compile/timeout/…)
+ expect_stage: str | None = None
+ screen_needle: str | None = None # required substring in critique issues
+ stdin: str = ""
+ tags: tuple[str, ...] = field(default_factory=tuple)
+
+
+# ---------------------------------------------------------------------------
+# Corpus — keep each source short; the suite runs every probe on every check.
+# ---------------------------------------------------------------------------
+
+PROBES: tuple[Probe, ...] = (
+ # --- python (always available) -----------------------------------------
+ Probe(
+ id="py-hello",
+ lang="python",
+ source="print('probe-ok', 6 * 7)",
+ expect_stdout="probe-ok 42",
+ tags=("smoke", "python"),
+ ),
+ Probe(
+ id="py-stdin",
+ lang="python",
+ source="import sys\nprint(sys.stdin.read().strip().upper())",
+ stdin="nexus\n",
+ expect_stdout="NEXUS",
+ tags=("stdin", "python"),
+ ),
+ Probe(
+ id="py-nonzero-exit",
+ lang="python",
+ source="import sys\nprint('before')\nsys.exit(3)",
+ expect_stdout="before",
+ expect_exit=3,
+ tags=("exit", "python"),
+ ),
+ Probe(
+ id="py-traceback",
+ lang="python",
+ source="print(1 / 0)",
+ expect_exit=None, # nonzero, exact code is interpreter-dependent enough
+ expect_stderr_contains=("ZeroDivisionError",),
+ tags=("stderr", "python"),
+ ),
+ Probe(
+ id="py-alias-py",
+ lang="py",
+ source="print('alias')",
+ expect_stdout="alias",
+ tags=("alias", "python"),
+ ),
+ Probe(
+ id="py-tool-envelope",
+ lang="python",
+ source="print('via-tool', 2 + 2)",
+ kind="tool",
+ expect_stdout="via-tool 4",
+ tags=("tool", "python"),
+ ),
+ Probe(
+ id="py-screen-network",
+ lang="python",
+ source="import socket\nprint(socket.gethostname())",
+ kind="screen",
+ screen_needle="network",
+ tags=("screen", "python"),
+ ),
+ Probe(
+ id="py-screen-subprocess",
+ lang="python",
+ source="import subprocess\nsubprocess.run(['true'])",
+ kind="screen",
+ screen_needle="processes",
+ tags=("screen", "python"),
+ ),
+
+ # --- c -----------------------------------------------------------------
+ Probe(
+ id="c-hello",
+ lang="c",
+ source=(
+ "#include \n"
+ "int main(void) {\n"
+ ' printf("c-probe %d\\n", 6 * 7);\n'
+ " return 0;\n"
+ "}\n"
+ ),
+ expect_stdout="c-probe 42",
+ tags=("smoke", "c"),
+ ),
+ Probe(
+ id="c-math",
+ lang="c",
+ source=(
+ "#include \n"
+ "#include \n"
+ "int main(void) {\n"
+ ' printf("%.0f\\n", sqrt(144.0));\n'
+ " return 0;\n"
+ "}\n"
+ ),
+ expect_stdout="12",
+ tags=("libm", "c"),
+ ),
+ Probe(
+ id="c-compile-error",
+ lang="c",
+ source="int main(void) { return nope; }\n",
+ expect_ok=False,
+ expect_stage="compile",
+ expect_exit=None,
+ expect_stderr_contains=("nope",),
+ tags=("compile", "c"),
+ ),
+ Probe(
+ id="c-screen-system",
+ lang="c",
+ source='#include \nint main(void){ system("true"); return 0; }\n',
+ kind="screen",
+ screen_needle="processes",
+ tags=("screen", "c"),
+ ),
+ Probe(
+ id="c-screen-no-main",
+ lang="c",
+ source="int add(int a){ return a + 1; }\n",
+ kind="screen",
+ screen_needle="main()",
+ tags=("screen", "c"),
+ ),
+
+ # --- cpp ---------------------------------------------------------------
+ Probe(
+ id="cpp-hello",
+ lang="cpp",
+ source=(
+ "#include \n"
+ "int main() {\n"
+ ' std::cout << "cpp-probe " << (6 * 7) << "\\n";\n'
+ " return 0;\n"
+ "}\n"
+ ),
+ expect_stdout="cpp-probe 42",
+ tags=("smoke", "cpp"),
+ ),
+ Probe(
+ id="cpp-alias",
+ lang="c++",
+ source=(
+ "#include \n"
+ "int main(){ std::cout << 9 << \"\\n\"; }\n"
+ ),
+ expect_stdout="9",
+ tags=("alias", "cpp"),
+ ),
+ Probe(
+ id="cpp-screen-socket",
+ lang="cpp",
+ source="#include \nint main(){ return 0; }\n",
+ kind="screen",
+ screen_needle="network",
+ tags=("screen", "cpp"),
+ ),
+
+ # --- rust --------------------------------------------------------------
+ Probe(
+ id="rust-hello",
+ lang="rust",
+ source='fn main() { println!("rust-probe {}", (1..=10).sum::()); }\n',
+ expect_stdout="rust-probe 55",
+ tags=("smoke", "rust"),
+ ),
+ Probe(
+ id="rust-alias-rs",
+ lang="rs",
+ source='fn main() { println!("rs"); }\n',
+ expect_stdout="rs",
+ tags=("alias", "rust"),
+ ),
+ Probe(
+ id="rust-compile-error",
+ lang="rust",
+ source="fn main() { let x: i32 = \"nope\"; }\n",
+ expect_ok=False,
+ expect_stage="compile",
+ expect_exit=None,
+ tags=("compile", "rust"),
+ ),
+ Probe(
+ id="rust-screen-command",
+ lang="rust",
+ source='fn main() { let _ = std::process::Command::new("true"); }\n',
+ kind="screen",
+ screen_needle="processes",
+ tags=("screen", "rust"),
+ ),
+
+ # --- erlang ------------------------------------------------------------
+ Probe(
+ id="erl-escript",
+ lang="erlang",
+ source='main(_) -> io:format("erl-probe ~p~n", [lists:sum(lists:seq(1, 10))]).\n',
+ expect_stdout="erl-probe 55",
+ tags=("smoke", "erlang"),
+ ),
+ Probe(
+ id="erl-module",
+ lang="erlang",
+ source=(
+ "-module(main).\n"
+ "-export([main/1]).\n"
+ 'main(_) -> io:format("~p~n", [7 * 6]).\n'
+ ),
+ expect_stdout="42",
+ tags=("module", "erlang"),
+ ),
+ Probe(
+ id="erl-alias",
+ lang="erl",
+ source='main(_) -> io:format("alias~n").\n',
+ expect_stdout="alias",
+ tags=("alias", "erlang"),
+ ),
+ Probe(
+ id="erl-screen-os-cmd",
+ lang="erlang",
+ source='main(_) -> os:cmd("true").\n',
+ kind="screen",
+ screen_needle="processes",
+ tags=("screen", "erlang"),
+ ),
+)
+
+
+def probes_by_tag(*tags: str) -> tuple[Probe, ...]:
+ wanted = set(tags)
+ return tuple(p for p in PROBES if wanted.intersection(p.tags))
+
+
+def covered_langs() -> set[str]:
+ """Canonical RUN_LANGS keys touched by at least one probe (aliases resolved)."""
+ from synapse import code_run
+ return {code_run.resolve_lang(p.lang) for p in PROBES}
diff --git a/tests/test_code_run.py b/tests/test_code_run.py
new file mode 100644
index 0000000..9a549da
--- /dev/null
+++ b/tests/test_code_run.py
@@ -0,0 +1,308 @@
+"""The execution track: synapse/code_run.py.
+
+Python is the only toolchain guaranteed present (it is the interpreter running
+these tests), so it carries the behavioural coverage — limits, streams, exit
+codes, isolation. The compiled languages are covered for the parts that hold
+without their toolchain installed (screening, argv shape, the missing-toolchain
+message) and skipped where they need it, so this file passes on a machine with
+no clang and no rustc as well as on one with both.
+"""
+import shutil
+import sys
+
+import pytest
+
+from synapse import code_run
+
+
+def _requires(lang):
+ entry = code_run.RUN_LANGS[lang]
+ return pytest.mark.skipif(
+ not entry["tool"](), reason=f"no {lang} toolchain on this machine"
+ )
+
+
+# ---------------------------------------------------------------------------
+# Registry
+# ---------------------------------------------------------------------------
+
+def test_every_lang_has_a_complete_driver():
+ for name, entry in code_run.RUN_LANGS.items():
+ assert entry["summary"], name
+ assert callable(entry["tool"]), name
+ assert callable(entry["source_name"]), name
+ assert callable(entry["run"]), name
+ assert callable(entry["critique"]), name
+ assert entry["install"], name
+
+
+def test_aliases_resolve_to_real_langs():
+ for alias, target in code_run.ALIASES.items():
+ assert target in code_run.RUN_LANGS, alias
+ assert code_run.resolve_lang("C++") == "cpp"
+ assert code_run.resolve_lang(" Py ") == "python"
+ assert code_run.resolve_lang("javascript") == "javascript" # unknown passes through
+
+
+def test_erlang_source_name_follows_the_dialect():
+ """escript reads the same bytes two ways depending on the extension. Writing
+ a bare main/1 script to main.erl makes it a module with no exports, which
+ fails at load with an error that says nothing about the real problem."""
+ assert code_run._erlang_source_name("main(_) -> ok.") == "main.escript"
+ assert code_run._erlang_source_name("-module(main).\nmain(_) -> ok.") == "main.erl"
+
+
+def test_erlang_escript_gets_a_shebang_when_missing():
+ """OTP 28+ rejects shebang-less .escript with 'Premature end of file'."""
+ bare = "main(_) -> ok.\n"
+ out = code_run._erlang_write_source(bare)
+ assert out.startswith("#!/usr/bin/env escript\n")
+ assert out.endswith(bare)
+ already = "#!/usr/bin/env escript\nmain(_) -> ok.\n"
+ assert code_run._erlang_write_source(already) == already
+ module = "-module(main).\n-export([main/1]).\nmain(_) -> ok.\n"
+ assert code_run._erlang_write_source(module) == module
+
+
+# ---------------------------------------------------------------------------
+# Screening
+# ---------------------------------------------------------------------------
+
+def test_clean_snippets_pass_screening():
+ assert code_run.critique("python", "print(sum(range(100)))") == []
+ assert code_run.critique("c", '#include \nint main(void){puts("x");}') == []
+ assert code_run.critique("rust", 'fn main(){ println!("x"); }') == []
+ assert code_run.critique("erlang", 'main(_) -> io:format("x~n").') == []
+
+
+@pytest.mark.parametrize("lang,source,needle", [
+ ("python", "import socket\nprint(socket)", "network"),
+ ("python", "import requests\nprint(requests)", "network"),
+ ("python", "import subprocess\nsubprocess.run(['ls'])", "processes"),
+ ("c", "#include \nint main(void){return 0;}", "network"),
+ ("c", '#include \nint main(void){system("ls");}', "processes"),
+ ("rust", "fn main(){ std::process::Command::new(\"ls\"); }", "processes"),
+ ("erlang", 'main(_) -> os:cmd("ls").', "processes"),
+])
+def test_screening_catches_the_obvious_reaches(lang, source, needle):
+ issues = code_run.critique(lang, source)
+ assert issues, f"{lang} snippet passed screening"
+ assert any(needle in i for i in issues), issues
+
+
+def test_screening_catches_absolute_paths_and_urls():
+ assert any("absolute" in i for i in code_run.critique(
+ "python", 'data = open("/etc/passwd").read()\nprint(data)'))
+ # Escaped and raw Windows paths both count.
+ assert any("absolute" in i for i in code_run.critique(
+ "python", 'data = open("C:\\\\Users\\\\me").read()'))
+ assert any("absolute" in i for i in code_run.critique(
+ "python", r'data = open(r"C:\Users\me").read()'))
+ assert any("http" in i for i in code_run.critique(
+ "python", 'print("see https://example.com/data.csv")'))
+
+
+def test_url_in_a_comment_is_not_a_network_reach():
+ """A citation in a comment is not fetch(); bouncing it costs a useless retry."""
+ c = (
+ "/* spec: https://en.cppreference.com/w/c/string */\n"
+ "#include \n"
+ "int main(void){ puts(\"ok\"); return 0; }\n"
+ )
+ assert code_run.critique("c", c) == []
+
+
+def test_erlang_nproc_ceiling_is_relative_to_the_user(monkeypatch):
+ """RLIMIT_NPROC is UID-scoped; an absolute 512 is not 512 of headroom."""
+ monkeypatch.setattr(code_run, "_user_process_count", lambda: 400)
+ assert code_run._max_procs_for("erlang") == 400 + code_run._ERLANG_PROC_HEADROOM
+ assert code_run._max_procs_for("python") == code_run._MAX_PROCS
+ monkeypatch.setattr(code_run, "_user_process_count", lambda: None)
+ assert code_run._max_procs_for("erlang") is None
+
+
+def test_missing_entry_point_is_named_up_front():
+ """Without this the user sees a linker error about _main, or an escript load
+ failure — neither of which says 'you forgot the entry point'."""
+ assert any("main()" in i for i in code_run.critique("c", "int add(int a){return a+1;}"))
+ assert any("main()" in i for i in code_run.critique("rust", "fn add(a: i32) -> i32 { a }"))
+ assert any("main/1" in i for i in code_run.critique("erlang", "add(A) -> A + 1."))
+
+
+def test_empty_source_says_so_and_stops():
+ issues = code_run.critique("python", " ")
+ assert issues == ["source is empty or too short to run."]
+
+
+def test_unknown_lang_is_refused_by_both_entry_points():
+ assert code_run.critique("brainfuck", "+++.")[0].startswith("cannot run")
+ assert code_run.run("brainfuck", "+++.")["ok"] is False
+
+
+def test_oversized_source_is_refused_without_running():
+ issues = code_run.critique("python", "x = 1\n" * 40_000)
+ assert issues and "characters" in issues[0]
+
+
+# ---------------------------------------------------------------------------
+# Execution (python: always available)
+# ---------------------------------------------------------------------------
+
+def test_stdout_and_exit_code_come_back():
+ out = code_run.run("python", "print('hello', 6 * 7)")
+ assert out["ok"] is True
+ assert out["stdout"].strip() == "hello 42"
+ assert out["exit_code"] == 0
+
+
+def test_streams_are_kept_separate():
+ out = code_run.run("python", "import sys\nprint('o')\nprint('e', file=sys.stderr)")
+ assert out["stdout"].strip() == "o"
+ assert out["stderr"].strip() == "e"
+
+
+def test_nonzero_exit_is_still_a_successful_run():
+ out = code_run.run("python", "raise SystemExit(3)")
+ assert out["ok"] is True
+ assert out["exit_code"] == 3
+
+
+def test_a_traceback_is_returned_not_raised():
+ out = code_run.run("python", "print(1/0)")
+ assert out["ok"] is True
+ assert out["exit_code"] != 0
+ assert "ZeroDivisionError" in out["stderr"]
+
+
+def test_stdin_is_piped_in():
+ out = code_run.run("python", "import sys\nprint(sys.stdin.read().upper())", stdin="abc")
+ assert out["stdout"].strip() == "ABC"
+
+
+def test_a_program_reading_stdin_with_none_given_does_not_hang():
+ """stdin is always closed rather than inherited: a snippet calling input()
+ with nothing piped in would otherwise block on the server's own stdin until
+ the wall clock killed it, five seconds of 'thinking' for nothing."""
+ out = code_run.run("python", "print(input('prompt: '))")
+ assert out["ok"] is True
+ assert "EOFError" in out["stderr"]
+
+
+def test_an_infinite_loop_is_killed_and_explained():
+ out = code_run.run("python", "while True:\n pass")
+ assert out["ok"] is False
+ assert out["stage"] == "run"
+ assert "did not finish" in out["error"]
+
+
+def test_output_is_truncated_not_streamed_whole():
+ out = code_run.run("python", f"print('x' * {code_run.MAX_OUTPUT * 3})")
+ assert len(out["stdout"]) < code_run.MAX_OUTPUT + 200
+ assert "truncated" in out["stdout"]
+
+
+def test_the_working_directory_is_ephemeral():
+ """Two runs must not see each other's files. The temp dir is the only thing
+ standing between a snippet and the user's cwd, so its lifetime is worth a
+ test rather than an assumption."""
+ first = code_run.run("python", "open('note.txt', 'w').write('hi')\nprint('wrote')")
+ assert first["ok"] is True, first
+ second = code_run.run("python", "import os\nprint(os.path.exists('note.txt'))")
+ assert second["stdout"].strip() == "False"
+
+
+def test_the_child_does_not_inherit_the_servers_environment():
+ """A snippet has no business reading the process environment it happens to
+ be spawned from — that is where an API key would be."""
+ import os
+ os.environ["NEXUS_RUN_LEAK_PROBE"] = "secret"
+ try:
+ out = code_run.run(
+ "python", "import os\nprint(os.environ.get('NEXUS_RUN_LEAK_PROBE'))"
+ )
+ finally:
+ os.environ.pop("NEXUS_RUN_LEAK_PROBE", None)
+ assert out["stdout"].strip() == "None"
+
+
+def test_home_points_at_the_scratch_dir():
+ """HOME is rewritten so a snippet writing a dotfile writes it somewhere that
+ gets deleted, rather than into the user's real home."""
+ out = code_run.run("python", "import os\nprint(os.path.expanduser('~'))")
+ assert "nexus-run-" in out["stdout"]
+
+
+# ---------------------------------------------------------------------------
+# Toolchains
+# ---------------------------------------------------------------------------
+
+def test_a_missing_toolchain_explains_itself(monkeypatch):
+ """The model has to be able to tell 'you cannot run this here' from 'your
+ code is wrong' — otherwise it rewrites a correct program repeatedly."""
+ monkeypatch.setitem(
+ code_run.RUN_LANGS["rust"], "tool", lambda: None
+ )
+ out = code_run.run("rust", 'fn main(){ println!("x"); }')
+ assert out["ok"] is False
+ assert out["stage"] == "toolchain"
+ assert "rustup.rs" in out["error"]
+ assert "Show the code instead" in out["error"]
+
+
+@_requires("c")
+def test_c_compiles_and_runs():
+ out = code_run.run("c", '#include \nint main(void){printf("%d\\n", 6*7);return 0;}')
+ assert out["ok"] is True, out
+ assert out["stdout"].strip() == "42"
+
+
+@_requires("c")
+def test_a_compile_error_comes_back_as_a_compile_error():
+ """Stage matters: the compiler's diagnostics are the useful payload, and
+ labelling this a run failure would hide that nothing ever executed."""
+ out = code_run.run("c", "int main(void){ return oops; }")
+ assert out["ok"] is False
+ assert out["stage"] == "compile"
+ assert "oops" in out["stderr"]
+
+
+@_requires("cpp")
+def test_cpp_compiles_and_runs():
+ out = code_run.run("cpp", '#include \nint main(){std::cout << 6*7 << "\\n";}')
+ assert out["ok"] is True, out
+ assert out["stdout"].strip() == "42"
+
+
+@_requires("rust")
+def test_rust_compiles_and_runs():
+ out = code_run.run("rust", 'fn main(){ println!("{}", (1..=10).sum::()); }')
+ assert out["ok"] is True, out
+ assert out["stdout"].strip() == "55"
+
+
+@_requires("erlang")
+def test_erlang_runs_a_bare_escript():
+ out = code_run.run("erlang", 'main(_) -> io:format("~p~n", [lists:sum(lists:seq(1,10))]).')
+ assert out["ok"] is True, out
+ assert out["stdout"].strip() == "55"
+
+
+@_requires("erlang")
+def test_erlang_runs_a_module_form_script():
+ out = code_run.run(
+ "erlang",
+ "-module(main).\n-export([main/1]).\nmain(_) -> io:format(\"~p~n\", [7*6]).",
+ )
+ assert out["ok"] is True, out
+ assert out["stdout"].strip() == "42"
+
+
+@pytest.mark.skipif(sys.platform != "linux", reason="unshare is Linux-only")
+def test_network_isolation_probe_is_honest():
+ """Either we got a namespace or we did not — but the probe must never claim
+ one it did not verify, because the tool description promises 'no network' on
+ the strength of it."""
+ prefix = code_run._net_isolation()
+ assert prefix in ([], ["unshare", "-rn"])
+ if prefix:
+ assert shutil.which("unshare")
diff --git a/tests/test_snippet_probes.py b/tests/test_snippet_probes.py
new file mode 100644
index 0000000..d5f8797
--- /dev/null
+++ b/tests/test_snippet_probes.py
@@ -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))
diff --git a/tests/test_tools.py b/tests/test_tools.py
index 81be8e2..fb17aa7 100644
--- a/tests/test_tools.py
+++ b/tests/test_tools.py
@@ -10,6 +10,7 @@ import asyncio
import json
from synapse import tools
+from synapse import code_run
from synapse.chat import _run_tool_loop
@@ -347,6 +348,118 @@ def test_preview_langs_match_the_frontend_registry():
)
+_FRONTEND_RUN_REGISTRY = ("interface", "web", "src", "preview", "run-langs.js")
+
+
+def _frontend_registry_keys(parts: tuple, name: str) -> list:
+ """Top-level keys of a `export const = {...}` object literal."""
+ import re
+ from pathlib import Path
+ src = Path(__file__).resolve().parents[1].joinpath(*parts)
+ text = src.read_text(encoding="utf-8")
+ body = re.search(rf"^export const {name} = \{{\n(.*?)^\}};", text, re.S | re.M)
+ assert body, f"could not find a {name} object literal in {src}"
+ return re.findall(r"^ (\w+):", body.group(1), re.M)
+
+
+def test_run_langs_match_the_frontend_registry():
+ """Same failure mode as the preview registries, one track over: a language
+ the backend can run but the frontend does not know about renders as raw JSON
+ in the chat, and one the frontend labels but the backend refuses produces a
+ tool error the user never asked for. Nothing at runtime couples them."""
+ assert _frontend_registry_keys(_FRONTEND_RUN_REGISTRY, "RUN_LANGS") == list(
+ code_run.RUN_LANGS
+ ), (
+ "RUN_LANGS differs between synapse/code_run.py and "
+ "interface/web/src/preview/run-langs.js - add the language to both."
+ )
+
+
+def test_run_fence_tag_matches_the_frontend():
+ """The tag is the handshake: run_snippet emits it, Markdown.jsx dispatches on
+ it. A mismatch shows the JSON envelope to the user as a code block."""
+ import re
+ from pathlib import Path
+ src = Path(__file__).resolve().parents[1].joinpath(*_FRONTEND_RUN_REGISTRY)
+ m = re.search(r'export const RUN_FENCE_LANG = "([^"]+)"', src.read_text(encoding="utf-8"))
+ assert m and m.group(1) == tools._RUN_FENCE_LANG
+
+
+def test_run_lang_enum_is_derived_not_repeated():
+ schema, _ = tools.REGISTRY["run_snippet"]
+ enum = schema["function"]["parameters"]["properties"]["lang"]["enum"]
+ assert enum == list(code_run.RUN_LANGS)
+
+
+def test_run_snippet_is_an_action_tool():
+ """It executes code on the host, so action_tool_policy has to gate it.
+ Slipping into STANDING_TOOLS (where render_preview lives, ungated) would make
+ every 'run this' a subprocess with no consent step anywhere."""
+ assert tools.is_action("run_snippet")
+ assert "run_snippet" not in tools.STANDING_TOOLS
+ assert "run_snippet" in tools.CUED_ACTION_TOOLS
+ # ...and withholding actions has to actually withhold it.
+ assert tools.schemas_for(["run_snippet"], allow_actions=False) == []
+
+
+def test_wants_code_run_needs_a_verb_not_a_language():
+ """A language name must not arm the run track. `python` in _RUN_HINTS would
+ drag every mention of the language into a non-stream tool round - the exact
+ 'stuck thinking' problem that kept render_preview off by default."""
+ assert tools.wants_code_run("run this and show me the output")
+ assert tools.wants_code_run("does this compile?")
+ assert not tools.wants_code_run("write me a python function that sorts a list")
+ assert not tools.wants_code_run("explain how rust ownership works")
+
+
+def test_run_snippet_rejects_a_preview_language():
+ out = json.loads(asyncio.run(tools.dispatch("run_snippet", {
+ "lang": "html", "source": "
hello there
",
+ })))
+ assert out["ok"] is False
+ assert "render_preview" in out["error"]
+
+
+def test_run_snippet_fence_survives_backticks_in_the_source():
+ """A backtick in the source would close the ```nexus-run fence early, and the
+ rest of the envelope would spill into the chat as prose."""
+ out = json.loads(asyncio.run(tools.dispatch("run_snippet", {
+ "lang": "python", "source": "s = '``` still inside'\nprint(len(s))",
+ })))
+ assert out["ok"] is True, out
+ body = out["fence"].split("\n", 1)[1].rsplit("\n", 1)[0]
+ assert "```" not in body
+ assert json.loads(body)["source"].startswith("s = '```")
+
+
+def test_run_snippet_reports_a_program_that_fails():
+ """A non-zero exit is a successful run, not a tool failure: its stderr is the
+ answer. Reporting ok=False here would send the model into a retry loop over
+ a program that did exactly what it was asked to demonstrate."""
+ out = json.loads(asyncio.run(tools.dispatch("run_snippet", {
+ "lang": "python",
+ "source": "import sys\nprint('before')\nsys.exit(2)",
+ })))
+ assert out["ok"] is True
+ assert out["exit_code"] == 2
+ assert "before" in out["stdout"]
+
+
+def test_run_snippet_escalates_a_scaffold_on_retry():
+ """Same discipline as render_preview: first reject is issues-only; second
+ gets a pattern. _attempt is supplied by the tool loop."""
+ first = json.loads(asyncio.run(tools.dispatch("run_snippet", {
+ "lang": "python", "source": "import socket\nprint(1)", "_attempt": 0,
+ })))
+ assert first["ok"] is False
+ assert "scaffold" not in first
+ second = json.loads(asyncio.run(tools.dispatch("run_snippet", {
+ "lang": "python", "source": "import socket\nprint(1)", "_attempt": 1,
+ })))
+ assert second["ok"] is False
+ assert "scaffold" in second and "print" in second["scaffold"]
+
+
def test_preview_lang_enum_is_derived_not_repeated():
schema, _ = tools.REGISTRY["render_preview"]
enum = schema["function"]["parameters"]["properties"]["lang"]["enum"]