Primitive Generator

primitive-generator

1. Purpose and scope

This skill produces the executable artifact of every Mobius simulation: the Python code that fills primitives[i].python_code in the SimSpec. The SimSpec arrives from simspec-author with the physics, simulation, and primitives subtrees populated, but with python_code fields empty. This skill writes that code, primitive by primitive, then hands the SimSpec back to the orchestrator. science-integrity runs immediately afterward and gates execution by Managed Agents.

Three things this skill does NOT do. It does not invent physics — physics-interpreter already chose the regime, the equations, and the approximations. It does not validate units, CFL, or conservation — those are five separate scripts shipped by science-integrity. It does not orchestrate primitive ordering or DAG resolution — that is the orchestrator's job using primitives[i].depends_on.

The skill is stateless across primitives: each primitive is generated independently, and depends_on ordering is enforced by the caller, not by this skill. This decouples primitive generation from graph topology and makes the skill reusable by any future Mobius DAG variant.

2. Input

A full SimSpec object on disk as simspec.json, or passed inline by the orchestrator. Required state:

• paper.* — only used for cache-key computation (abstractHash).
• physics.regime, physics.approximations, physics.governing_equations[*].domain_hints — drive pattern matching.
• primitives[*] — every entry has id, type, name, parameters, depends_on, viz_hint, but python_code is empty.
• simulation.rng_seed — injected into every stochastic primitive.
• simulation.solver_hints.scheme — used to disambiguate when multiple patterns could match.

If python_code is already non-empty for a primitive, the skill skips it (idempotent re-runs are safe).

3. Output

The same SimSpec object with python_code filled for every primitive. The skill also appends entries to integrity.checks[] of type ambiguity:

• One pass-status entry per primitive that resolved via Tier 1 (pattern match), with details.pattern_id.
• One warn-status entry per primitive that resolved via Tier 2 (LLM synthesis), with details.nearest_patterns[] for audit.
• One fail-status entry per primitive that failed both tiers, with details.last_error from the validator.

The orchestrator decides whether to retry, escalate, or abort based on integrity.overall.

3.bis Output format (mandatory)

