Add complete operator node declarations

README.md

@@ -69,10 +69,251 @@ legality, state commits, and side effects.
                            └─────────────┘                  └─────────────┘
 ```
 
-The same graph expressed as NeoSyntropy code:
+### 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.
+
+```python
+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:
 
 ```python
-from neosyntropy import END, FSM, edge_deterministic
+from neosyntropy import END, FSM, edge_deterministic, edge_fallback
 
 def signal_is(expected):
     return lambda state: state.get("signal") == expected
@@ -94,6 +335,7 @@ graph = FSM(
         repair,
         compress_memory,
         success,
+        operator_fallback,
     ],
     edges=[
         edge_deterministic("Understand", "Decompose", guard=signal_is("CONTINUE")),
@@ -113,6 +355,15 @@ graph = FSM(
         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"),
     ],
 )
 ```