feat(worker): bound intermediate native results

internal/worker/child.go

@@ -139,10 +139,10 @@ func executionLimits(limits childLimits) execution.Limits {
 // nativeForwarder sends one native_call and waits for the parent reply.
 func nativeForwarder(conn *childConn) execution.NativeCall {
 	return func(id string, arguments map[string]any) (any, error) {
-		if err := validateBoundedValue(arguments, conn.limits); err != nil {
-			return nil, execution.ErrResourceLimit
-		}
 		if err := conn.writeNativeCall(id, arguments); err != nil {
+			if errors.Is(err, errInvalidValue) || errors.Is(err, errFrameTooLarge) {
+				return nil, execution.ErrResourceLimit
+			}
 			return nil, fmt.Errorf("%w: %w", errChildService, err)
 		}
 		frame, err := conn.read()

internal/worker/child_test.go

@@ -8,6 +8,8 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+
+	"github.com/meigma/codemode/internal/execution"
 )
 
 // TestServeProbe proves one probe request is acknowledged and serve returns nil.
@@ -222,6 +224,41 @@ func TestServeFinalWriteClassification(t *testing.T) {
 	assert.False(t, ok)
 }
 
+// TestNativeForwarderClassifiesBoundedWriteErrors proves oversized or nil
+// argument maps fail as a resource limit before a native_call is written.
+func TestNativeForwarderClassifiesBoundedWriteErrors(t *testing.T) {
+	tests := []struct {
+		// name identifies the rejected argument map.
+		name string
+
+		// arguments are presented to the native forwarder.
+		arguments map[string]any
+	}{
+		{name: "nil arguments", arguments: nil},
+		{name: "oversized arguments", arguments: map[string]any{"org": "meigma"}},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			exec := validExecFrame()
+			exec.Limits.MaxValueBytes = 8
+			parent, child := newBufferedExecPair(t, exec)
+			require.NoError(t, parent.writeExec(exec))
+			_, err := child.read()
+			require.NoError(t, err)
+			written, ok := parent.r.(*bytes.Buffer)
+			require.True(t, ok)
+			require.Zero(t, written.Len())
+
+			_, err = nativeForwarder(child)("cap.lookup", tt.arguments)
+
+			require.ErrorIs(t, err, execution.ErrResourceLimit)
+			require.NotErrorIs(t, err, errChildService)
+			assert.Zero(t, written.Len())
+		})
+	}
+}
+
 // TestServeEngineAbortSuppressesFinalError proves a real Engine native_abort
 // exits successfully without a terminal child frame.
 func TestServeEngineAbortSuppressesFinalError(t *testing.T) {
@@ -250,7 +287,7 @@ func startServeExec(t *testing.T, exec execFrame) (*parentConn, <-chan error, fu
 
 	childReader, parentWriter := io.Pipe()
 	parentReader, childWriter := io.Pipe()
-	parent := newParentExecConn(parentWriter, parentReader, parentCap, childCap)
+	parent := newParentExecConn(parentWriter, parentReader, parentCap, childCap, exec.Limits.MaxValueBytes)
 	done := make(chan error, 1)
 	go func() {
 		done <- serve(childReader, childWriter)

internal/worker/frame.go

@@ -207,6 +207,9 @@ type parentConn struct {
 
 	// hasLimits reports whether exec limits have been stored.
 	hasLimits bool
+
+	// remainingIntermediate is the unused native-result value-body budget.
+	remainingIntermediate int
 }
 
 // childConn is the child side of one probe or execution exchange.
@@ -316,16 +319,21 @@ func encodeExec(frame execFrame) ([]byte, error) {
 
 // encodeNativeCall encodes one child-originated native invocation.
 func encodeNativeCall(capabilityID string, arguments map[string]any) ([]byte, error) {
-	if capabilityID == "" {
-		return nil, errInvalidValue
-	}
 	if arguments == nil {
 		return nil, errInvalidValue
 	}
 	encodedArgs, err := encodeNormalizedValue(arguments)
 	if err != nil {
 		return nil, err
 	}
+	return encodeNativeCallBytes(capabilityID, encodedArgs)
+}
+
+// encodeNativeCallBytes wraps one already-encoded argument body in a native_call frame.
+func encodeNativeCallBytes(capabilityID string, encodedArgs []byte) ([]byte, error) {
+	if capabilityID == "" {
+		return nil, errInvalidValue
+	}
 	encodedID, err := json.Marshal(capabilityID)
 	if err != nil {
 		return nil, errInvalidValue
@@ -346,12 +354,17 @@ func encodeNativeResult(result any) ([]byte, error) {
 	if err != nil {
 		return nil, err
 	}
+	return encodeNativeResultBytes(encoded), nil
+}
+
+// encodeNativeResultBytes wraps one already-encoded value body in a native_result frame.
+func encodeNativeResultBytes(encoded []byte) []byte {
 	var buf bytes.Buffer
 	buf.Grow(len(nativeResultPrefix) + len(encoded) + len(nativeResultSuffix))
 	buf.WriteString(nativeResultPrefix)
 	buf.Write(encoded)
 	buf.WriteString(nativeResultSuffix)
-	return buf.Bytes(), nil
+	return buf.Bytes()
 }
 
 // encodeNativeAbort encodes the payload-free parent abort frame.
@@ -365,12 +378,17 @@ func encodeFinal(result any) ([]byte, error) {
 	if err != nil {
 		return nil, err
 	}
+	return encodeFinalBytes(encoded), nil
+}
+
+// encodeFinalBytes wraps one already-encoded value body in a final frame.
+func encodeFinalBytes(encoded []byte) []byte {
 	var buf bytes.Buffer
 	buf.Grow(len(finalPrefix) + len(encoded) + len(finalSuffix))
 	buf.WriteString(finalPrefix)
 	buf.Write(encoded)
 	buf.WriteString(finalSuffix)
-	return buf.Bytes(), nil
+	return buf.Bytes()
 }
 
 // encodeFinalError encodes one child-owned terminal failure.
@@ -459,14 +477,15 @@ func newChildProbeConn(r io.Reader, w io.Writer) *childConn {
 }
 
 // newParentExecConn constructs the parent side of an execution exchange.
-func newParentExecConn(w io.Writer, r io.Reader, writeCap uint32, readCap uint32) *parentConn {
+func newParentExecConn(w io.Writer, r io.Reader, writeCap uint32, readCap uint32, intermediateBudget int) *parentConn {
 	return &parentConn{
-		w:        w,
-		r:        r,
-		writeCap: writeCap,
-		readCap:  readCap,
-		kind:     connKindExec,
-		state:    stateInit,
+		w:                     w,
+		r:                     r,
+		writeCap:              writeCap,
+		readCap:               readCap,
+		kind:                  connKindExec,
+		state:                 stateInit,
+		remainingIntermediate: intermediateBudget,
 	}
 }
 
@@ -527,16 +546,18 @@ func (c *parentConn) writeNativeResult(result any) error {
 	if c == nil || c.kind != connKindExec || c.state != stateAwaitNative || !c.hasLimits {
 		return errIllegalState
 	}
-	if err := validateBoundedValue(result, c.limits); err != nil {
-		return err
-	}
-	payload, err := encodeNativeResult(result)
+	encoded, err := encodeBoundedValue(result, c.limits)
 	if err != nil {
 		return err
 	}
+	if len(encoded) > c.remainingIntermediate {
+		return errFrameTooLarge
+	}
+	payload := encodeNativeResultBytes(encoded)
 	if err := writeFrame(c.w, payload, c.writeCap); err != nil {
 		return err
 	}
+	c.remainingIntermediate -= len(encoded)
 	c.state = stateReady
 	return nil
 }
@@ -654,10 +675,14 @@ func (c *childConn) writeNativeCall(capabilityID string, arguments map[string]an
 	if c.nativeCalls >= c.limits.MaxNativeCalls {
 		return errIllegalState
 	}
-	if err := validateBoundedValue(arguments, c.limits); err != nil {
+	if arguments == nil {
+		return errInvalidValue
+	}
+	encodedArgs, err := encodeBoundedValue(arguments, c.limits)
+	if err != nil {
 		return err
 	}
-	payload, err := encodeNativeCall(capabilityID, arguments)
+	payload, err := encodeNativeCallBytes(capabilityID, encodedArgs)
 	if err != nil {
 		return err
 	}
@@ -674,13 +699,11 @@ func (c *childConn) writeFinal(result any) error {
 	if c == nil || c.kind != connKindExec || c.state != stateReady || !c.hasLimits {
 		return errIllegalState
 	}
-	if err := validateBoundedValue(result, c.limits); err != nil {
-		return err
-	}
-	payload, err := encodeFinal(result)
+	encoded, err := encodeBoundedValue(result, c.limits)
 	if err != nil {
 		return err
 	}
+	payload := encodeFinalBytes(encoded)
 	if err := writeFrame(c.w, payload, c.writeCap); err != nil {
 		return err
 	}

internal/worker/frame_test.go

@@ -696,6 +696,53 @@ func TestFrameLimitsCheckedCaps(t *testing.T) {
 	assert.Greater(t, expanded, unexpanded)
 }
 
+// TestFrameNativeResultChargesValueBodyOnly proves the envelope is excluded from accounting.
+func TestFrameNativeResultChargesValueBodyOnly(t *testing.T) {
+	exec := validExecFrame()
+	result := map[string]any{"zeta": int64(1), "alpha": true}
+	body, err := encodeNormalizedValue(result)
+	require.NoError(t, err)
+	frame, err := encodeNativeResult(result)
+	require.NoError(t, err)
+	assert.Equal(t, encodeNativeResultBytes(body), frame)
+	assert.Len(t, frame, len(nativeResultPrefix)+len(body)+len(nativeResultSuffix))
+	assert.Greater(t, len(frame), len(body))
+
+	parent, child := newBufferedExecPair(t, exec)
+	parent.remainingIntermediate = len(body)
+	require.NoError(t, parent.writeExec(exec))
+	_, err = child.read()
+	require.NoError(t, err)
+	require.NoError(t, child.writeNativeCall("cap.lookup", map[string]any{"org": "meigma"}))
+	_, err = parent.read()
+	require.NoError(t, err)
+	require.NoError(t, parent.writeNativeResult(result))
+	assert.Zero(t, parent.remainingIntermediate)
+	reply, err := child.read()
+	require.NoError(t, err)
+	got, ok := reply.(nativeResultFrame)
+	require.True(t, ok)
+	assert.Equal(t, result, got.Result)
+	require.NoError(t, child.writeNativeCall("cap.lookup", map[string]any{"org": "meigma"}))
+	_, err = parent.read()
+	require.NoError(t, err)
+	err = parent.writeNativeResult(result)
+	require.ErrorIs(t, err, errFrameTooLarge)
+}
+
+// TestWriteNativeCallRejectsNilArguments proves a nil map never encodes as JSON null.
+func TestWriteNativeCallRejectsNilArguments(t *testing.T) {
+	exec := validExecFrame()
+	parent, child := newBufferedExecPair(t, exec)
+	require.NoError(t, parent.writeExec(exec))
+	_, err := child.read()
+	require.NoError(t, err)
+
+	err = child.writeNativeCall("cap.lookup", nil)
+
+	require.ErrorIs(t, err, errInvalidValue)
+}
+
 // TestFrameLimitsRejectUint32Overflow proves checked addition never wraps the prefix.
 func TestFrameLimitsRejectUint32Overflow(t *testing.T) {
 	manifest := validManifest()
@@ -805,7 +852,7 @@ func newExecPair(t *testing.T, exec execFrame) (*parentConn, *childConn, func())
 		_ = childWriter.Close()
 	}
 	t.Cleanup(closePipes)
-	return newParentExecConn(parentWriter, parentReader, parentCap, childCap),
+	return newParentExecConn(parentWriter, parentReader, parentCap, childCap, exec.Limits.MaxValueBytes),
 		newChildExecConn(childReader, childWriter, execCap, parentCap, childCap),
 		closePipes
 }
@@ -832,7 +879,7 @@ func newBufferedExecPair(t *testing.T, exec execFrame) (*parentConn, *childConn)
 
 	var toChild bytes.Buffer
 	var toParent bytes.Buffer
-	return newParentExecConn(&toChild, &toParent, parentCap, childCap),
+	return newParentExecConn(&toChild, &toParent, parentCap, childCap, exec.Limits.MaxValueBytes),
 		newChildExecConn(&toChild, &toParent, execCap, parentCap, childCap)
 }
 

internal/worker/parent.go

@@ -103,6 +103,11 @@ type Limits struct {
 	// MaxValueBytes bounds type-preserving encoded value size.
 	MaxValueBytes int
 
+	// MaxIntermediateValueBytes bounds cumulative successful native-result
+	// value-body bytes for one Execute call. It is independent of MaxValueBytes
+	// and is never projected to the child.
+	MaxIntermediateValueBytes int
+
 	// MaxConcurrentExecutions bounds live worker children.
 	MaxConcurrentExecutions int
 }
@@ -200,7 +205,7 @@ func NewRunner(bindings []execution.CapabilityBinding, limits Limits, dispatch D
 	if dispatch == nil {
 		return nil, errors.New("nil dispatch")
 	}
-	if limits.MaxExecutionTime <= 0 || limits.MaxConcurrentExecutions <= 0 {
+	if limits.MaxExecutionTime <= 0 || limits.MaxConcurrentExecutions <= 0 || limits.MaxIntermediateValueBytes <= 0 {
 		return nil, errInvalidLimits
 	}
 	child := childLimits{
@@ -669,7 +674,7 @@ func (r *Runner) runExecExchange(
 	subject authz.Subject,
 	source string,
 ) execOutcome {
-	conn := newParentExecConn(stdin, stdout, r.writeCap, r.readCap)
+	conn := newParentExecConn(stdin, stdout, r.writeCap, r.readCap, r.limits.MaxIntermediateValueBytes)
 	frame := execFrame{
 		Type:     frameTypeExec,
 		Version:  protocolVersion,