You MUST emit your final output as a single fenced ``json block at the END of your reply, with NO prose after the closing fence. The orchestrator parses that block by regex (/``(?:json)?\s*\n([\s\S]*?)\n\s*`/) and ignores everything else in your text content. Writing intermediate files via code_execution to /mnt/user-data/outputs/ is fine, but the canonical artifact MUST be inlined as the final fenced JSON block of your reply.

The canonical artifact for primitive-generator (consumed by orchestrator.ts Phase C, which reads result.primitives[].python_code and merges it into simspec.primitives[]):

{
  "primitives": [
    {
      "id": "fraunhofer",
      "type": "fraunhofer_diffraction",
      "name": "Fraunhofer Double-Slit",
      "parameters": {
        "wavelength": { "value": 6.328e-7, "unit": "m" }
      },
      "python_code": "import numpy as np\n\ndef initialize(params):\n    ...\n",
      "viz_hint": { "representation": "2D-field", "palette_index": "c0" }
    }
  ]
}

Required per primitive: id (matches simspec.primitives[i].id) and python_code (non-empty, satisfies the runtime contract in §5). The other fields (type, name, parameters, viz_hint) MAY be passed through from the SimSpec for traceability — the orchestrator only merges python_code — but emitting them helps downstream debugging.

If a primitive cannot be generated (validator fails three times in a row, match-key cannot be built, regime missing), emit it WITHOUT python_code and with an explicit error field on that primitive entry:

{
  "primitives": [
    { "id": "fraunhofer", "error": "validator failed 3x: missing parameter 'lambda_m'" }
  ]
}

If the entire run is degenerate (no SimSpec, no primitives to generate), emit a single top-level error:

{ "error": "simspec.json missing; nothing to generate" }

DO NOT emit narration, summaries, or follow-up questions after the closing fence — they break the orchestrator's downstream consumption and get silently dropped.

4. Two-tier strategy

Tier 1 — pattern match (fast, deterministic, LLM-free)

For each primitive, build the match key:

key = (primitive.type, physics.regime, dominant_domain_hint, sorted(physics.approximations))

Walk every JSON file in patterns/. A pattern matches iff:

• pattern.match.type == primitive.type
• pattern.match.regime == physics.regime
• pattern.match.technique is in the SimSpec's solver scheme OR is the empty string
• pattern.match.approximations is a subset of physics.approximations

If exactly one pattern matches, instantiate its template_python by substituting {{param_name}} placeholders with values from primitive.parameters[param_name].value. If multiple patterns match, prefer the one with the longer approximations list (more specific wins). If none match, fall through to Tier 2.

The substitution is a strict text replacement: every placeholder must resolve to a parameter present in primitive.parameters, else the skill emits a fail integrity check and the primitive proceeds to Tier 2 with the missing-parameter error in context.

Tier 2 — LLM synthesis (fallback, cached)

When no pattern matches, invoke Opus 4.7 with a structured prompt:

1. The runtime contract (§5) verbatim.
2. The 2-3 nearest patterns (by Hamming distance over the match tuple) as few-shot examples.
3. The primitive's full spec (id, type, name, parameters, viz_hint).
4. The relevant physics subtree slice.
5. The simulation.solver_hints and simulation.rng_seed.

Cache the result keyed by sha256(canonical_json(primitive_spec) || canonical_json(physics)) so identical primitives across re-runs return identical code. Cache lives at patterns/_cache/ (gitignored, populated at runtime).

Tier 2 outputs are subject to the same validator as Tier 1 outputs. If validation fails, the skill retries up to 3 times with the failure message appended to the prompt as a previous_attempt_failed block.

5. The runtime contract

Every generated Python primitive must satisfy the following. Violations are caught by validate_primitive.py (§7) before the code reaches Managed Agents.

Function shape. Define exactly one of:

def initialize(params: dict) -> dict:
    """Sources, initial conditions, fixed media. Called once at t=t0."""
    ...

def run(state: dict, params: dict, dt: float) -> dict:
    """Step the simulation forward by dt. Called once per frame."""
    ...

state is the cumulative simulation state, a JSON-serializable dict. params carries the primitive's parameters as a flat float-keyed dict. dt is the wall clock advance for this step.

Imports. Whitelist enforced statically:

• Allowed: numpy, scipy, sympy, pint, mobius_runtime.
• Allowed stdlib: math, dataclasses, typing.
• Rejected: os, subprocess, socket, requests, urllib, http, ftp, pickle, eval, exec, __import__. Any rejected import is a hard validator fail.

Determinism. Stochastic primitives must read the seed from params["rng_seed"] and instantiate a numpy.random.Generator via np.random.default_rng(params["rng_seed"]). Never call numpy.random legacy functions, never call random.random(), never read time.time(). Context7 confirms default_rng is the canonical 2.x API; the legacy RandomState and module-level functions exist only for backward compatibility.

No I/O. No file reads except the runtime helper module. No network. No subprocess. The orchestrator wraps each generated python_code with a runner harness that calls initialize(params) or run(state, params, dt) and serializes the returned dict via print(json.dumps(...)). Your code MUST NOT contain bare print() calls — but it MUST return a dict from initialize or run. Diagnostics return inside the output dict under state["_diagnostics"]. Parameters arrive flattened from {value, unit, range, label} to scalar value (the harness extracts .value before calling your function), so write float(params["wavelength"]) not float(params["wavelength"]["value"]).

Required functions. Every primitive MUST define exactly one of:

• initialize(params: dict) -> dict — for source-style primitives that emit

initial conditions or fixed media. Called once at t=t0 with the flattened per-primitive parameters. Returns the seed state dict for downstream primitives.

• run(state: dict, params: dict, dt: float) -> dict — for evolution-style

primitives (operators, detectors, observables) that read merged upstream state, apply their transform, and return the updated state. Called once per frame. The orchestrator merges depends_on parents' return-value dicts into state before invoking, so anything an upstream primitive emitted is available here.

The orchestrator runs primitives in topologically-sorted depends_on order and chains return values through state. Two primitives at the same DAG level both see the same merged upstream state but DO NOT see each other's outputs.

Return shape. A JSON-serializable dict. Numpy arrays must be converted via .tolist() or, for arrays > 1000 elements, base64-encoded float32 buffers under the key state[""] = {"shape": [...], "dtype": "f4", "buffer_b64": "..."}. The viz-mapper skill knows how to decode this.

Idempotence. run(state, params, dt) called twice in a row with the same (state, params, dt) must return numerically identical dicts (relative tolerance 1e-12). The validator smoke-tests this when no RNG is involved.

6. The mobius_runtime helper module

Shipped at patterns/_runtime/__init__.py. Generated primitives import from this module exclusively for shared physics — they never reinvent FFT, RK4, or Metropolis sweeps. The module is part of the skill bundle; Managed Agents pip installs it from a local wheel built at session start.

The helper signatures:

| Function | Purpose | | --- | --- | | gauss_2d(N, L, waist, x0, y0, amp) | Returns complex (N, N) Gaussian beam profile on a square grid of side L. Centred at (x0, y0). | | fft_propagate(field, z, lambda_m, paraxial=True) | Fresnel angular-spectrum propagation by distance z. Direct port of wave-engine.js#propagateAngularSpectrum. Set paraxial=False for Rayleigh-Sommerfeld. | | fraunhofer(field, z, lambda_m) | Far-field FFT propagation, returns rescaled grid. Port of wave-engine.js#propagateFraunhofer. | | rk4_step(rhs, y, t, dt) | Standard 4th-order Runge-Kutta step. rhs(y, t) returns dy/dt. | | monte_carlo_sample(prob_fn, n, rng) | Rejection sampling on prob_fn, returns n samples. rng is a numpy.random.Generator. | | schroed_split_step(psi, V, dt, dx, hbar, m) | Split-step Fourier evolution of the time-dependent Schrödinger equation: half-step in V, full step in T (kinetic via FFT), half-step in V. | | ising_metropolis_sweep(lattice, beta, J, rng) | One Metropolis-Hastings sweep over a 2D Ising lattice, in-place. |

The optics helpers (gauss_2d, fft_propagate, fraunhofer) are direct ports of the existing spike-visual/shared/wave-engine.js (399 LOC, FFT + Fresnel + Fraunhofer + masks). The pattern library prefers these over reinventing FFT — every optics primitive in §9 calls mobius_runtime rather than writing its own propagator.

The signatures are stable; bodies will land in Phase 3. Stubs ship with this skill (file: patterns/_runtime/__init__.py) so the import audit passes during dry-run.

7. Validation

After every primitive is generated, the skill calls scripts/validate_primitive.py::validate_primitive(primitive, simspec). The validator does six things:

1. AST parse. python_code must be valid Python 3.12. A SyntaxError is fail.
2. Import audit. Every import and from ... import ... statement is checked against the whitelist. Any rejected name is fail.
3. Signature check. Exactly one of initialize(params) or run(state, params, dt) must be defined with the right argument names. Missing or extra positional args are fail.
4. Idempotence smoke test. For non-stochastic primitives, the validator imports the code in a sandboxed namespace, calls the entrypoint twice with the primitive's parameters, and asserts numpy.testing.assert_allclose with rtol=1e-12. Mismatch is fail.
5. Boundedness audit. Static AST scan for while True without break, recursion without depth limit, or unbounded list comprehensions. Each finding is a warn.
6. Parameter coverage. Every params["..."] key referenced by the code must exist in primitive.parameters with a unit. Missing keys are warn (the LLM may have hallucinated a parameter; the orchestrator decides whether to add it to simspec-author's output).

The validator returns {"status": "pass"|"warn"|"fail", "issues": [...], "ast_summary": {...}}. The skill retries Tier 2 up to 3 times on fail; warn is reported but does not block.

8. Examples

Three fully-worked primitive generations covering source, medium, and detector. Each example shows: the SimSpec primitive entry, the matched pattern, the generated Python (post-substitution), and the integrity checks.

8.1 Source — Gaussian laser beam

SimSpec primitive entry:

{
  "id": "laser-source",
  "type": "source",
  "name": "Coherent Gaussian laser",
  "parameters": {
    "wavelength": { "value": 6.328e-7, "unit": "m",  "range": [4.0e-7, 7.5e-7], "label": "lambda" },
    "waist":      { "value": 8.0e-4,   "unit": "m",  "range": [1.0e-4, 2.0e-3], "label": "w0" },
    "grid_size":  { "value": 256,      "unit": "1",  "range": [128, 512],       "label": "N" },
    "extent":     { "value": 2.0e-3,   "unit": "m",  "range": [1.0e-3, 5.0e-3], "label": "L" }
  },
  "depends_on": [],
  "viz_hint": { "representation": "2D-field", "palette_index": "c0" }
}

Matched pattern. patterns/wave_gaussian_source.json. Match key ("source", "linear", "wave", ["paraxial"]) matches the SimSpec key with physics.regime == "linear", dominant domain_hint == "wave", and approximations == ["paraxial", "thin-lens"].

Generated Python (post-substitution):

import numpy as np
from mobius_runtime import gauss_2d

def initialize(params: dict) -> dict:
    N = int(params["grid_size"])
    L = float(params["extent"])
    waist = float(params["waist"])
    wavelength = float(params["wavelength"])
    field = gauss_2d(N=N, L=L, waist=waist, x0=0.0, y0=0.0, amp=1.0)
    intensity = (field.real ** 2 + field.imag ** 2)
    total_power = float(intensity.sum() * (L / N) ** 2)
    return {
        "field_re": field.real.astype(np.float32).tolist(),
        "field_im": field.imag.astype(np.float32).tolist(),
        "wavelength": wavelength,
        "extent": L,
        "grid_size": N,
        "total_power": total_power,
    }

science-integrity will check. Units (wavelength must be m, waist m); CFL is N/A here (no time stepping); conservation: total_power returned for downstream comparison; AST parse + import audit pass.

8.2 Medium — thin lens

SimSpec primitive entry:

{
  "id": "converging-lens",
  "type": "medium",
  "name": "Thin converging lens",
  "parameters": {
    "focal_length": { "value": 0.05, "unit": "m", "range": [0.01, 0.5], "label": "f" }
  },
  "depends_on": ["laser-source"],
  "viz_hint": { "representation": "2D-field", "palette_index": "c1" }
}

Matched pattern. `pa

…

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: leventilo
• Source: leventilo/mobius
• License: MIT

Install and usage instructions live in the source repository linked above.