Neosyntropy/runtime-operators

NeoSyntropy Runtime Operators

Overfit the graph, not the benchmark.

Language models are good at proposing actions. Production software must decide which actions are legal, what evidence is required, and when a task is complete.

NeoSyntropy Runtime Operators is an experimental model family for testing one idea: small, specialized models can become reliable software-engineering workers when they are trained for narrow state-machine operators and executed inside an application-owned graph.

This repository is currently a research card and evaluation specification. It does not contain trained weights or make benchmark-performance claims yet.

The model family

The graph composes several model roles instead of asking one unconstrained model to own the entire workflow.

Model	Responsibility
Runtime Structure	Convert observations into application-owned schemas
Runtime Route	Propose a legal next node from declared candidates
Runtime Deterministic Reasoning	Perform a bounded reasoning step and declare tool calls
Runtime Stochastic Reasoning	Generate alternative plans or repairs inside an allowed search space
Runtime Guard	Validate claims against rules, tests, and supplied evidence
Runtime Score	Produce rubric-grounded measurements for evaluation and selection

The planned unified operator adapter conditions these roles with explicit tokens such as <OPERATOR:UNDERSTAND>, <OPERATOR:PROPOSE>, and <OPERATOR:REPAIR>. NeoSyntropy remains responsible for execution, transition legality, state commits, and side effects.

The operator graph

                            ┌───────────────┐
                            │   RETRIEVE    │◄──────────────┐
                            └───────┬───────┘               │
                                    │ evidence              │ need information
                                    ▼                       │
┌────────────┐  requirements  ┌─────────────┐  candidates  ┌┴───────────┐
│ UNDERSTAND │───────────────►│  DECOMPOSE  │─────────────►│  PROPOSE   │
└────────────┘                └─────────────┘              └─────┬──────┘
                                                                │ selected plan
                                                                ▼
┌────────────┐  complete   ┌─────────────┐  evidence      ┌─────────────┐
│  SUCCESS   │◄────────────│   VERIFY    │◄───────────────│   OBSERVE   │
└────────────┘             └──────┬──────┘                └──────▲──────┘
                                  │ failed test                   │ result
                                  ▼                               │
                           ┌─────────────┐  corrected plan  ┌─────┴───────┐
                           │   REPAIR    │─────────────────►│   EXECUTE   │
                           └─────────────┘                  └─────────────┘

Complete node declarations

The following definitions make every symbol in the graph explicit. The three tool names are application integrations: search_repository, apply_plan, and run_tests. Learned operators use schema-constrained model calls; execution and verification stay in trusted Python handlers.

from typing import Any, Literal

from pydantic import BaseModel, ConfigDict

from neosyntropy import NodeContext, OpenInput, SchemaNode, TextOutput, node

Signal = Literal[
    "CONTINUE",
    "NEED_INFORMATION",
    "EXECUTED",
    "FAILED_TEST",
    "WRONG_PLAN",
    "MEMORY_PRESSURE",
    "COMPLETE",
]


class StrictModel(BaseModel):
    model_config = ConfigDict(extra="forbid")


class TaskInput(StrictModel):
    repository: str
    issue: str


class UnderstandOutput(StrictModel):
    requirements: list[str]
    unknowns: list[str]
    signal: Literal["CONTINUE", "NEED_INFORMATION"]


class DecomposeOutput(StrictModel):
    task_tree: list[str]
    search_queries: list[str]
    signal: Literal["CONTINUE", "NEED_INFORMATION"]


class RetrieveOutput(StrictModel):
    retrieved: list[str]
    signal: Literal["CONTINUE", "NEED_INFORMATION"]


class PlanOutput(StrictModel):
    current_plan: str
    signal: Literal["CONTINUE", "NEED_INFORMATION"]


class ExecuteOutput(StrictModel):
    executed: bool
    summary: str
    signal: Literal["EXECUTED"]


class ObserveOutput(StrictModel):
    observations: list[str]
    errors: list[str]
    signal: Literal["CONTINUE", "MEMORY_PRESSURE"]


class Verification(StrictModel):
    passed: bool
    summary: str


