Add finalize validation for committed bindings

Author: squld

CHANGELOG.md

@@ -14,6 +14,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Tool-boundary regression test ensuring recoverable `ToolBoundaryError` is wrapped into JSON payload.
 
 ### Changed
+- Finalized committed binding validation at step boundaries, added write-root dirty tracking for dotted `nh_assign`, and aligned prompt/docs with the new validation contract.
 - Redesigned `nh.scope()` around `mode` semantics (`"inherit"` default, `"replace"` for explicit replacement).
 - Scope prompt suffix arguments now use list-based forms: `system_prompt_suffix_fragments` and `user_prompt_suffix_fragments`.
 - Introduced `ExecutionRef` (`run_id`, `scope_id`, optional `step_id`) and renamed runtime accessor to `get_execution_ref`.

docs/natural-blocks.md

@@ -129,7 +129,7 @@ def detect_language(text: str):
     return language
 ```
 
-Type annotations on write bindings enable validation and coercion at commit time.
+Type annotations on write bindings enable validation and coercion at step finalization for values that are actually committed.
 
 ### Pydantic model write bindings
 

docs/philosophy.md

@@ -23,7 +23,7 @@ calculate_average([1, "2", "three", "cuatro", "五"])  # 3.0
 
 Binding functions like `<python_average>` appear in the prompt as a compact signature line. The LLM's pre-trained Python knowledge lets it reason about types, return values, and composition from the signature alone, without JSON Schema or protocol overhead. See [Tool exposure efficiency](#tool-exposure-efficiency) for the quantitative comparison with MCP and CLI tool exposure.
 
-With provider-backed executors, each Natural block is a single LLM call. A sentiment classifier whose write binding is typed as `Literal["positive", "negative", "neutral"]` rejects any output outside the declared set -- Pydantic validates the type annotation at runtime, not as a hint. The same mechanism applies to numeric extraction (`int`, `float`), structured parsing (Pydantic models), and any task where the judgment space is bounded. Because the host program owns the loop, a misclassified result can be retried, logged, or routed to a fallback -- all in ordinary Python.
+With provider-backed executors, each Natural block is a single LLM call. A sentiment classifier whose write binding is typed as `Literal["positive", "negative", "neutral"]` rejects any output outside the declared set -- Pydantic validates committed write bindings at step finalization, not as a hint. The same mechanism applies to numeric extraction (`int`, `float`), structured parsing (Pydantic models), and any task where the judgment space is bounded. Because the host program owns the loop, a misclassified result can be retried, logged, or routed to a fallback -- all in ordinary Python.
 
 With [coding agent backends](coding-agent-backends.md), the same boundary contract applies, but each Natural block becomes an autonomous agent execution. The agent can read files, run commands, and invoke skills -- while typed bindings enforce what crosses the boundary back to Python. The same `scope()` and `run()` context managers that structure human-written workflows are equally legible to a coding agent constructing workflows programmatically. When a coding agent operates inside a Natural block, binding functions appear as Python signatures in the prompt:
 

docs/specification.md

@@ -393,6 +393,7 @@ Eval tool:
 
 - `nh_eval(expression: str) -> object`
     - Evaluate a Python expression and return the result. Use to inspect values, call functions, and mutate objects in-place.
+    - In-place mutations performed via `nh_eval` are not runtime-validated.
     - If the evaluated expression is awaitable, it is awaited before returning.
 
 Binding tool:
@@ -422,13 +423,16 @@ Semantics of `nh_assign`:
     - Traverse attributes for each intermediate segment.
     - Assign using attribute assignment on the final segment.
     - Validation:
-        - Validate only when runtime type metadata is available; otherwise assign without validation.
+        - Validate only the final assigned field when runtime type metadata is available; otherwise assign without validation.
 
 Commit and mutation notes:
 
 - Commit selection is controlled only by `<:name>` bindings.
 - `<:name>` selects which top-level names are committed from `step_locals` into Python locals at Natural block boundaries.
-- Dotted mutation is independent of `<:name>`.
+- Dotted `nh_assign` on a write binding root marks that root as dirty for commit selection.
+- Dotted `nh_assign` on a read binding root does not participate in commit selection.
+- Final validation is applied only to committed write bindings at step finalization.
+- Dotted mutation on read bindings and in-place mutation via `nh_eval` are outside the final validation guarantee.
 
 Write tool return value:
 

src/nighthawk/runtime/runner.py

@@ -402,12 +402,42 @@ def _apply_bindings_and_validate_kind(
         step_outcome: StepOutcome,
         bindings: dict[str, object],
         allowed_step_kinds: tuple[StepKind, ...],
-    ) -> str:
+    ) -> tuple[str, dict[str, object]]:
         step_outcome_kind = step_outcome.kind
         if step_outcome_kind not in allowed_step_kinds:
             raise ExecutionError(f"Step '{step_outcome_kind}' is not allowed for this step. Allowed kinds: {allowed_step_kinds}")
-        step_context.step_locals.update(bindings)
-        return step_outcome_kind
+        if step_outcome_kind == "raise":
+            return step_outcome_kind, dict(bindings)
+        validated_bindings = self._validate_and_coerce_output_bindings(
+            step_context=step_context,
+            bindings=bindings,
+        )
+        step_context.step_locals.update(validated_bindings)
+        return step_outcome_kind, validated_bindings
+
+    def _validate_and_coerce_output_bindings(
+        self,
+        *,
+        step_context: StepContext,
+        bindings: dict[str, object],
+    ) -> dict[str, object]:
+        validated_binding_name_to_value = dict(bindings)
+        for binding_name in step_context.binding_commit_targets:
+            if binding_name not in validated_binding_name_to_value:
+                continue
+
+            expected_type = step_context.binding_name_to_type.get(binding_name)
+            if expected_type is None:
+                continue
+
+            try:
+                validated_binding_name_to_value[binding_name] = TypeAdapter(expected_type).validate_python(
+                    validated_binding_name_to_value[binding_name]
+                )
+            except Exception as exception:
+                raise ExecutionError(f"Output binding '{binding_name}' failed validation: {exception}") from exception
+
+        return validated_binding_name_to_value
 
     def _apply_step_oversight_if_needed(
         self,
@@ -477,7 +507,7 @@ async def _finalize_step(
         allow_awaitable_return: bool,
     ) -> StepEnvelope:
         try:
-            step_outcome_kind = self._apply_bindings_and_validate_kind(
+            step_outcome_kind, validated_bindings = self._apply_bindings_and_validate_kind(
                 step_context=preparation.step_context,
                 step_outcome=step_outcome,
                 bindings=bindings,
@@ -515,7 +545,7 @@ async def _finalize_step(
         return StepEnvelope(
             step_outcome=step_outcome,
             input_bindings=dict(preparation.input_binding_name_to_value),
-            bindings=bindings,
+            bindings=validated_bindings,
             return_value=return_value,
         )
 

src/nighthawk/runtime/step_context.py

@@ -32,7 +32,7 @@ class StepContext:
     """Mutable, per-step execution context passed to tools and executors.
 
     ``step_globals`` and ``step_locals`` are mutable dicts. All mutations to ``step_locals`` MUST go through :meth:`record_assignment` (for top-level name bindings) or through the dotted-path assignment in ``tools.assignment`` (which bumps ``step_locals_revision`` directly).
-    Direct dict writes bypass revision tracking and ``assigned_binding_names`` bookkeeping, which will cause incorrect commit behavior at Natural block boundaries.
+    Direct dict writes bypass revision tracking, ``assigned_binding_names``, and ``dirty_output_binding_names`` bookkeeping, which will cause incorrect commit behavior at Natural block boundaries.
     """
 
     step_id: str
@@ -49,6 +49,7 @@ class StepContext:
     processed_natural_program: str = ""
     binding_name_to_type: dict[str, object] = field(default_factory=dict)
     assigned_binding_names: set[str] = field(default_factory=set)
+    dirty_output_binding_names: set[str] = field(default_factory=set)
     step_locals_revision: int = 0
     tool_result_rendering_policy: ToolResultRenderingPolicy | None = None
 
@@ -61,6 +62,11 @@ def record_assignment(self, name: str, value: object) -> None:
         self.assigned_binding_names.add(name)
         self.step_locals_revision += 1
 
+    def record_output_binding_mutation(self, name: str) -> None:
+        """Record an in-place mutation affecting a committed output binding root."""
+        self.dirty_output_binding_names.add(name)
+        self.step_locals_revision += 1
+
 
 _step_context_stack_var: ContextVar[tuple[StepContext, ...]] = ContextVar(
     "nighthawk_step_context_stack",

src/nighthawk/runtime/step_executor.py

@@ -283,7 +283,7 @@ def _extract_bindings(
         """Extract committed bindings from the step context."""
         bindings: dict[str, object] = {}
         for name in binding_names:
-            if name in step_context.assigned_binding_names:
+            if name in step_context.assigned_binding_names or name in step_context.dirty_output_binding_names:
                 bindings[name] = step_context.step_locals[name]
         return bindings
 

src/nighthawk/tools/assignment.py

@@ -176,10 +176,13 @@ def _assign_value_to_target_path(
             guidance="Fix the target path so the referenced attributes are assignable, then retry.",
         )
 
-    # Dotted mutation bypasses record_assignment because commit selection is
-    # controlled only by <:name> bindings (top-level names).  See design.md
-    # Section 8.3 "Commit and mutation notes" for the distinction.
-    step_context.step_locals_revision += 1
+    # Dotted mutation bypasses record_assignment. Top-level rebinding still
+    # drives ordinary assignment tracking, while write-binding roots touched by
+    # dotted nh_assign are tracked separately for commit selection.
+    if root_name in step_context.binding_commit_targets:
+        step_context.record_output_binding_mutation(root_name)
+    else:
+        step_context.step_locals_revision += 1
 
     return {
         "target_path": target_path,

tests/execution/test_runtime.py

@@ -2,20 +2,59 @@
 import json
 from dataclasses import dataclass, field
 from pathlib import Path
+from typing import Annotated, Literal
 
 import pytest
+from pydantic import BaseModel, Field, model_validator
 
 import nighthawk as nh
 from nighthawk.errors import ExecutionError, NighthawkError
 from nighthawk.runtime.prompt import build_system_prompt
 from nighthawk.runtime.step_context import StepContext
-from nighthawk.runtime.step_contract import PassStepOutcome, ReturnStepOutcome, StepKind
+from nighthawk.runtime.step_contract import PassStepOutcome, ReturnStepOutcome, StepFinalResult, StepKind
 from tests.execution.stub_executor import StubExecutor
 
 GLOBAL_NUMBER = 7
 SHADOWED_NUMBER = 1
 
 
+class RuntimeChildModel(BaseModel):
+    value: int
+
+
+class RuntimeResultModel(BaseModel):
+    child: RuntimeChildModel | None = None
+    value: int = 0
+
+
+class RuntimeApproveProposal(BaseModel):
+    kind: Literal["approve"]
+    score: int
+
+
+class RuntimeRejectProposal(BaseModel):
+    kind: Literal["reject"]
+    reason: str
+
+
+RuntimeProposal = Annotated[RuntimeApproveProposal | RuntimeRejectProposal, Field(discriminator="kind")]
+
+
+class RuntimeProposalEnvelope(BaseModel):
+    proposals: list[RuntimeProposal]
+
+
+class RuntimePlanUpdateDecision(BaseModel):
+    should_update: bool
+    next_plan: str | None = None
+
+    @model_validator(mode="after")
+    def validate_next_plan(self) -> "RuntimePlanUpdateDecision":
+        if self.should_update and self.next_plan is None:
+            raise ValueError("next_plan is required when should_update is true")
+        return self
+
+
 def global_import_file(file_path: Path | str) -> str:
     _ = file_path
     return '{"step_outcome": {"kind": "pass"}, "bindings": {"result": 20}}'
@@ -113,6 +152,54 @@ async def f(x: int) -> int:
         assert asyncio.run(f(10)) == 11
 
 
+def test_pass_step_finalize_coerces_committed_model_binding() -> None:
+    with nh.run(StubExecutor()):
+
+        @nh.natural_function
+        def f() -> str:
+            """natural
+            <:result>
+            {"step_outcome": {"kind": "pass"}, "bindings": {"result": {"child": {"value": "7"}}}}
+            """
+            result: RuntimeResultModel
+            assert result.child is not None  # noqa: F821  # pyright: ignore[reportUndefinedVariable, reportUnboundVariable, reportAttributeAccessIssue]
+            return f"{type(result).__name__}:{type(result.child).__name__}:{result.child.value}"  # noqa: F821  # pyright: ignore[reportUndefinedVariable, reportUnboundVariable, reportAttributeAccessIssue]
+
+        assert f() == "RuntimeResultModel:RuntimeChildModel:7"
+
+
+def test_pass_step_finalize_coerces_discriminated_union_list_items() -> None:
+    with nh.run(StubExecutor()):
+
+        @nh.natural_function
+        def f() -> str:
+            """natural
+            <:result>
+            {"step_outcome": {"kind": "pass"}, "bindings": {"result": {"proposals": [{"kind": "approve", "score": "5"}]}}}
+            """
+            result: RuntimeProposalEnvelope
+            first_proposal = result.proposals[0]  # noqa: F821  # pyright: ignore[reportUndefinedVariable, reportUnboundVariable, reportAttributeAccessIssue]
+            return f"{type(first_proposal).__name__}:{first_proposal.score}"
+
+        assert f() == "RuntimeApproveProposal:5"
+
+
+def test_pass_step_finalize_rejects_cross_field_model_violation() -> None:
+    with nh.run(StubExecutor()):
+
+        @nh.natural_function
+        def f() -> int:
+            """natural
+            <:decision>
+            {"step_outcome": {"kind": "pass"}, "bindings": {"decision": {"should_update": true, "next_plan": null}}}
+            """
+            decision: RuntimePlanUpdateDecision
+            return 1 if decision.should_update else 0  # noqa: F821  # pyright: ignore[reportUndefinedVariable, reportUnboundVariable, reportAttributeAccessIssue]
+
+        with pytest.raises(ExecutionError, match="Output binding 'decision' failed validation"):
+            f()
+
+
 def test_async_step_execution_sets_execution_ref_step_id() -> None:
     observed_step_ids: list[str | None] = []
 
@@ -648,6 +735,36 @@ def test_step_system_prompt_injects_tool_result_max_tokens() -> None:
     assert "$tool_result_max_tokens" not in resolved_system_prompt_text
 
 
+def test_agent_executor_commits_write_binding_after_dotted_assignment() -> None:
+    class FakeRunResult:
+        def __init__(self, output: object) -> None:
+            self.output = output
+
+    class FakeAgent:
+        def run_sync(self, user_prompt: str, *, deps=None, **kwargs):  # type: ignore[no-untyped-def]
+            from nighthawk.tools.assignment import assign_tool
+
+            assert deps is not None
+            _ = user_prompt
+            _ = kwargs
+
+            assign_tool(deps, "result.value", "9")
+            return FakeRunResult(StepFinalResult(result=PassStepOutcome(kind="pass")))
+
+    with nh.run(nh.AgentStepExecutor.from_agent(agent=FakeAgent())):
+
+        @nh.natural_function
+        def f() -> int:
+            result: RuntimeResultModel = RuntimeResultModel(value=0)
+            """natural
+            <:result>
+            Update the result value.
+            """
+            return result.value
+
+        assert f() == 9
+
+
 def test_natural_function_can_override_step_executor_configuration_model_within_scope() -> None:
     class FakeRunResult:
         def __init__(self, output: object) -> None:

tests/governance/test_governance.py

@@ -10,6 +10,7 @@
 from opentelemetry.sdk.trace.export import SimpleSpanProcessor
 from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
 from opentelemetry.trace import StatusCode
+from pydantic import BaseModel
 
 import nighthawk as nh
 from nighthawk.errors import ExecutionError, NighthawkError
@@ -19,6 +20,10 @@
 from tests.execution.stub_executor import StubExecutor
 
 
+class GovernanceResultModel(BaseModel):
+    value: int
+
+
 @pytest.fixture
 def step_span_exporter() -> Generator[InMemorySpanExporter, None, None]:
     span_exporter = InMemorySpanExporter()
@@ -169,6 +174,25 @@ def natural_value_function() -> int:
             natural_value_function()
 
 
+def test_step_commit_rewrite_is_coerced_by_finalize_validation() -> None:
+    def rewrite_step(review: nh.oversight.StepCommitProposal) -> nh.oversight.Rewrite:
+        assert review.proposed_binding_name_to_value["result"] == {"value": 1}
+        return nh.oversight.Rewrite(rewritten_binding_name_to_value={"result": {"value": "29"}})
+
+    with nh.run(StubExecutor()), nh.scope(oversight=nh.oversight.Oversight(inspect_step_commit=rewrite_step)):
+
+        @nh.natural_function
+        def natural_value_function() -> int:
+            """natural
+            <:result>
+            {"step_outcome": {"kind": "pass"}, "bindings": {"result": {"value": 1}}}
+            """
+            result: GovernanceResultModel
+            return result.value  # noqa: F821  # pyright: ignore[reportUndefinedVariable, reportUnboundVariable, reportAttributeAccessIssue]
+
+        assert natural_value_function() == 29
+
+
 def test_empty_rewrite_is_rejected() -> None:
     with pytest.raises(ValueError, match="Rewrite must change"):
         nh.oversight.Rewrite()