class VerifyOutput(StrictModel):
    verification: Verification
    evidence: list[str]
    signal: Literal[
        "COMPLETE", "FAILED_TEST", "NEED_INFORMATION", "WRONG_PLAN"
    ]


class CompressOutput(StrictModel):
    observations: list[str]
    errors: list[str]
    history: list[str]
    signal: Literal["CONTINUE"]


class SuccessOutput(StrictModel):
    outcome: Literal["satisfies_spec"]


# Learned operator: infer explicit requirements without inventing facts.
understand = SchemaNode(
    id="Understand",
    input_schema=TaskInput,
    output_schema=UnderstandOutput,
    prompt=(
        "<OPERATOR:UNDERSTAND> Read the repository issue and current state. "
        "Return explicit requirements, unresolved unknowns, and a legal signal."
    ),
    metadata={"operator": "UNDERSTAND", "model_role": "runtime-structure"},
)


# Learned operator: turn requirements into an ordered task tree and searches.
decompose = SchemaNode(
    id="Decompose",
    input_schema=OpenInput,
    output_schema=DecomposeOutput,
    prompt=(
        "<OPERATOR:DECOMPOSE> Decompose the goal into ordered, testable subgoals. "
        "Generate only repository searches needed by those subgoals."
    ),
    prerequisites=("Understand",),
    metadata={"operator": "DECOMPOSE", "model_role": "runtime-deterministic-reasoning"},
)


# Trusted tool operator: execute only repository searches already in state.
@node(
    id="Retrieve",
    input_schema=OpenInput,
    output_schema=RetrieveOutput,
    tools=("search_repository",),
)
def retrieve(ctx: NodeContext):
    facts: list[str] = []
    for query in ctx.state.get("search_queries", []):
        result = ctx.tools.invoke("search_repository", {"query": query})
        facts.extend(result if isinstance(result, list) else [str(result)])
    signal = "CONTINUE" if facts else "NEED_INFORMATION"
    output = {"retrieved": facts, "signal": signal}
    return ctx.result(output=output, state_updates=output)


# Learned operator: propose one bounded implementation plan from evidence.
propose = SchemaNode(
    id="Propose",
    input_schema=OpenInput,
    output_schema=PlanOutput,
    prompt=(
        "<OPERATOR:PROPOSE> Use requirements, task_tree, and retrieved evidence. "
        "Return one implementation plan grounded in repository symbols."
    ),
    metadata={"operator": "PROPOSE", "model_role": "runtime-stochastic-reasoning"},
)


# Trusted tool operator: apply the selected plan and run the real test harness.
@node(
    id="Execute",
    input_schema=OpenInput,
    output_schema=ExecuteOutput,
    tools=("apply_plan", "run_tests"),
)
def execute(ctx: NodeContext):
    plan = ctx.state["current_plan"]
    ctx.tools.invoke("apply_plan", {"plan": plan})
    last_run = ctx.tools.invoke("run_tests", {})
    output = {
        "executed": True,
        "summary": str(last_run),
        "signal": "EXECUTED",
    }
    return ctx.result(output=output, state_updates={**output, "last_run": last_run})


# Trusted operator: normalize tool output into observations and errors.
@node(id="Observe", input_schema=OpenInput, output_schema=ObserveOutput)
def observe(ctx: NodeContext):
    run = ctx.state.get("last_run", {})
    errors = list(run.get("errors", [])) if isinstance(run, dict) else []
    observations = [str(run)]
    signal = "MEMORY_PRESSURE" if len(observations) > 20 else "CONTINUE"
    output = {"observations": observations, "errors": errors, "signal": signal}
    return ctx.result(output=output, state_updates=output)


# Trusted gate: only real test evidence may produce COMPLETE.
@node(id="Verify", input_schema=OpenInput, output_schema=VerifyOutput)
def verify(ctx: NodeContext):
    run = ctx.state.get("last_run", {})
    passed = bool(isinstance(run, dict) and run.get("passed") is True)
    summary = "All required tests passed." if passed else "Required tests failed."
    output = {
        "verification": {"passed": passed, "summary": summary},
        "evidence": [str(run)],
        "signal": "COMPLETE" if passed else "FAILED_TEST",
    }
    return ctx.result(output=output, state_updates=output)


# Learned operator: repair the plan using concrete failure evidence.
repair = SchemaNode(
    id="Repair",
    input_schema=OpenInput,
    output_schema=PlanOutput,
    prompt=(
        "<OPERATOR:REPAIR> Read current_plan, errors, and verification evidence. "
        "Return a corrected plan that addresses the observed failure only."
    ),
    metadata={"operator": "REPAIR", "model_role": "runtime-stochastic-reasoning"},
)


# Trusted operator: bound state growth without rewriting facts.
@node(id="Compress", input_schema=OpenInput, output_schema=CompressOutput)
def compress_memory(ctx: NodeContext):
    output = {
        "observations": list(ctx.state.get("observations", []))[-4:],
        "errors": list(ctx.state.get("errors", []))[-4:],
        "history": list(ctx.state.get("history", []))[-12:],
        "signal": "CONTINUE",
    }
    return ctx.result(output=output, state_updates=output)


# Terminal node: reachable only through the verified_complete graph guard.
@node(id="Success", input_schema=OpenInput, output_schema=SuccessOutput)
def success(ctx: NodeContext):
    output = {"outcome": "satisfies_spec"}
    return ctx.result(output=output, state_updates=output)


@node(
    id="OperatorFallback",
    input_schema=OpenInput,
    output_schema=TextOutput,
    is_fallback=True,
)
def operator_fallback(ctx: NodeContext):
    return ctx.result(output={"message": "No legal operator transition."})

Schema-node outputs are validated before their fields are proposed as state updates. Python handlers explicitly return state_updates; mutating ctx.state does not commit anything.

Graph wiring

The same nodes are connected as a NeoSyntropy FSM:

from neosyntropy import END, FSM, edge_deterministic, edge_fallback

def signal_is(expected):
    return lambda state: state.get("signal") == expected

def verified_complete(state):
    verification = state.get("verification", {})
    return state.get("signal") == "COMPLETE" and verification.get("passed") is True

graph = FSM(
    entry=understand,
    nodes=[
        understand,
        decompose,
        retrieve,
        propose,
        execute,
        observe,
        verify,
        repair,
        compress_memory,
        success,
        operator_fallback,
    ],
    edges=[
        edge_deterministic("Understand", "Decompose", guard=signal_is("CONTINUE")),
        edge_deterministic("Understand", "Retrieve", guard=signal_is("NEED_INFORMATION")),
        edge_deterministic("Decompose", "Propose", guard=signal_is("CONTINUE")),
        edge_deterministic("Decompose", "Retrieve", guard=signal_is("NEED_INFORMATION")),
        edge_deterministic("Retrieve", "Propose", guard=signal_is("CONTINUE")),
        edge_deterministic("Propose", "Execute", guard=signal_is("CONTINUE")),
        edge_deterministic("Execute", "Observe", guard=signal_is("EXECUTED")),
        edge_deterministic("Observe", "Verify", guard=signal_is("CONTINUE")),
        edge_deterministic("Observe", "Compress", guard=signal_is("MEMORY_PRESSURE")),
        edge_deterministic("Verify", "Success", guard=verified_complete),
        edge_deterministic("Verify", "Repair", guard=signal_is("FAILED_TEST")),
        edge_deterministic("Verify", "Retrieve", guard=signal_is("NEED_INFORMATION")),
        edge_deterministic("Verify", "Decompose", guard=signal_is("WRONG_PLAN")),
        edge_deterministic("Repair", "Execute", guard=signal_is("CONTINUE")),
        edge_deterministic("Repair", "Retrieve", guard=signal_is("NEED_INFORMATION")),
        edge_deterministic("Compress", "Decompose", guard=signal_is("CONTINUE")),
        edge_deterministic("Success", END),
        edge_fallback("Understand", "OperatorFallback"),
        edge_fallback("Decompose", "OperatorFallback"),
        edge_fallback("Retrieve", "OperatorFallback"),
        edge_fallback("Propose", "OperatorFallback"),
        edge_fallback("Execute", "OperatorFallback"),
        edge_fallback("Observe", "OperatorFallback"),
        edge_fallback("Verify", "OperatorFallback"),
        edge_fallback("Repair", "OperatorFallback"),
        edge_fallback("Compress", "OperatorFallback"),
    ],
)

Edge guards are authoritative. A model may propose a route, but it cannot commit an undeclared transition or bypass a failed verification gate.

One shared state

Operators do not own isolated hidden memories. They read a projection of one auditable workflow state and return a validated state patch.

{
  "goal": "Fix the reported repository issue",
  "requirements": [],
  "task_tree": [],
  "current_plan": "",
  "retrieved": [],
  "observations": [],
  "errors": [],
  "evidence": [],
  "history": [],
  "scratch": {},
  "signal": "CONTINUE"
}

Each operator receives only the fields it needs:

Operator	Reads	Writes
UNDERSTAND	goal, initial context	requirements, unknowns, signal
DECOMPOSE	goal, requirements	task tree, signal
RETRIEVE	goal, errors, repository context	retrieved evidence, signal
PROPOSE	requirements, task tree, evidence	candidate plan, signal
EXECUTE	selected plan	trusted execution result
OBSERVE	execution result	observations, errors, signal
VERIFY	requirements, observations, tests	evidence, verification result, signal
REPAIR	current plan, errors, evidence	corrected plan, signal
COMPRESS	accumulated state	compact state patch, signal
SUCCESS	verified evidence	terminal outcome

EXECUTE, deterministic observation parsing, transition checks, and final success gates should remain trusted runtime operations. Learned models propose; the graph validates and commits.

What “overfit the graph” means

The phrase describes deliberate specialization to a stable execution protocol:

• fixed operator vocabulary;
• explicit input projections and output schemas;
• declared tools and legal transitions;
• repair loops driven by real test evidence;
• consistent prompts across teacher generation, training, and inference.

It does not mean training on benchmark test patches, hidden tests, or expected answers. Benchmark instances and repositories used for final evaluation must be kept out of training and teacher-label generation. The hypothesis is that a model can learn the reusable procedure while still generalizing to unseen issues.

SWE evaluation plan

The first experiment will compare the same base model and tool environment under four scaffolds:

1. Direct agent — one unconstrained model call loop.
2. Prompted operators — operator prompts without fine-tuning.
3. Trained operators — the operator adapter without graph enforcement.
4. NeoSyntropy graph — trained operators with schemas, guards, state, and verified transitions.

Evaluation will begin with the official SWE-bench harness for reproducibility, while newer or contamination-resistant suites should be used for primary generalization claims. Static benchmark scores will always be reported with the exact harness, model, scaffold, context limit, tool budget, retry budget, and task exclusions.

Primary metrics:

• resolved instances (pass@1);
• legal-transition rate;
• schema-valid output rate;
• tool-call success rate;
• verification precision and false-success rate;
• repair-loop recovery rate;
• model tokens, wall-clock time, tool calls, and cost per resolved task;
• average state transitions and repeated-transition rate;
• outcome consistency across repeated seeds.

The most important comparison is not only whether a task was solved, but whether the graph can reach the same or better result with smaller models, fewer wasted actions, and no unverified success transition.

Data-generation protocol

Training traces are generated from complete graph executions rather than isolated first-step prompts:

1. Sample an unseen repository task and construct the initial state.
2. Use a stronger teacher to label the current operator only.
3. Validate the output schema and reject invented tools or evidence.
4. Execute approved actions in the benchmark environment.
5. Record observations and test results as new evidence.
6. Route through the declared graph and continue until success or budget expiry.
7. Store each operator transition with its state projection, output, signal, provenance, and final task outcome.
8. Freeze repository-level train, validation, and test splits before fine-tuning.

This produces training examples for the decision that was actually available at each state, including failed attempts and evidence-grounded repairs.

Status

• Graph contract: specified.
• Role-model cards: published separately.
• Unified operator dataset: planned.
• Unified operator adapter: not trained yet.
• SWE baseline runs: not run yet.
• Benchmark claims: none.

Results, weights, datasets, and exact run manifests will be published only after reproducible evaluation.