diff --git a/v23/services/syncbase/nosql/service.vdl b/v23/services/syncbase/nosql/service.vdl
index ebeccf5..16182de 100644
--- a/v23/services/syncbase/nosql/service.vdl
+++ b/v23/services/syncbase/nosql/service.vdl
@@ -73,6 +73,10 @@
 	// SchemaManager implements the API for managing schema metadata attached
 	// to a Database.
 	SchemaManager
+
+	// ConflictManager implements the API for registering resolvers, receiving
+	// conflicts and sending resolutions.
+	ConflictManager
 }
 
 // Table represents a collection of Rows.
@@ -234,6 +238,30 @@
 	SetSchemaMetadata(metadata SchemaMetadata) error {access.Write}
 }
 
+// ConflictManager interface provides all the methods necessary to handle
+// conflict resolution for a given database.
+type ConflictManager interface {
+	// StartConflictResolver registers a resolver for the database that is
+	// associated with this ConflictManager and creates a stream to receive
+	// conflicts and send resolutions.
+	// Batches of ConflictInfos will be sent over with the Continued field
+	// within the ConflictInfo representing the batch boundary. Client must
+	// respond with a batch of ResolutionInfos in the same fashion.
+	// A key is under conflict if two different values were written to it
+	// concurrently (in logical time), i.e. neither value is an ancestor of the
+	// other in the history graph.
+	// A key under conflict can be a part of a batch committed on local or
+	// remote or both syncbases. ConflictInfos for all keys in these two batches
+	// are grouped together. These keys may themselves be under conflict; the
+	// presented batch is a transitive closure of all batches containing keys
+	// under conflict.
+	// For example, for local batch {key1, key2} and remote batch {key1, key3},
+	// the batch sent for conflict resolution will be {key1, key2, key3}.
+	// If there was another concurrent batch {key2, key4}, then the batch sent
+	// for conflict resolution will be {key1, key2, key3, key4}.
+	StartConflictResolver() stream<ResolutionInfo, ConflictInfo> error {access.Write}
+}
+
 // BlobManager is the interface for blob operations.
 type BlobManager interface {
 	// API for resumable blob creation (append-only). After commit, a blob
diff --git a/v23/services/syncbase/nosql/service.vdl.go b/v23/services/syncbase/nosql/service.vdl.go
index 2d3d4ba..a908cfa 100644
--- a/v23/services/syncbase/nosql/service.vdl.go
+++ b/v23/services/syncbase/nosql/service.vdl.go
@@ -1452,6 +1452,356 @@
 	},
 }
 
+// ConflictManagerClientMethods is the client interface
+// containing ConflictManager methods.
+//
+// ConflictManager interface provides all the methods necessary to handle
+// conflict resolution for a given database.
+type ConflictManagerClientMethods interface {
+	// StartConflictResolver registers a resolver for the database that is
+	// associated with this ConflictManager and creates a stream to receive
+	// conflicts and send resolutions.
+	// Batches of ConflictInfos will be sent over with the Continued field
+	// within the ConflictInfo representing the batch boundary. Client must
+	// respond with a batch of ResolutionInfos in the same fashion.
+	// A key is under conflict if two different values were written to it
+	// concurrently (in logical time), i.e. neither value is an ancestor of the
+	// other in the history graph.
+	// A key under conflict can be a part of a batch committed on local or
+	// remote or both syncbases. ConflictInfos for all keys in these two batches
+	// are grouped together. These keys may themselves be under conflict; the
+	// presented batch is a transitive closure of all batches containing keys
+	// under conflict.
+	// For example, for local batch {key1, key2} and remote batch {key1, key3},
+	// the batch sent for conflict resolution will be {key1, key2, key3}.
+	// If there was another concurrent batch {key2, key4}, then the batch sent
+	// for conflict resolution will be {key1, key2, key3, key4}.
+	StartConflictResolver(*context.T, ...rpc.CallOpt) (ConflictManagerStartConflictResolverClientCall, error)
+}
+
+// ConflictManagerClientStub adds universal methods to ConflictManagerClientMethods.
+type ConflictManagerClientStub interface {
+	ConflictManagerClientMethods
+	rpc.UniversalServiceMethods
+}
+
+// ConflictManagerClient returns a client stub for ConflictManager.
+func ConflictManagerClient(name string) ConflictManagerClientStub {
+	return implConflictManagerClientStub{name}
+}
+
+type implConflictManagerClientStub struct {
+	name string
+}
+
+func (c implConflictManagerClientStub) StartConflictResolver(ctx *context.T, opts ...rpc.CallOpt) (ocall ConflictManagerStartConflictResolverClientCall, err error) {
+	var call rpc.ClientCall
+	if call, err = v23.GetClient(ctx).StartCall(ctx, c.name, "StartConflictResolver", nil, opts...); err != nil {
+		return
+	}
+	ocall = &implConflictManagerStartConflictResolverClientCall{ClientCall: call}
+	return
+}
+
+// ConflictManagerStartConflictResolverClientStream is the client stream for ConflictManager.StartConflictResolver.
+type ConflictManagerStartConflictResolverClientStream interface {
+	// RecvStream returns the receiver side of the ConflictManager.StartConflictResolver client stream.
+	RecvStream() interface {
+		// Advance stages an item so that it may be retrieved via Value.  Returns
+		// true iff there is an item to retrieve.  Advance must be called before
+		// Value is called.  May block if an item is not available.
+		Advance() bool
+		// Value returns the item that was staged by Advance.  May panic if Advance
+		// returned false or was not called.  Never blocks.
+		Value() ConflictInfo
+		// Err returns any error encountered by Advance.  Never blocks.
+		Err() error
+	}
+	// SendStream returns the send side of the ConflictManager.StartConflictResolver client stream.
+	SendStream() interface {
+		// Send places the item onto the output stream.  Returns errors
+		// encountered while sending, or if Send is called after Close or
+		// the stream has been canceled.  Blocks if there is no buffer
+		// space; will unblock when buffer space is available or after
+		// the stream has been canceled.
+		Send(item ResolutionInfo) error
+		// Close indicates to the server that no more items will be sent;
+		// server Recv calls will receive io.EOF after all sent items.
+		// This is an optional call - e.g. a client might call Close if it
+		// needs to continue receiving items from the server after it's
+		// done sending.  Returns errors encountered while closing, or if
+		// Close is called after the stream has been canceled.  Like Send,
+		// blocks if there is no buffer space available.
+		Close() error
+	}
+}
+
+// ConflictManagerStartConflictResolverClientCall represents the call returned from ConflictManager.StartConflictResolver.
+type ConflictManagerStartConflictResolverClientCall interface {
+	ConflictManagerStartConflictResolverClientStream
+	// Finish performs the equivalent of SendStream().Close, then blocks until
+	// the server is done, and returns the positional return values for the call.
+	//
+	// Finish returns immediately if the call has been canceled; depending on the
+	// timing the output could either be an error signaling cancelation, or the
+	// valid positional return values from the server.
+	//
+	// Calling Finish is mandatory for releasing stream resources, unless the call
+	// has been canceled or any of the other methods return an error.  Finish should
+	// be called at most once.
+	Finish() error
+}
+
+type implConflictManagerStartConflictResolverClientCall struct {
+	rpc.ClientCall
+	valRecv ConflictInfo
+	errRecv error
+}
+
+func (c *implConflictManagerStartConflictResolverClientCall) RecvStream() interface {
+	Advance() bool
+	Value() ConflictInfo
+	Err() error
+} {
+	return implConflictManagerStartConflictResolverClientCallRecv{c}
+}
+
+type implConflictManagerStartConflictResolverClientCallRecv struct {
+	c *implConflictManagerStartConflictResolverClientCall
+}
+
+func (c implConflictManagerStartConflictResolverClientCallRecv) Advance() bool {
+	c.c.valRecv = ConflictInfo{}
+	c.c.errRecv = c.c.Recv(&c.c.valRecv)
+	return c.c.errRecv == nil
+}
+func (c implConflictManagerStartConflictResolverClientCallRecv) Value() ConflictInfo {
+	return c.c.valRecv
+}
+func (c implConflictManagerStartConflictResolverClientCallRecv) Err() error {
+	if c.c.errRecv == io.EOF {
+		return nil
+	}
+	return c.c.errRecv
+}
+func (c *implConflictManagerStartConflictResolverClientCall) SendStream() interface {
+	Send(item ResolutionInfo) error
+	Close() error
+} {
+	return implConflictManagerStartConflictResolverClientCallSend{c}
+}
+
+type implConflictManagerStartConflictResolverClientCallSend struct {
+	c *implConflictManagerStartConflictResolverClientCall
+}
+
+func (c implConflictManagerStartConflictResolverClientCallSend) Send(item ResolutionInfo) error {
+	return c.c.Send(item)
+}
+func (c implConflictManagerStartConflictResolverClientCallSend) Close() error {
+	return c.c.CloseSend()
+}
+func (c *implConflictManagerStartConflictResolverClientCall) Finish() (err error) {
+	err = c.ClientCall.Finish()
+	return
+}
+
+// ConflictManagerServerMethods is the interface a server writer
+// implements for ConflictManager.
+//
+// ConflictManager interface provides all the methods necessary to handle
+// conflict resolution for a given database.
+type ConflictManagerServerMethods interface {
+	// StartConflictResolver registers a resolver for the database that is
+	// associated with this ConflictManager and creates a stream to receive
+	// conflicts and send resolutions.
+	// Batches of ConflictInfos will be sent over with the Continued field
+	// within the ConflictInfo representing the batch boundary. Client must
+	// respond with a batch of ResolutionInfos in the same fashion.
+	// A key is under conflict if two different values were written to it
+	// concurrently (in logical time), i.e. neither value is an ancestor of the
+	// other in the history graph.
+	// A key under conflict can be a part of a batch committed on local or
+	// remote or both syncbases. ConflictInfos for all keys in these two batches
+	// are grouped together. These keys may themselves be under conflict; the
+	// presented batch is a transitive closure of all batches containing keys
+	// under conflict.
+	// For example, for local batch {key1, key2} and remote batch {key1, key3},
+	// the batch sent for conflict resolution will be {key1, key2, key3}.
+	// If there was another concurrent batch {key2, key4}, then the batch sent
+	// for conflict resolution will be {key1, key2, key3, key4}.
+	StartConflictResolver(*context.T, ConflictManagerStartConflictResolverServerCall) error
+}
+
+// ConflictManagerServerStubMethods is the server interface containing
+// ConflictManager methods, as expected by rpc.Server.
+// The only difference between this interface and ConflictManagerServerMethods
+// is the streaming methods.
+type ConflictManagerServerStubMethods interface {
+	// StartConflictResolver registers a resolver for the database that is
+	// associated with this ConflictManager and creates a stream to receive
+	// conflicts and send resolutions.
+	// Batches of ConflictInfos will be sent over with the Continued field
+	// within the ConflictInfo representing the batch boundary. Client must
+	// respond with a batch of ResolutionInfos in the same fashion.
+	// A key is under conflict if two different values were written to it
+	// concurrently (in logical time), i.e. neither value is an ancestor of the
+	// other in the history graph.
+	// A key under conflict can be a part of a batch committed on local or
+	// remote or both syncbases. ConflictInfos for all keys in these two batches
+	// are grouped together. These keys may themselves be under conflict; the
+	// presented batch is a transitive closure of all batches containing keys
+	// under conflict.
+	// For example, for local batch {key1, key2} and remote batch {key1, key3},
+	// the batch sent for conflict resolution will be {key1, key2, key3}.
+	// If there was another concurrent batch {key2, key4}, then the batch sent
+	// for conflict resolution will be {key1, key2, key3, key4}.
+	StartConflictResolver(*context.T, *ConflictManagerStartConflictResolverServerCallStub) error
+}
+
+// ConflictManagerServerStub adds universal methods to ConflictManagerServerStubMethods.
+type ConflictManagerServerStub interface {
+	ConflictManagerServerStubMethods
+	// Describe the ConflictManager interfaces.
+	Describe__() []rpc.InterfaceDesc
+}
+
+// ConflictManagerServer returns a server stub for ConflictManager.
+// It converts an implementation of ConflictManagerServerMethods into
+// an object that may be used by rpc.Server.
+func ConflictManagerServer(impl ConflictManagerServerMethods) ConflictManagerServerStub {
+	stub := implConflictManagerServerStub{
+		impl: impl,
+	}
+	// Initialize GlobState; always check the stub itself first, to handle the
+	// case where the user has the Glob method defined in their VDL source.
+	if gs := rpc.NewGlobState(stub); gs != nil {
+		stub.gs = gs
+	} else if gs := rpc.NewGlobState(impl); gs != nil {
+		stub.gs = gs
+	}
+	return stub
+}
+
+type implConflictManagerServerStub struct {
+	impl ConflictManagerServerMethods
+	gs   *rpc.GlobState
+}
+
+func (s implConflictManagerServerStub) StartConflictResolver(ctx *context.T, call *ConflictManagerStartConflictResolverServerCallStub) error {
+	return s.impl.StartConflictResolver(ctx, call)
+}
+
+func (s implConflictManagerServerStub) Globber() *rpc.GlobState {
+	return s.gs
+}
+
+func (s implConflictManagerServerStub) Describe__() []rpc.InterfaceDesc {
+	return []rpc.InterfaceDesc{ConflictManagerDesc}
+}
+
+// ConflictManagerDesc describes the ConflictManager interface.
+var ConflictManagerDesc rpc.InterfaceDesc = descConflictManager
+
+// descConflictManager hides the desc to keep godoc clean.
+var descConflictManager = rpc.InterfaceDesc{
+	Name:    "ConflictManager",
+	PkgPath: "v.io/syncbase/v23/services/syncbase/nosql",
+	Doc:     "// ConflictManager interface provides all the methods necessary to handle\n// conflict resolution for a given database.",
+	Methods: []rpc.MethodDesc{
+		{
+			Name: "StartConflictResolver",
+			Doc:  "// StartConflictResolver registers a resolver for the database that is\n// associated with this ConflictManager and creates a stream to receive\n// conflicts and send resolutions.\n// Batches of ConflictInfos will be sent over with the Continued field\n// within the ConflictInfo representing the batch boundary. Client must\n// respond with a batch of ResolutionInfos in the same fashion.\n// A key is under conflict if two different values were written to it\n// concurrently (in logical time), i.e. neither value is an ancestor of the\n// other in the history graph.\n// A key under conflict can be a part of a batch committed on local or\n// remote or both syncbases. ConflictInfos for all keys in these two batches\n// are grouped together. These keys may themselves be under conflict; the\n// presented batch is a transitive closure of all batches containing keys\n// under conflict.\n// For example, for local batch {key1, key2} and remote batch {key1, key3},\n// the batch sent for conflict resolution will be {key1, key2, key3}.\n// If there was another concurrent batch {key2, key4}, then the batch sent\n// for conflict resolution will be {key1, key2, key3, key4}.",
+			Tags: []*vdl.Value{vdl.ValueOf(access.Tag("Write"))},
+		},
+	},
+}
+
+// ConflictManagerStartConflictResolverServerStream is the server stream for ConflictManager.StartConflictResolver.
+type ConflictManagerStartConflictResolverServerStream interface {
+	// RecvStream returns the receiver side of the ConflictManager.StartConflictResolver server stream.
+	RecvStream() interface {
+		// Advance stages an item so that it may be retrieved via Value.  Returns
+		// true iff there is an item to retrieve.  Advance must be called before
+		// Value is called.  May block if an item is not available.
+		Advance() bool
+		// Value returns the item that was staged by Advance.  May panic if Advance
+		// returned false or was not called.  Never blocks.
+		Value() ResolutionInfo
+		// Err returns any error encountered by Advance.  Never blocks.
+		Err() error
+	}
+	// SendStream returns the send side of the ConflictManager.StartConflictResolver server stream.
+	SendStream() interface {
+		// Send places the item onto the output stream.  Returns errors encountered
+		// while sending.  Blocks if there is no buffer space; will unblock when
+		// buffer space is available.
+		Send(item ConflictInfo) error
+	}
+}
+
+// ConflictManagerStartConflictResolverServerCall represents the context passed to ConflictManager.StartConflictResolver.
+type ConflictManagerStartConflictResolverServerCall interface {
+	rpc.ServerCall
+	ConflictManagerStartConflictResolverServerStream
+}
+
+// ConflictManagerStartConflictResolverServerCallStub is a wrapper that converts rpc.StreamServerCall into
+// a typesafe stub that implements ConflictManagerStartConflictResolverServerCall.
+type ConflictManagerStartConflictResolverServerCallStub struct {
+	rpc.StreamServerCall
+	valRecv ResolutionInfo
+	errRecv error
+}
+
+// Init initializes ConflictManagerStartConflictResolverServerCallStub from rpc.StreamServerCall.
+func (s *ConflictManagerStartConflictResolverServerCallStub) Init(call rpc.StreamServerCall) {
+	s.StreamServerCall = call
+}
+
+// RecvStream returns the receiver side of the ConflictManager.StartConflictResolver server stream.
+func (s *ConflictManagerStartConflictResolverServerCallStub) RecvStream() interface {
+	Advance() bool
+	Value() ResolutionInfo
+	Err() error
+} {
+	return implConflictManagerStartConflictResolverServerCallRecv{s}
+}
+
+type implConflictManagerStartConflictResolverServerCallRecv struct {
+	s *ConflictManagerStartConflictResolverServerCallStub
+}
+
+func (s implConflictManagerStartConflictResolverServerCallRecv) Advance() bool {
+	s.s.valRecv = ResolutionInfo{}
+	s.s.errRecv = s.s.Recv(&s.s.valRecv)
+	return s.s.errRecv == nil
+}
+func (s implConflictManagerStartConflictResolverServerCallRecv) Value() ResolutionInfo {
+	return s.s.valRecv
+}
+func (s implConflictManagerStartConflictResolverServerCallRecv) Err() error {
+	if s.s.errRecv == io.EOF {
+		return nil
+	}
+	return s.s.errRecv
+}
+
+// SendStream returns the send side of the ConflictManager.StartConflictResolver server stream.
+func (s *ConflictManagerStartConflictResolverServerCallStub) SendStream() interface {
+	Send(item ConflictInfo) error
+} {
+	return implConflictManagerStartConflictResolverServerCallSend{s}
+}
+
+type implConflictManagerStartConflictResolverServerCallSend struct {
+	s *ConflictManagerStartConflictResolverServerCallStub
+}
+
+func (s implConflictManagerStartConflictResolverServerCallSend) Send(item ConflictInfo) error {
+	return s.s.Send(item)
+}
+
 // DatabaseClientMethods is the client interface
 // containing Database methods.
 //
@@ -1539,6 +1889,9 @@
 	// SchemaManager implements the API for managing schema metadata attached
 	// to a Database.
 	SchemaManagerClientMethods
+	// ConflictManager interface provides all the methods necessary to handle
+	// conflict resolution for a given database.
+	ConflictManagerClientMethods
 	// Create creates this Database.
 	// If perms is nil, we inherit (copy) the App perms.
 	// Create requires the caller to have Write permission at the App.
@@ -1579,7 +1932,7 @@
 
 // DatabaseClient returns a client stub for Database.
 func DatabaseClient(name string) DatabaseClientStub {
-	return implDatabaseClientStub{name, permissions.ObjectClient(name), DatabaseWatcherClient(name), SyncGroupManagerClient(name), BlobManagerClient(name), SchemaManagerClient(name)}
+	return implDatabaseClientStub{name, permissions.ObjectClient(name), DatabaseWatcherClient(name), SyncGroupManagerClient(name), BlobManagerClient(name), SchemaManagerClient(name), ConflictManagerClient(name)}
 }
 
 type implDatabaseClientStub struct {
@@ -1590,6 +1943,7 @@
 	SyncGroupManagerClientStub
 	BlobManagerClientStub
 	SchemaManagerClientStub
+	ConflictManagerClientStub
 }
 
 func (c implDatabaseClientStub) Create(ctx *context.T, i0 *SchemaMetadata, i1 access.Permissions, opts ...rpc.CallOpt) (err error) {
@@ -1786,6 +2140,9 @@
 	// SchemaManager implements the API for managing schema metadata attached
 	// to a Database.
 	SchemaManagerServerMethods
+	// ConflictManager interface provides all the methods necessary to handle
+	// conflict resolution for a given database.
+	ConflictManagerServerMethods
 	// Create creates this Database.
 	// If perms is nil, we inherit (copy) the App perms.
 	// Create requires the caller to have Write permission at the App.
@@ -1899,6 +2256,9 @@
 	// SchemaManager implements the API for managing schema metadata attached
 	// to a Database.
 	SchemaManagerServerStubMethods
+	// ConflictManager interface provides all the methods necessary to handle
+	// conflict resolution for a given database.
+	ConflictManagerServerStubMethods
 	// Create creates this Database.
 	// If perms is nil, we inherit (copy) the App perms.
 	// Create requires the caller to have Write permission at the App.
@@ -1949,6 +2309,7 @@
 		SyncGroupManagerServerStub: SyncGroupManagerServer(impl),
 		BlobManagerServerStub:      BlobManagerServer(impl),
 		SchemaManagerServerStub:    SchemaManagerServer(impl),
+		ConflictManagerServerStub:  ConflictManagerServer(impl),
 	}
 	// Initialize GlobState; always check the stub itself first, to handle the
 	// case where the user has the Glob method defined in their VDL source.
@@ -1967,6 +2328,7 @@
 	SyncGroupManagerServerStub
 	BlobManagerServerStub
 	SchemaManagerServerStub
+	ConflictManagerServerStub
 	gs *rpc.GlobState
 }
 
@@ -2003,7 +2365,7 @@
 }
 
 func (s implDatabaseServerStub) Describe__() []rpc.InterfaceDesc {
-	return []rpc.InterfaceDesc{DatabaseDesc, permissions.ObjectDesc, DatabaseWatcherDesc, watch.GlobWatcherDesc, SyncGroupManagerDesc, BlobManagerDesc, SchemaManagerDesc}
+	return []rpc.InterfaceDesc{DatabaseDesc, permissions.ObjectDesc, DatabaseWatcherDesc, watch.GlobWatcherDesc, SyncGroupManagerDesc, BlobManagerDesc, SchemaManagerDesc, ConflictManagerDesc}
 }
 
 // DatabaseDesc describes the Database interface.
@@ -2020,6 +2382,7 @@
 		{"SyncGroupManager", "v.io/syncbase/v23/services/syncbase/nosql", "// SyncGroupManager is the interface for SyncGroup operations.\n// TODO(hpucha): Add blessings to create/join and add a refresh method."},
 		{"BlobManager", "v.io/syncbase/v23/services/syncbase/nosql", "// BlobManager is the interface for blob operations."},
 		{"SchemaManager", "v.io/syncbase/v23/services/syncbase/nosql", "// SchemaManager implements the API for managing schema metadata attached\n// to a Database."},
+		{"ConflictManager", "v.io/syncbase/v23/services/syncbase/nosql", "// ConflictManager interface provides all the methods necessary to handle\n// conflict resolution for a given database."},
 	},
 	Methods: []rpc.MethodDesc{
 		{
diff --git a/v23/services/syncbase/nosql/types.vdl b/v23/services/syncbase/nosql/types.vdl
index eae8e9d..407e956 100644
--- a/v23/services/syncbase/nosql/types.vdl
+++ b/v23/services/syncbase/nosql/types.vdl
@@ -111,6 +111,158 @@
 	Defer
 }
 
+// ConflictInfo contains information to fully specify a conflict
+// for a key, providing the (local, remote, ancestor) tuple.
+// A key under conflict can be a part of a batch in local, remote or both
+// updates. Since the batches can have more than one key, all ConflictInfos
+// for the keys within the batches are grouped together into a single conflict
+// batch and sent as a stream with the Continued field representing conflict
+// batch boundaries.
+type ConflictInfo struct {
+	// Data is a unit chunk of ConflictInfo which can be sent over the conflict
+	// stream.
+	Data ConflictData
+
+	// Continued represents whether the batch of ConflictInfos has ended.
+	Continued bool
+}
+
+// ConflictData represents a unit of conflict data sent over the stream. It
+// can either contain information about a Batch or about an operation done
+// on a row.
+type ConflictData union {
+	Batch BatchInfo
+	Row RowInfo
+}
+
+type BatchInfo struct {
+	// Id is an identifier for a batch contained in a conflict. It is
+	// unique only in the context of a given conflict. Its purpose is solely to
+	// group one or more RowInfo objects together to represent a batch that
+	// was committed by the client.
+	Id uint16
+
+	// Hint is the hint provided by the client when this batch was committed.
+	Hint string
+
+	// Source states where the batch comes from.
+	Source BatchSource
+}
+
+// BatchSource represents where the batch was committed.
+type BatchSource enum {
+	// Local represents that the batch was committed on local syncbase.
+	Local
+
+	// Remote represents that the batch was committed on remote syncbase.
+	Remote
+}
+
+// RowInfo contains a single operation performed on a row (in case of read or
+// write) or a range or rows (in case of scan) along with a mapping to each
+// of the batches that this operation belongs to.
+// For example, if Row1 was updated on local syncbase conflicting with a write
+// on remote syncbase as part of two separate batches, then it will be
+// represented by a single RowInfo with Write Operation containing the
+// respective local and remote values along with the batch id for both batches
+// stored in the BatchIds field.
+type RowInfo struct {
+	// Op is a specific operation represented by RowInfo
+	Op Operation
+	// BatchIds contains ids of all batches that this RowInfo is a part of.
+	BatchIds []uint16
+}
+
+// Operation represents a specific operation on a row or a set of rows that is
+// a part of the conflict.
+type Operation union {
+	// Read represents a read operation performed on a specific row. For a given
+	// row key there can only be at max one Read operation within a conflict.
+	Read  RowOp
+
+	// Write represents a write operation performed on a specific row. For a
+	// given row key there can only be at max one Write operation within a
+	// conflict.
+	Write RowOp
+
+	// Scan represents a scan operation performed over a specific range of keys.
+	// For a given key range there can be at max one ScanOp within the Conflict.
+	Scan  ScanOp
+}
+
+// RowOp represents a read or write operation on a row corresponding to the
+// given key.
+type RowOp struct {
+	// The key under conflict.
+	Key string
+
+	// LocalValue contains the value read or written by local syncbase or nil.
+	LocalValue ?Value
+
+	// RemoteValue contains the value read or written by remote syncbase or nil.
+	RemoteValue ?Value
+
+	// AncestorValue contains the value for the key which is the lowest common
+	// ancestor of the two values represented by LocalValue and RemoteValue or
+	// nil if no ancestor exists or if the operation was read.
+	AncestorValue ?Value
+}
+
+// ScanOp provides details of a scan operation.
+type ScanOp struct {
+	// Start contains the starting key for a range scan.
+	Start string
+
+	// Limit contains the end key for a range scan.
+	Limit string
+}
+
+// Value contains the encoded bytes for a row's value stored in syncbase.
+type Value struct {
+	// VOM encoded bytes for a row's value
+	Bytes []byte
+
+	// Write timestamp for this value in nanoseconds.
+	// TODO(jlodhia): change the timestamp to vdl.time here, in commit timestamp
+	// and clock data.
+	WriteTs int64
+}
+
+// ValueSelection represents the value that was selected as the final resolution
+// for a conflict.
+type ValueSelection enum {
+	// Local should be used if local value is picked as the final result.
+	Local
+
+	// Remote should be used if remote value is picked as the final result.
+	Remote
+
+	// Other should be used if a new value different from local and remote is
+	// used as the final result.
+	Other
+}
+
+// ResolutionInfo contains the application’s reply to a conflict for a key,
+// providing the resolution value. The resolution may be over a group of keys
+// in which case the application must send a stream of ResolutionInfos with
+// the Continued field for the last ResolutionInfo representing the end of the
+// batch with a value false. ResolutionInfos sent as part of a batch will be
+// committed as a batch. If the commit fails, the Conflict will be re-sent.
+type ResolutionInfo struct {
+	// Key is the key under conflict.
+	Key	string
+
+	// Selection represents the value that was selected as resolution.
+	Selection ValueSelection
+
+	// Result is the resolved value for the key. This field should be used only
+	// if value of Selection field is 'Other'
+	Result ?Value
+
+	// Continued represents whether the batch of ResolutionInfos has ended.
+	Continued bool
+}
+
 // SchemaMetadata maintains metadata related to the schema of a given database.
 // There is one SchemaMetadata per database.
 type SchemaMetadata struct {
diff --git a/v23/services/syncbase/nosql/types.vdl.go b/v23/services/syncbase/nosql/types.vdl.go
index adb4e66..6a8e0c1 100644
--- a/v23/services/syncbase/nosql/types.vdl.go
+++ b/v23/services/syncbase/nosql/types.vdl.go
@@ -166,6 +166,336 @@
 }) {
 }
 
+// ConflictInfo contains information to fully specify a conflict
+// for a key, providing the (local, remote, ancestor) tuple.
+// A key under conflict can be a part of a batch in local, remote or both
+// updates. Since the batches can have more than one key, all ConflictInfos
+// for the keys within the batches are grouped together into a single conflict
+// batch and sent as a stream with the Continued field representing conflict
+// batch boundaries.
+type ConflictInfo struct {
+	// Data is a unit chunk of ConflictInfo which can be sent over the conflict
+	// stream.
+	Data ConflictData
+	// Continued represents whether the batch of ConflictInfos has ended.
+	Continued bool
+}
+
+func (ConflictInfo) __VDLReflect(struct {
+	Name string `vdl:"v.io/syncbase/v23/services/syncbase/nosql.ConflictInfo"`
+}) {
+}
+
+type (
+	// ConflictData represents any single field of the ConflictData union type.
+	//
+	// ConflictData represents a unit of conflict data sent over the stream. It
+	// can either contain information about a Batch or about an operation done
+	// on a row.
+	ConflictData interface {
+		// Index returns the field index.
+		Index() int
+		// Interface returns the field value as an interface.
+		Interface() interface{}
+		// Name returns the field name.
+		Name() string
+		// __VDLReflect describes the ConflictData union type.
+		__VDLReflect(__ConflictDataReflect)
+	}
+	// ConflictDataBatch represents field Batch of the ConflictData union type.
+	ConflictDataBatch struct{ Value BatchInfo }
+	// ConflictDataRow represents field Row of the ConflictData union type.
+	ConflictDataRow struct{ Value RowInfo }
+	// __ConflictDataReflect describes the ConflictData union type.
+	__ConflictDataReflect struct {
+		Name  string `vdl:"v.io/syncbase/v23/services/syncbase/nosql.ConflictData"`
+		Type  ConflictData
+		Union struct {
+			Batch ConflictDataBatch
+			Row   ConflictDataRow
+		}
+	}
+)
+
+func (x ConflictDataBatch) Index() int                         { return 0 }
+func (x ConflictDataBatch) Interface() interface{}             { return x.Value }
+func (x ConflictDataBatch) Name() string                       { return "Batch" }
+func (x ConflictDataBatch) __VDLReflect(__ConflictDataReflect) {}
+
+func (x ConflictDataRow) Index() int                         { return 1 }
+func (x ConflictDataRow) Interface() interface{}             { return x.Value }
+func (x ConflictDataRow) Name() string                       { return "Row" }
+func (x ConflictDataRow) __VDLReflect(__ConflictDataReflect) {}
+
+type BatchInfo struct {
+	// Id is an identifier for a batch contained in a conflict. It is
+	// unique only in the context of a given conflict. Its purpose is solely to
+	// group one or more RowInfo objects together to represent a batch that
+	// was committed by the client.
+	Id uint16
+	// Hint is the hint provided by the client when this batch was committed.
+	Hint string
+	// Source states where the batch comes from.
+	Source BatchSource
+}
+
+func (BatchInfo) __VDLReflect(struct {
+	Name string `vdl:"v.io/syncbase/v23/services/syncbase/nosql.BatchInfo"`
+}) {
+}
+
+// BatchSource represents where the batch was committed.
+type BatchSource int
+
+const (
+	BatchSourceLocal BatchSource = iota
+	BatchSourceRemote
+)
+
+// BatchSourceAll holds all labels for BatchSource.
+var BatchSourceAll = [...]BatchSource{BatchSourceLocal, BatchSourceRemote}
+
+// BatchSourceFromString creates a BatchSource from a string label.
+func BatchSourceFromString(label string) (x BatchSource, err error) {
+	err = x.Set(label)
+	return
+}
+
+// Set assigns label to x.
+func (x *BatchSource) Set(label string) error {
+	switch label {
+	case "Local", "local":
+		*x = BatchSourceLocal
+		return nil
+	case "Remote", "remote":
+		*x = BatchSourceRemote
+		return nil
+	}
+	*x = -1
+	return fmt.Errorf("unknown label %q in nosql.BatchSource", label)
+}
+
+// String returns the string label of x.
+func (x BatchSource) String() string {
+	switch x {
+	case BatchSourceLocal:
+		return "Local"
+	case BatchSourceRemote:
+		return "Remote"
+	}
+	return ""
+}
+
+func (BatchSource) __VDLReflect(struct {
+	Name string `vdl:"v.io/syncbase/v23/services/syncbase/nosql.BatchSource"`
+	Enum struct{ Local, Remote string }
+}) {
+}
+
+// RowInfo contains a single operation performed on a row (in case of read or
+// write) or a range or rows (in case of scan) along with a mapping to each
+// of the batches that this operation belongs to.
+// For example, if Row1 was updated on local syncbase conflicting with a write
+// on remote syncbase as part of two separate batches, then it will be
+// represented by a single RowInfo with Write Operation containing the
+// respective local and remote values along with the batch id for both batches
+// stored in the BatchIds field.
+type RowInfo struct {
+	// Op is a specific operation represented by RowInfo
+	Op Operation
+	// BatchIds contains ids of all batches that this RowInfo is a part of.
+	BatchIds []uint16
+}
+
+func (RowInfo) __VDLReflect(struct {
+	Name string `vdl:"v.io/syncbase/v23/services/syncbase/nosql.RowInfo"`
+}) {
+}
+
+type (
+	// Operation represents any single field of the Operation union type.
+	//
+	// Operation represents a specific operation on a row or a set of rows that is
+	// a part of the conflict.
+	Operation interface {
+		// Index returns the field index.
+		Index() int
+		// Interface returns the field value as an interface.
+		Interface() interface{}
+		// Name returns the field name.
+		Name() string
+		// __VDLReflect describes the Operation union type.
+		__VDLReflect(__OperationReflect)
+	}
+	// OperationRead represents field Read of the Operation union type.
+	//
+	// Read represents a read operation performed on a specific row. For a given
+	// row key there can only be at max one Read operation within a conflict.
+	OperationRead struct{ Value RowOp }
+	// OperationWrite represents field Write of the Operation union type.
+	//
+	// Write represents a write operation performed on a specific row. For a
+	// given row key there can only be at max one Write operation within a
+	// conflict.
+	OperationWrite struct{ Value RowOp }
+	// OperationScan represents field Scan of the Operation union type.
+	//
+	// Scan represents a scan operation performed over a specific range of keys.
+	// For a given key range there can be at max one ScanOp within the Conflict.
+	OperationScan struct{ Value ScanOp }
+	// __OperationReflect describes the Operation union type.
+	__OperationReflect struct {
+		Name  string `vdl:"v.io/syncbase/v23/services/syncbase/nosql.Operation"`
+		Type  Operation
+		Union struct {
+			Read  OperationRead
+			Write OperationWrite
+			Scan  OperationScan
+		}
+	}
+)
+
+func (x OperationRead) Index() int                      { return 0 }
+func (x OperationRead) Interface() interface{}          { return x.Value }
+func (x OperationRead) Name() string                    { return "Read" }
+func (x OperationRead) __VDLReflect(__OperationReflect) {}
+
+func (x OperationWrite) Index() int                      { return 1 }
+func (x OperationWrite) Interface() interface{}          { return x.Value }
+func (x OperationWrite) Name() string                    { return "Write" }
+func (x OperationWrite) __VDLReflect(__OperationReflect) {}
+
+func (x OperationScan) Index() int                      { return 2 }
+func (x OperationScan) Interface() interface{}          { return x.Value }
+func (x OperationScan) Name() string                    { return "Scan" }
+func (x OperationScan) __VDLReflect(__OperationReflect) {}
+
+// RowOp represents a read or write operation on a row corresponding to the
+// given key.
+type RowOp struct {
+	// The key under conflict.
+	Key string
+	// LocalValue contains the value read or written by local syncbase or nil.
+	LocalValue *Value
+	// RemoteValue contains the value read or written by remote syncbase or nil.
+	RemoteValue *Value
+	// AncestorValue contains the value for the key which is the lowest common
+	// ancestor of the two values represented by LocalValue and RemoteValue or
+	// nil if no ancestor exists or if the operation was read.
+	AncestorValue *Value
+}
+
+func (RowOp) __VDLReflect(struct {
+	Name string `vdl:"v.io/syncbase/v23/services/syncbase/nosql.RowOp"`
+}) {
+}
+
+// ScanOp provides details of a scan operation.
+type ScanOp struct {
+	// Start contains the starting key for a range scan.
+	Start string
+	// Limit contains the end key for a range scan.
+	Limit string
+}
+
+func (ScanOp) __VDLReflect(struct {
+	Name string `vdl:"v.io/syncbase/v23/services/syncbase/nosql.ScanOp"`
+}) {
+}
+
+// Value contains the encoded bytes for a row's value stored in syncbase.
+type Value struct {
+	// VOM encoded bytes for a row's value
+	Bytes []byte
+	// Write timestamp for this value in nanoseconds.
+	// TODO(jlodhia): change the timestamp to vdl.time here, in commit timestamp
+	// and clock data.
+	WriteTs int64
+}
+
+func (Value) __VDLReflect(struct {
+	Name string `vdl:"v.io/syncbase/v23/services/syncbase/nosql.Value"`
+}) {
+}
+
+// ValueSelection represents the value that was selected as the final resolution
+// for a conflict.
+type ValueSelection int
+
+const (
+	ValueSelectionLocal ValueSelection = iota
+	ValueSelectionRemote
+	ValueSelectionOther
+)
+
+// ValueSelectionAll holds all labels for ValueSelection.
+var ValueSelectionAll = [...]ValueSelection{ValueSelectionLocal, ValueSelectionRemote, ValueSelectionOther}
+
+// ValueSelectionFromString creates a ValueSelection from a string label.
+func ValueSelectionFromString(label string) (x ValueSelection, err error) {
+	err = x.Set(label)
+	return
+}
+
+// Set assigns label to x.
+func (x *ValueSelection) Set(label string) error {
+	switch label {
+	case "Local", "local":
+		*x = ValueSelectionLocal
+		return nil
+	case "Remote", "remote":
+		*x = ValueSelectionRemote
+		return nil
+	case "Other", "other":
+		*x = ValueSelectionOther
+		return nil
+	}
+	*x = -1
+	return fmt.Errorf("unknown label %q in nosql.ValueSelection", label)
+}
+
+// String returns the string label of x.
+func (x ValueSelection) String() string {
+	switch x {
+	case ValueSelectionLocal:
+		return "Local"
+	case ValueSelectionRemote:
+		return "Remote"
+	case ValueSelectionOther:
+		return "Other"
+	}
+	return ""
+}
+
+func (ValueSelection) __VDLReflect(struct {
+	Name string `vdl:"v.io/syncbase/v23/services/syncbase/nosql.ValueSelection"`
+	Enum struct{ Local, Remote, Other string }
+}) {
+}
+
+// ResolutionInfo contains the application’s reply to a conflict for a key,
+// providing the resolution value. The resolution may be over a group of keys
+// in which case the application must send a stream of ResolutionInfos with
+// the Continued field for the last ResolutionInfo representing the end of the
+// batch with a value false. ResolutionInfos sent as part of a batch will be
+// committed as a batch. If the commit fails, the Conflict will be re-sent.
+type ResolutionInfo struct {
+	// Key is the key under conflict.
+	Key string
+	// Selection represents the value that was selected as resolution.
+	Selection ValueSelection
+	// Result is the resolved value for the key. This field should be used only
+	// if value of Selection field is 'Other'
+	Result *Value
+	// Continued represents whether the batch of ResolutionInfos has ended.
+	Continued bool
+}
+
+func (ResolutionInfo) __VDLReflect(struct {
+	Name string `vdl:"v.io/syncbase/v23/services/syncbase/nosql.ResolutionInfo"`
+}) {
+}
+
 // SchemaMetadata maintains metadata related to the schema of a given database.
 // There is one SchemaMetadata per database.
 type SchemaMetadata struct {
@@ -319,6 +649,17 @@
 	vdl.Register((*SyncGroupSpec)(nil))
 	vdl.Register((*SyncGroupMemberInfo)(nil))
 	vdl.Register((*ResolverType)(nil))
+	vdl.Register((*ConflictInfo)(nil))
+	vdl.Register((*ConflictData)(nil))
+	vdl.Register((*BatchInfo)(nil))
+	vdl.Register((*BatchSource)(nil))
+	vdl.Register((*RowInfo)(nil))
+	vdl.Register((*Operation)(nil))
+	vdl.Register((*RowOp)(nil))
+	vdl.Register((*ScanOp)(nil))
+	vdl.Register((*Value)(nil))
+	vdl.Register((*ValueSelection)(nil))
+	vdl.Register((*ResolutionInfo)(nil))
 	vdl.Register((*SchemaMetadata)(nil))
 	vdl.Register((*CrPolicy)(nil))
 	vdl.Register((*CrRule)(nil))
diff --git a/v23/syncbase/nosql/cr_connection_test.go b/v23/syncbase/nosql/cr_connection_test.go
new file mode 100644
index 0000000..b99c15d
--- /dev/null
+++ b/v23/syncbase/nosql/cr_connection_test.go
@@ -0,0 +1,138 @@
+// Copyright 2015 The Vanadium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package nosql
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	wire "v.io/syncbase/v23/services/syncbase/nosql"
+	"v.io/syncbase/v23/syncbase/nosql/crtestutil"
+	"v.io/v23/context"
+)
+
+type resolverImpl1 struct {
+}
+
+func (ri *resolverImpl1) OnConflict(ctx *context.T, conflict *Conflict) (r Resolution) {
+	return
+}
+
+// TestCrConnectionClose tests if db.Close() correctly shuts down the goroutine
+// that runs conflict resolution code in an infinite loop.
+// Setup:
+//   - A mutex lock is used to cause Advance() on the receive stream to block.
+//     This lock is held by the test before starting off the CR thread.
+//   - ctx.Cancel() is simulated by releasing the mutex lock. This is done right
+//     after calling db.Close()
+//   - The mutex is held again by the test to cause the CR thread to block on
+//     Advance() in case the CR thead did not shutdown. This will lead to
+//     failure of the test.
+// Expected result:
+//   - The CR thread should not get/remain blocked on Advance() after db.Close()
+func TestCrConnectionClose(t *testing.T) {
+	db := NewDatabase("parentName", "db1", getSchema(&resolverImpl1{}))
+	advance := func(st *crtestutil.State) bool {
+		fmt.Println("Advance() for ConflictStreamImpl called")
+		st.IsBlocked = true
+		st.Mu.Lock()
+		defer st.Mu.Unlock()
+		fmt.Println("Advance() for ConflictStreamImpl returning")
+		st.IsBlocked = false
+		return false
+	}
+
+	st := &crtestutil.State{}
+	crStream := &crtestutil.CrStreamImpl{
+		C: &crtestutil.ConflictStreamImpl{St: st, AdvanceFn: advance},
+		R: &crtestutil.ResolutionStreamImpl{St: st},
+	}
+	db.c = crtestutil.MockDbClient(db.c, crStream)
+	db.crState.reconnectWaitTime = 10 * time.Millisecond
+
+	ctx, _ := context.RootContext()
+	st.Mu.Lock()                                // causes Advance() to block
+	db.EnforceSchema(ctx)                       // kicks off the CR thread
+	for i := 0; i < 100 && !st.IsBlocked; i++ { // wait till Advance() call is blocked
+		time.Sleep(time.Millisecond)
+	}
+	db.Close()
+	st.Mu.Unlock() // simulate ctx cancel()
+	for i := 0; i < 100 && st.IsBlocked; i++ {
+		time.Sleep(time.Millisecond)
+	}
+
+	st.Mu.Lock() // causes Advance() to block if called again
+	// let conflict resolution routine complete its cleanup.
+	// If the code does not work as intended then it will end up reconnecting
+	// the stream and blocking on the mutex
+	time.Sleep(time.Millisecond)
+	if st.IsBlocked {
+		t.Error("Error: The conflict resolution routine did not die")
+	}
+}
+
+// TestCrConnectionReestablish tests if the CR thread reestablishes a broken
+// connection stream with syncbase.
+// Setup:
+//   - Advance() is allowed to return immediately 3 times with return value
+//     false signifying 3 consecutive breaks in the connection.
+//   - A mutex lock is used to cause Advance() on the recieve stream to block
+//     on the 4th attempt. This lock is held by the test before starting off the
+//     CR thread. This simulates a stable connection between CR and syncbase.
+// Expected result:
+//   - Advance() should get blocked after some time.
+//   - st.AdvanceCount should be equal to 3
+func TestCrConnectionReestablish(t *testing.T) {
+	db := NewDatabase("parentName", "db1", getSchema(&resolverImpl1{}))
+	advance := func(st *crtestutil.State) bool {
+		fmt.Println("Advance() for ConflictStreamImpl called")
+		st.AdvanceCount++
+		if st.AdvanceCount > 2 {
+			// Connection stays stable after 3 attempts
+			st.IsBlocked = true
+			st.Mu.Lock()
+			defer st.Mu.Unlock()
+		}
+		fmt.Println("Advance() for ConflictStreamImpl returning")
+		st.IsBlocked = false
+		return false
+	}
+
+	st := &crtestutil.State{}
+	crStream := &crtestutil.CrStreamImpl{
+		C: &crtestutil.ConflictStreamImpl{St: st, AdvanceFn: advance},
+		R: &crtestutil.ResolutionStreamImpl{St: st},
+	}
+	db.c = crtestutil.MockDbClient(db.c, crStream)
+	db.crState.reconnectWaitTime = 10 * time.Millisecond
+
+	ctx, _ := context.RootContext()
+	st.Mu.Lock() // causes Advance() to block
+	db.EnforceSchema(ctx)
+	for i := 0; i < 100 && !st.IsBlocked; i++ {
+		time.Sleep(time.Millisecond) // wait till Advance() call is blocked
+	}
+
+	if !st.IsBlocked {
+		t.Error("Error: Advance() did not block")
+	}
+	if st.AdvanceCount != 3 {
+		t.Errorf("Error: Advance() expected to be called 3 times, instead called %d times", st.AdvanceCount)
+	}
+
+	// Shutdown the cr routine
+	db.Close()
+	st.Mu.Unlock()
+}
+
+func getSchema(cr ConflictResolver) *Schema {
+	return &Schema{
+		Metadata: wire.SchemaMetadata{},
+		Upgrader: nil,
+		Resolver: cr,
+	}
+}
diff --git a/v23/syncbase/nosql/cr_test.go b/v23/syncbase/nosql/cr_test.go
new file mode 100644
index 0000000..f9618d9
--- /dev/null
+++ b/v23/syncbase/nosql/cr_test.go
@@ -0,0 +1,605 @@
+// Copyright 2015 The Vanadium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package nosql
+
+import (
+	"bytes"
+	"strings"
+	"testing"
+	"time"
+
+	wire "v.io/syncbase/v23/services/syncbase/nosql"
+	"v.io/syncbase/v23/syncbase/nosql/crtestutil"
+	"v.io/v23/context"
+	"v.io/v23/vom"
+)
+
+// TODO(jlodhia): extract this test (along with the crtest helpers) to
+// nosql_test, or alternatively, move the crtest helper to an internal dir.
+
+
+////////////////////////////////////////
+// Client Code
+
+// Client constants
+const (
+	TaskIdPrefix = "task:"
+	ListIdPrefix = "list:"
+
+	List1 = "list:1"
+
+	Task1 = "task:1:1"
+	Task2 = "task:1:2"
+
+	HintListDelete = "ListDelete"
+
+	HintTextEdit   = "TaskTextEdit"
+	HintDone       = "TaskDoneToggle"
+	HintTaskDelete = "TaskDelete"
+	HintTaskAdd    = "TaskAdd"
+
+	TextNew = "EditedText"
+)
+
+// Client data structures
+type Task struct {
+	Text   string
+	Done   bool
+	DoneTs int64
+	ListId string
+}
+
+type List struct {
+	Name      string
+	TaskCount int16
+	DoneCount int16
+}
+
+// Client conflict reolution impl
+type ConflictResolverImpl struct {
+}
+
+func (ri *ConflictResolverImpl) OnConflict(ctx *context.T, conflict *Conflict) Resolution {
+	taskRows := findRowsByType(conflict, TaskIdPrefix)
+	result := ri.handleTasks(ctx, taskRows)
+
+	listRows := findRowsByType(conflict, ListIdPrefix)
+	ri.handleLists(ctx, listRows, result)
+	return Resolution{ResultSet: result}
+}
+
+// handleTasks does a merge of fields within each task under conflict.
+// It handles conflict on fields as follows:
+// Text: Last writer wins
+// Done: Depends on DoneTs
+// DoneTs: The side that is not equal to ancestor and is the lower of the two
+//         wins. This means that if a user marks a task done on one device
+//         and then later marks the same task done on another device that has
+//         not synced yet, the done timestamp that is selected will be from the
+//         former action.
+func (ri *ConflictResolverImpl) handleTasks(ctx *context.T, tasks []ConflictRow) map[string]ResolvedRow {
+	result := map[string]ResolvedRow{}
+	for _, t := range tasks {
+		if isRowAdded(t) {
+			result[t.Key] = handleRowAdd(t)
+			continue
+		}
+		if isRowDeleted(t) {
+			result[t.Key] = ResolvedRow{Key: t.Key, Result: nil}
+			continue
+		}
+
+		var localTask, remoteTask, ancestorTask, resolvedTask Task
+		t.LocalValue.Get(&localTask)
+		t.RemoteValue.Get(&remoteTask)
+		t.AncestorValue.Get(&ancestorTask)
+
+		// prime the reolved task with local version for simplicity.
+		resolvedTask = localTask
+
+		// check field text for conflict
+		if localTask.Text != remoteTask.Text {
+			if remoteTask.Text != ancestorTask.Text {
+				if (localTask.Text == ancestorTask.Text) || (t.RemoteValue.WriteTs.After(t.LocalValue.WriteTs)) {
+					resolvedTask.Text = remoteTask.Text
+				}
+			}
+		}
+
+		// check field done for conflict
+		if localTask.DoneTs != remoteTask.DoneTs {
+			if remoteTask.DoneTs != ancestorTask.DoneTs {
+				if (localTask.DoneTs == ancestorTask.DoneTs) || (remoteTask.DoneTs < localTask.DoneTs) {
+					resolvedTask.DoneTs = remoteTask.DoneTs
+					resolvedTask.Done = remoteTask.Done
+				}
+			}
+		}
+		resValue, _ := NewValue(ctx, &resolvedTask)
+		result[t.Key] = ResolvedRow{Key: t.Key, Result: resValue}
+	}
+	return result
+}
+
+// handleLists does a merge of fields within each list under conflict. If a list
+// is deleted, it looks at the resolution map, finds all tasks that belong to
+// the list and changes the resolution to delete for those tasks.
+// It handles conflict on fields as follows:
+// Name: Last writer wins
+// TaskCount: Add both counts and subtract the ancestor count.
+// DoneCount: Add both counts and subtract the ancestor count.
+// NOTE: The above counters are resolved in a naive way for simplicity of this
+// test. There are many other ways to represent the counters in a more correct
+// way.
+func (ri *ConflictResolverImpl) handleLists(ctx *context.T, lists []ConflictRow, resolution map[string]ResolvedRow) {
+	for _, l := range lists {
+		if isRowAdded(l) {
+			resolution[l.Key] = handleRowAdd(l)
+			continue
+		}
+		if isRowDeleted(l) {
+			resolution[l.Key] = ResolvedRow{Key: l.Key, Result: nil}
+			deleteTasksForList(resolution, l.Key)
+			continue
+		}
+
+		var localList, remoteList, ancestorList, resolvedList List
+		l.LocalValue.Get(&localList)
+		l.RemoteValue.Get(&remoteList)
+		l.AncestorValue.Get(&ancestorList)
+
+		// prime the reolved task with local version for simplicity.
+		resolvedList = localList
+
+		// check field Name for conflict
+		if localList.Name != remoteList.Name {
+			if remoteList.Name != ancestorList.Name {
+				if (localList.Name == ancestorList.Name) || (l.RemoteValue.WriteTs.After(l.LocalValue.WriteTs)) {
+					resolvedList.Name = remoteList.Name
+				}
+			}
+		}
+
+		// check field TaskCount for conflict
+		if localList.TaskCount != remoteList.TaskCount {
+			if remoteList.TaskCount != ancestorList.TaskCount {
+				if localList.TaskCount == ancestorList.TaskCount {
+					resolvedList.TaskCount = remoteList.TaskCount
+				} else {
+					resolvedList.TaskCount = localList.TaskCount + remoteList.TaskCount - ancestorList.TaskCount
+				}
+			}
+		}
+
+		// check field DoneCount for conflict
+		if localList.DoneCount != remoteList.DoneCount {
+			if remoteList.DoneCount != ancestorList.DoneCount {
+				if localList.DoneCount == ancestorList.DoneCount {
+					resolvedList.DoneCount = remoteList.DoneCount
+				} else {
+					resolvedList.DoneCount = localList.DoneCount + remoteList.DoneCount - ancestorList.DoneCount
+				}
+			}
+		}
+		resValue, _ := NewValue(ctx, &resolvedList)
+		resolution[l.Key] = ResolvedRow{Key: l.Key, Result: resValue}
+	}
+}
+
+func deleteTasksForList(resolution map[string]ResolvedRow, listKey string) {
+	// find key prefix for tasks belonging to list
+	taskKeyPrefix := strings.Replace(listKey, ListIdPrefix, TaskIdPrefix, 1) + ":"
+	for k, v := range resolution {
+		if strings.HasPrefix(k, taskKeyPrefix) {
+			v.Result = nil
+			resolution[k] = v
+		}
+	}
+}
+
+func handleRowAdd(r ConflictRow) ResolvedRow {
+	var resolvedTask *Value
+	// first one to add wins
+	if r.RemoteValue == nil {
+		resolvedTask = r.LocalValue
+	} else if r.LocalValue == nil {
+		resolvedTask = r.RemoteValue
+	} else if r.LocalValue.WriteTs.Before(r.RemoteValue.WriteTs) {
+		resolvedTask = r.LocalValue
+	} else {
+		resolvedTask = r.RemoteValue
+	}
+	return ResolvedRow{Key: r.Key, Result: resolvedTask}
+}
+
+func isRowDeleted(r ConflictRow) bool {
+	return r.AncestorValue != nil &&
+		(r.LocalValue == nil || r.RemoteValue == nil)
+}
+
+func isRowAdded(r ConflictRow) bool {
+	return r.AncestorValue == nil &&
+		(r.LocalValue != nil || r.RemoteValue != nil)
+}
+
+func findRowsByType(conflict *Conflict, typePrefix string) []ConflictRow {
+	rows := []ConflictRow{}
+	if conflict.WriteSet != nil {
+		for k, v := range conflict.WriteSet.ByKey {
+			if strings.HasPrefix(k, typePrefix) {
+				rows = append(rows, v)
+			}
+		}
+	}
+	return rows
+}
+
+////////////////////////////////////////
+// Tests
+
+func TestConflictResolutionSingleObject(t *testing.T) {
+	mockConflit, expResult := simpleConflictStream(t)
+	RunTest(t, mockConflit, expResult, false)
+}
+
+func TestConflictResolutionAddDelete(t *testing.T) {
+	mockConflit, expResult := addDeleteConflictStream(t)
+	RunTest(t, mockConflit, expResult, true)
+}
+
+func TestConflictResolutionIntersectingConflicts(t *testing.T) {
+	mockConflict, expResult := twoIntersectingBatchesConflictStream(t)
+	RunTest(t, mockConflict, expResult, true)
+}
+
+func TestConflictResolutionListDelete(t *testing.T) {
+	mockConflit, expResult := listDeleteConflictsWithTaskAdd(t)
+	RunTest(t, mockConflit, expResult, true)
+}
+
+func RunTest(t *testing.T, mockConflit []wire.ConflictInfo, expResult map[string]wire.ResolutionInfo, singleBatch bool) {
+	db := NewDatabase("parentName", "db1", getSchema(&ConflictResolverImpl{}))
+	advance := func(st *crtestutil.State) bool {
+		if st.ValIndex >= len(mockConflit) {
+			st.IsBlocked = true
+			st.Mu.Lock()
+			defer st.Mu.Unlock()
+		}
+		st.IsBlocked = false
+		st.Val = mockConflit[st.ValIndex]
+		st.ValIndex++
+		return true
+	}
+
+	st := &crtestutil.State{}
+	crStream := &crtestutil.CrStreamImpl{
+		C: &crtestutil.ConflictStreamImpl{St: st, AdvanceFn: advance},
+		R: &crtestutil.ResolutionStreamImpl{St: st},
+	}
+	db.c = crtestutil.MockDbClient(db.c, crStream)
+	db.crState.reconnectWaitTime = 10 * time.Millisecond
+
+	ctx, _ := context.RootContext()
+	st.Mu.Lock() // causes Advance() to block
+	db.EnforceSchema(ctx)
+	for i := 0; i < 100 && (len(st.Result) != len(expResult)); i++ {
+		time.Sleep(time.Millisecond) // wait till Advance() call is blocked
+	}
+	if len(st.Result) != len(expResult) {
+		t.Errorf("\n Unexpected number of results. Expected: %d, actual: %d", len(expResult), len(st.Result))
+	}
+	for i, result := range st.Result {
+		compareResult(t, expResult[result.Key], result)
+		if singleBatch {
+			if result.Continued != (i < len(st.Result)-1) {
+				t.Error("\nUnexpected value for continued in single batch")
+			}
+		} else {
+			if result.Continued != false {
+				t.Error("\nUnexpected value for continued in multi batch")
+			}
+		}
+	}
+}
+
+////////////////////////////////////////
+// Test helpers to create test data
+
+// Consider a database with a list List1 with two tasks Task1, Task2. The
+// following helper methods create conflicts with these three objects in various
+// ways.
+
+// simpleConflictStream prepares a conflict stream with two conflicts, one
+// each for Task1 and Task2. No batch is involved and Task1 is independent of
+// Task2. Assume that List does not get updated for these task updates.
+// Local syncbase changes Text field for Task1 and marks Task2 done.
+// Remote syncbase marks Task1 done and Task2 done.
+// Expected result is a merge for Task1 and Task2.
+func simpleConflictStream(t *testing.T) ([]wire.ConflictInfo, map[string]wire.ResolutionInfo) {
+	// Conflict for task1
+	row1 := makeRowInfo(Task1,
+		encode(&Task{"OriginalText", false, -1, List1}), 1, // ancestor
+		encode(&Task{TextNew, false, -1, List1}), 3, // local
+		encode(&Task{"OriginalText", true, 20, List1}), 2) // remote
+	c1 := makeConflictInfo(row1, false)
+
+	// Expected result
+	r1 := makeResolution(Task1, encode(&Task{TextNew, true, 20, List1}), wire.ValueSelectionOther)
+
+	// Conflict for task2
+	row2 := makeRowInfo(Task2,
+		encode(&Task{"Text1", false, -1, List1}), 1, // ancestor
+		encode(&Task{"Text1", true, 100, List1}), 3, // local
+		encode(&Task{"Text1", true, 20, List1}), 2) // remote
+	c2 := makeConflictInfo(row2, false)
+
+	// Expected result
+	r2 := makeResolution(Task2, encode(&Task{"Text1", true, 20, List1}),
+		wire.ValueSelectionOther) // for simplicity the test resolver does not optimize
+
+	respMap := map[string]wire.ResolutionInfo{}
+	respMap[r1.Key] = r1
+	respMap[r2.Key] = r2
+
+	return []wire.ConflictInfo{c1, c2}, respMap
+}
+
+// addDeleteConflictStream prepares a conflict stream with one conflict
+// containing Task1, Task2 and List1. List contains a task count and hence
+// addition and deletion of task leads to update to the list.
+// Local syncbase deletes Task1 and updates List1 in a batch.
+// Remote syncbase adds Task2 and updates List1 in a batch.
+// Expected result is Task1 gets deleted, Task2 gets added and List1 shows
+// the correct taskcount after the two operations.
+func addDeleteConflictStream(t *testing.T) ([]wire.ConflictInfo, map[string]wire.ResolutionInfo) {
+	// Batch1 is local and contains task1 and list1
+	b1 := makeConflictBatch(1, HintTaskDelete, wire.BatchSourceLocal, true)
+
+	// Batch2 is remote and contains task2 and list1
+	b2 := makeConflictBatch(2, HintTaskAdd, wire.BatchSourceRemote, true)
+
+	// Deletion for task1 on local syncbase
+	row1 := makeRowInfo(Task1,
+		encode(&Task{"TaskToBeRemoved", false, -1, List1}), 1, // ancestor
+		nil, -1, // local
+		encode(&Task{"TaskToBeRemoved", false, -1, List1}), 1) // remote
+	row1.BatchIds = []uint16{1}
+	c1 := makeConflictInfo(row1, true)
+
+
+	// Expected result
+	r1 := makeResolution(Task1, nil, wire.ValueSelectionOther)
+
+	// Addition for task2 on remote syncbase
+	row2 := makeRowInfo(Task2,
+		nil, -1, // ancestor value
+		nil, -1, // local value
+		encode(&Task{"AddedTask", false, -1, List1}), 2) // remote value
+	row2.BatchIds = []uint16{2}
+	c2 := makeConflictInfo(row2, true)
+
+	// Expected result
+	r2 := makeResolution(Task2, encode(&Task{"AddedTask", false, -1, List1}), wire.ValueSelectionRemote)
+
+	// Update to List on both sided on filed TaskCount
+	listName := "Groceries"
+	row3 := makeRowInfo(List1,
+		encode(&List{listName, 8, 0}), 25, // ancestor value
+		encode(&List{listName, 7, 0}), 33, // local value
+		encode(&List{listName, 9, 0}), 44) // remote value
+	row3.BatchIds = []uint16{1, 2}
+	c3 := makeConflictInfo(row3, false)
+
+	// Expected result
+	r3 := makeResolution(List1, encode(&List{listName, 8, 0}), wire.ValueSelectionOther)
+
+	respMap := map[string]wire.ResolutionInfo{}
+	respMap[r1.Key] = r1
+	respMap[r2.Key] = r2
+	respMap[r3.Key] = r3
+
+	return []wire.ConflictInfo{b1, b2, c1, c2, c3}, respMap
+}
+
+// twoIntersectingBatchesConflictStream prepares a conflict stream with one
+// conflict containing Task1, Task2 and List1. Here we assume that List keeps
+// a count of done tasks.
+// Local syncbase marks Task1 done and updates done count on List1. Later it
+// it also marks Task2 done and updates done count on List1.
+// Remote syncbase updates field Text on Task1 followed by an update to field
+// Text on Task2 (not in a batch).
+// Expected result is a merge for Task1 and Task2 and Local version of List1
+// being accepted.
+func twoIntersectingBatchesConflictStream(t *testing.T) ([]wire.ConflictInfo, map[string]wire.ResolutionInfo) {
+	// Batch1 is local and contains task1 and list1
+	b1 := makeConflictBatch(1, HintDone, wire.BatchSourceLocal, true)
+
+	// Batch2 is local and contains task2 and list1
+	b2 := makeConflictBatch(2, HintDone, wire.BatchSourceLocal, true)
+
+	// Batch3 is remote and contains task1
+	b3 := makeConflictBatch(3, HintTextEdit, wire.BatchSourceRemote, true)
+
+	// Batch4 is remote and contains task2
+	b4 := makeConflictBatch(4, HintTextEdit, wire.BatchSourceRemote, true)
+
+	// For task1, mark done on local and edit text on remote
+	row1 := makeRowInfo(Task1,
+		encode(&Task{"TaskOrig", false, -1, List1}), 1, // ancestor
+		encode(&Task{"TaskOrig", true, 204, List1}), 5, // local
+		encode(&Task{"TaskEdit", false, -1, List1}), 3) // remote
+	row1.BatchIds = []uint16{1, 3}
+	c1 := makeConflictInfo(row1, true)
+
+	// Expected result
+	r1 := makeResolution(Task1, encode(&Task{"TaskEdit", true, 204, List1}), wire.ValueSelectionOther)
+
+	// For task2, mark done on local and edit text on remote
+	row2 := makeRowInfo(Task2,
+		encode(&Task{"TaskOrig", false, -1, List1}), 2, // ancestor
+		encode(&Task{"TaskOrig", true, 204, List1}), 5, // local
+		encode(&Task{"TaskEdit", false, -1, List1}), 3) // remote
+	row2.BatchIds = []uint16{2, 4}
+	c2 := makeConflictInfo(row2, true)
+
+	// Expected result
+	r2 := makeResolution(Task2, encode(&Task{"TaskEdit", true, 204, List1}), wire.ValueSelectionOther)
+
+	// Update to List on Local for updating TaskDoneCount
+	listName := "Groceries"
+	row3 := makeRowInfo(List1,
+		encode(&List{listName, 8, 0}), 1, // ancestor value
+		encode(&List{listName, 8, 2}), 5, // local value
+		encode(&List{listName, 8, 0}), 3) // remote value
+	row3.BatchIds = []uint16{1, 2}
+	c3 := makeConflictInfo(row3, false)
+
+	// Expected result
+	r3 := makeResolution(List1, encode(&List{listName, 8, 2}), wire.ValueSelectionOther)
+
+	respMap := map[string]wire.ResolutionInfo{}
+	respMap[r1.Key] = r1
+	respMap[r2.Key] = r2
+	respMap[r3.Key] = r3
+
+	return []wire.ConflictInfo{b1, b2, b3, b4, c1, c2, c3}, respMap
+}
+
+// listDeleteConflictsWithTaskAdd prepares a conflict stream with one
+// conflict containing Task1, Task2 and List1. Assume Task2 does not exist yet.
+// Local syncbase deletes List1 and Task1 as a batch.
+// Remote syncbase adds Task2 along with an update to List1 as a batch.
+// Expected result is deletion of List1, Task1 and Task2.
+func listDeleteConflictsWithTaskAdd(t *testing.T) ([]wire.ConflictInfo, map[string]wire.ResolutionInfo) {
+	// Batch1 is local and contains task1 and list1
+	b1 := makeConflictBatch(1, HintListDelete, wire.BatchSourceLocal, true)
+
+	// Batch2 is remote and contains task2 and list1
+	b2 := makeConflictBatch(2, HintTaskAdd, wire.BatchSourceRemote, true)
+
+	// Delition for list1 on local syncbase
+	listName := "Groceries"
+	row0 := makeRowInfo(List1,
+		encode(&List{listName, 1, 0}), 1, // ancestor value
+		nil, 5, // local value
+		encode(&List{listName, 2, 0}), 3) // remote value
+	row0.BatchIds = []uint16{1, 2}
+	c0 := makeConflictInfo(row0, true)
+
+	// Expected result
+	r0 := makeResolution(List1, nil, wire.ValueSelectionOther)
+
+	// Deletion for task1 on local syncbase along with list1
+	row1 := makeRowInfo(Task1,
+		encode(&Task{"TaskToBeRemoved", false, -1, List1}), 1, // ancestor
+		nil, -1, // local
+		encode(&Task{"TaskToBeRemoved", false, -1, List1}), 1) // remote
+	row1.BatchIds = []uint16{1}
+	c1 := makeConflictInfo(row1, true)
+
+	// Expected result
+	r1 := makeResolution(Task1, nil, wire.ValueSelectionOther)
+
+	// Addition for task2 on remote syncbase
+	row2 := makeRowInfo(Task2,
+		nil, -1, // ancestor value
+		nil, -1, // local value
+		encode(&Task{"AddedTask", false, -1, List1}), 2) // remote value
+	row2.BatchIds = []uint16{2}
+	c2 := makeConflictInfo(row2, false)
+
+	// Expected result
+	r2 := makeResolution(Task2, nil, wire.ValueSelectionOther)
+
+	respMap := map[string]wire.ResolutionInfo{}
+	respMap[r0.Key] = r0
+	respMap[r1.Key] = r1
+	respMap[r2.Key] = r2
+
+	return []wire.ConflictInfo{b1, b2, c0, c1, c2}, respMap
+}
+
+func makeResolution(key string, result []byte, selection wire.ValueSelection) wire.ResolutionInfo {
+	r := wire.ResolutionInfo{}
+	r.Key = key
+	if result != nil {
+		r.Result = &wire.Value{
+			Bytes:   result,
+			WriteTs: -1,
+		}
+	}
+	r.Selection = selection
+	return r
+}
+
+func makeConflictBatch(id uint16, hint string, source wire.BatchSource, continued bool) wire.ConflictInfo {
+	batch := wire.BatchInfo{Id: id, Hint: hint, Source: source}
+	return wire.ConflictInfo{Data: wire.ConflictDataBatch{Value: batch}, Continued: continued}
+}
+
+func makeConflictInfo(row wire.RowInfo, continued bool) wire.ConflictInfo {
+	return wire.ConflictInfo{
+		Data: wire.ConflictDataRow{Value: row},
+		Continued: continued,
+	}
+}
+
+func makeRowInfo(key string, ancestor []byte, ats int64, local []byte, lts int64, remote []byte, rts int64) wire.RowInfo {
+	op := wire.RowOp{}
+	op.Key = key
+
+	if ancestor != nil {
+		op.AncestorValue = &wire.Value{
+			Bytes:   ancestor,
+			WriteTs: ats,
+		}
+	}
+
+	if local != nil {
+		op.LocalValue = &wire.Value{
+			Bytes:   local,
+			WriteTs: lts,
+		}
+	}
+
+	if remote != nil {
+		op.RemoteValue = &wire.Value{
+			Bytes:   remote,
+			WriteTs: rts,
+		}
+	}
+	return wire.RowInfo{
+		Op: wire.OperationWrite{Value: op},
+	}
+}
+
+func compareResult(t *testing.T, expected wire.ResolutionInfo, actual wire.ResolutionInfo) {
+	if actual.Key != expected.Key {
+		t.Error("Key does not match")
+	}
+	if actual.Selection != expected.Selection {
+		t.Errorf("Key: %s", expected.Key)
+		t.Errorf("Expected selection: %v, actual selection: %v", expected.Selection, actual.Selection)
+	}
+	if (expected.Result == nil) && (actual.Result != nil) {
+		t.Errorf("Key: %s", expected.Key)
+		t.Error("Result expected to be nil but found non nil")
+	}
+	if expected.Result != nil {
+		if actual.Result == nil {
+			t.Errorf("Key: %s", expected.Key)
+			t.Error("Result found nil")
+		}
+		if bytes.Compare(actual.Result.Bytes, expected.Result.Bytes) != 0 {
+			t.Errorf("Key: %s", expected.Key)
+			t.Error("Result bytes do not match")
+		}
+	}
+}
+
+func encode(value interface{}) []byte {
+	v, _ := vom.Encode(value)
+	return v
+}
diff --git a/v23/syncbase/nosql/crtestutil/mock_dbclient.go b/v23/syncbase/nosql/crtestutil/mock_dbclient.go
new file mode 100644
index 0000000..99e27ee
--- /dev/null
+++ b/v23/syncbase/nosql/crtestutil/mock_dbclient.go
@@ -0,0 +1,34 @@
+// Copyright 2015 The Vanadium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package crtestutil
+
+import (
+	wire "v.io/syncbase/v23/services/syncbase/nosql"
+	"v.io/v23/context"
+	"v.io/v23/rpc"
+	"v.io/v23/verror"
+)
+
+type MockWireDatabaseClient struct {
+	wire.DatabaseClientMethods
+	crs *CrStreamImpl
+}
+
+func MockDbClient(c wire.DatabaseClientMethods, crs *CrStreamImpl) *MockWireDatabaseClient {
+	return &MockWireDatabaseClient{c, crs}
+}
+
+func (wclient *MockWireDatabaseClient) GetSchemaMetadata(ctx *context.T, opts ...rpc.CallOpt) (sm wire.SchemaMetadata, err error) {
+	err = verror.NewErrNoExist(ctx)
+	return
+}
+
+func (wclient *MockWireDatabaseClient) SetSchemaMetadata(ctx *context.T, i0 wire.SchemaMetadata, opts ...rpc.CallOpt) error {
+	return nil
+}
+
+func (wclient *MockWireDatabaseClient) StartConflictResolver(ctx *context.T, opts ...rpc.CallOpt) (wire.ConflictManagerStartConflictResolverClientCall, error) {
+	return wclient.crs, nil
+}
diff --git a/v23/syncbase/nosql/crtestutil/mock_stream.go b/v23/syncbase/nosql/crtestutil/mock_stream.go
new file mode 100644
index 0000000..3bb8f4d
--- /dev/null
+++ b/v23/syncbase/nosql/crtestutil/mock_stream.go
@@ -0,0 +1,119 @@
+// Copyright 2015 The Vanadium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package crtestutil
+
+import (
+	"sync"
+
+	wire "v.io/syncbase/v23/services/syncbase/nosql"
+)
+
+var _ wire.ConflictManagerStartConflictResolverClientCall = (*CrStreamImpl)(nil)
+
+type State struct {
+	// ConflictStream state
+	IsBlocked    bool
+	Mu           sync.Mutex
+	Val          wire.ConflictInfo
+	AdvanceCount int
+	ValIndex     int
+	// ResolutionStream state variables
+	Result []wire.ResolutionInfo
+}
+
+type CrStreamImpl struct {
+	C ConflictStream
+	R ResolutionStream
+}
+
+func (s *CrStreamImpl) RecvStream() interface {
+	Advance() bool
+	Value() wire.ConflictInfo
+	Err() error
+} {
+	return recvStreamImpl{s.C}
+}
+
+func (s *CrStreamImpl) SendStream() interface {
+	Send(item wire.ResolutionInfo) error
+	Close() error
+} {
+	return sendStreamImpl{s.R}
+}
+
+func (s *CrStreamImpl) Finish() error {
+	return nil
+}
+
+type recvStreamImpl struct {
+	c ConflictStream
+}
+
+func (rs recvStreamImpl) Advance() bool {
+	return rs.c.Advance()
+}
+func (rs recvStreamImpl) Value() wire.ConflictInfo {
+	return rs.c.Value()
+}
+func (rs recvStreamImpl) Err() error {
+	return rs.c.Err()
+}
+
+type sendStreamImpl struct {
+	r ResolutionStream
+}
+
+func (ss sendStreamImpl) Send(item wire.ResolutionInfo) error {
+	return ss.r.Send(item)
+}
+func (c sendStreamImpl) Close() error {
+	return nil
+}
+
+type ConflictStream interface {
+	Advance() bool
+	Value() wire.ConflictInfo
+	Err() error
+}
+
+type ResolutionStream interface {
+	Send(item wire.ResolutionInfo) error
+}
+
+type ConflictStreamImpl struct {
+	St        *State
+	AdvanceFn func(*State) bool
+}
+
+func (cs *ConflictStreamImpl) Advance() bool {
+	return cs.AdvanceFn(cs.St)
+}
+func (cs *ConflictStreamImpl) Value() wire.ConflictInfo {
+	return cs.St.Val
+}
+func (cs *ConflictStreamImpl) Err() error {
+	return &TestError{"Stream broken"}
+}
+
+type ResolutionStreamImpl struct {
+	St *State
+}
+
+func (rs *ResolutionStreamImpl) Send(item wire.ResolutionInfo) error {
+	if rs.St.Result == nil {
+		rs.St.Result = []wire.ResolutionInfo{item}
+		return nil
+	}
+	rs.St.Result = append(rs.St.Result, item)
+	return nil
+}
+
+type TestError struct {
+	str string
+}
+
+func (e *TestError) Error() string {
+	return e.str
+}
diff --git a/v23/syncbase/nosql/database.go b/v23/syncbase/nosql/database.go
index d48f450..42b5270 100644
--- a/v23/syncbase/nosql/database.go
+++ b/v23/syncbase/nosql/database.go
@@ -5,6 +5,9 @@
 package nosql
 
 import (
+	"sync"
+	"time"
+
 	wire "v.io/syncbase/v23/services/syncbase/nosql"
 	"v.io/syncbase/v23/syncbase/util"
 	"v.io/v23/context"
@@ -15,6 +18,12 @@
 	"v.io/x/lib/vlog"
 )
 
+const (
+	// Wait time before we try to reconnect a broken conflict resolution stream.
+	waitBeforeReconnectInMillis = 2 * time.Second
+	reconnectionCount = "rcc"
+)
+
 func NewDatabase(parentFullName, relativeName string, schema *Schema) *database {
 	fullName := naming.Join(parentFullName, relativeName)
 	return &database{
@@ -23,6 +32,9 @@
 		fullName:       fullName,
 		name:           relativeName,
 		schema:         schema,
+		crState: conflictResolutionState{
+			reconnectWaitTime: waitBeforeReconnectInMillis,
+		},
 	}
 }
 
@@ -32,6 +44,31 @@
 	fullName       string
 	name           string
 	schema         *Schema
+	crState        conflictResolutionState
+}
+
+// conflictResolutionState maintains data about the connection of
+// conflict resolution stream with syncbase. It provides a way to disconnect
+// an existing open stream.
+type conflictResolutionState struct {
+	mu                sync.Mutex // guards access to all fields in this struct
+	crContext         *context.T
+	cancelFn          context.CancelFunc
+	isClosed          bool
+	reconnectWaitTime time.Duration
+}
+
+func (crs *conflictResolutionState) disconnect() {
+	crs.mu.Lock()
+	defer crs.mu.Unlock()
+	crs.isClosed = true
+	crs.cancelFn()
+}
+
+func (crs *conflictResolutionState) isDisconnected() bool {
+	crs.mu.Lock()
+	defer crs.mu.Unlock()
+	return crs.isClosed
 }
 
 var _ Database = (*database)(nil)
@@ -168,17 +205,44 @@
 	return createBlob(ctx, d.fullName)
 }
 
-// UpgradeIfOutdated implements Database.UpgradeIfOutdated.
-func (d *database) UpgradeIfOutdated(ctx *context.T) (bool, error) {
+// EnforceSchema implements Database.EnforceSchema.
+func (d *database) EnforceSchema(ctx *context.T) error {
 	var schema *Schema = d.schema
 	if schema == nil {
-		return false, verror.New(verror.ErrBadState, ctx, "Schema or SchemaMetadata cannot be nil. A valid Schema needs to be used when creating DB handle.")
+		return verror.New(verror.ErrBadState, ctx, "Schema or SchemaMetadata cannot be nil. A valid Schema needs to be used when creating DB handle.")
 	}
 
 	if schema.Metadata.Version < 0 {
-		return false, verror.New(verror.ErrBadState, ctx, "Schema version cannot be less than zero.")
+		return verror.New(verror.ErrBadState, ctx, "Schema version cannot be less than zero.")
 	}
 
+	if needsResolver(d.schema.Metadata) && d.schema.Resolver == nil {
+		return verror.New(verror.ErrBadState, ctx, "ResolverTypeAppResolves cannot be used in CrRule without providing a ConflictResolver in Schema.")
+	}
+
+	if _, err := d.upgradeIfOutdated(ctx); err != nil {
+		return err
+	}
+
+	if d.schema.Resolver == nil {
+		return nil
+	}
+
+	childCtx, cancelFn := context.WithCancel(ctx)
+	d.crState.crContext = childCtx
+	d.crState.cancelFn = cancelFn
+
+	go d.establishConflictResolution(childCtx)
+	return nil
+}
+
+// Close implements Database.Close.
+func (d *database) Close() {
+	d.crState.disconnect()
+}
+
+func (d *database) upgradeIfOutdated(ctx *context.T) (bool, error) {
+	var schema *Schema = d.schema
 	schemaMgr := d.getSchemaManager()
 	currMeta, err := schemaMgr.getSchemaMetadata(ctx)
 	if err != nil {
@@ -222,8 +286,182 @@
 	return true, nil
 }
 
+func (d *database) establishConflictResolution(ctx *context.T) {
+	count := 0
+	for {
+		count++
+		vlog.Infof("Starting a new conflict resolution connection. Re-Connection count: %d", count)
+		childCtx := context.WithValue(ctx, reconnectionCount, count)
+		// listenForConflicts is a blocking method which returns only when the
+		// conflict stream is broken.
+		if err := d.listenForConflicts(childCtx); err != nil {
+			vlog.Errorf("Conflict resolution connection ended with error: %v", err)
+		}
+
+		// Check if database is closed and if we need to shutdown conflict
+		// resolution.
+		if d.crState.isDisconnected() {
+			vlog.Infof("Shutting down conflict resolution connection.")
+			break
+		}
+
+		// The connection might have broken because syncbase service went down.
+		// Sleep for a few seconds to allow syncbase to come back up.
+		time.Sleep(d.crState.reconnectWaitTime)
+	}
+}
+
+func (d *database) listenForConflicts(ctx *context.T) error {
+	resolver, err := d.c.StartConflictResolver(ctx)
+	if err != nil {
+		return err
+	}
+	conflictStream := resolver.RecvStream()
+	resolutionStream := resolver.SendStream()
+	var c *Conflict = &Conflict{}
+	for conflictStream.Advance() {
+		row := conflictStream.Value()
+		addRowToConflict(c, &row)
+		if !row.Continued {
+			resolution := d.schema.Resolver.OnConflict(ctx, c)
+			if err := sendResolution(resolutionStream, resolution); err != nil {
+				return err
+			}
+			c = &Conflict{}  // create a new conflict object for the next batch
+		}
+	}
+	if err := conflictStream.Err(); err != nil {
+		return err
+	}
+	return resolver.Finish()
+}
+
+// TODO(jlodhia): Should we check if the Resolution received addresses all
+// conflicts in write set?
+func sendResolution(stream interface {
+	Send(item wire.ResolutionInfo) error
+}, resolution Resolution) error {
+	size := len(resolution.ResultSet)
+	count := 0
+	for _, v := range resolution.ResultSet {
+		count++
+		ri := toResolutionInfo(v, count != size)
+		if err := stream.Send(ri); err != nil {
+			vlog.Error("Error while sending resolution")
+			return err
+		}
+	}
+	return nil
+}
+
+func addRowToConflict(c *Conflict, ci *wire.ConflictInfo) {
+	switch v := ci.Data.(type) {
+	case wire.ConflictDataBatch:
+		if c.Batches == nil {
+			c.Batches = map[uint16]wire.BatchInfo{}
+		}
+		c.Batches[v.Value.Id] = v.Value
+	case wire.ConflictDataRow:
+		rowInfo := v.Value
+		switch op := rowInfo.Op.(type) {
+		case wire.OperationWrite:
+			if c.WriteSet == nil {
+				c.WriteSet = &ConflictRowSet{map[string]ConflictRow{}, map[uint16][]ConflictRow{}}
+			}
+			cr := toConflictRow(op.Value, rowInfo.BatchIds)
+			c.WriteSet.ByKey[cr.Key] = cr
+			for _, bid := range rowInfo.BatchIds {
+				c.WriteSet.ByBatch[bid] = append(c.WriteSet.ByBatch[bid], cr)
+			}
+		case wire.OperationRead:
+			if c.ReadSet == nil {
+				c.ReadSet = &ConflictRowSet{map[string]ConflictRow{}, map[uint16][]ConflictRow{}}
+			}
+			cr := toConflictRow(op.Value, rowInfo.BatchIds)
+			c.ReadSet.ByKey[cr.Key] = cr
+			for _, bid := range rowInfo.BatchIds {
+				c.ReadSet.ByBatch[bid] = append(c.ReadSet.ByBatch[bid], cr)
+			}
+		case wire.OperationScan:
+			if c.ScanSet == nil {
+				c.ScanSet = &ConflictScanSet{map[uint16][]wire.ScanOp{}}
+			}
+			for _, bid := range rowInfo.BatchIds {
+				c.ScanSet.ByBatch[bid] = append(c.ScanSet.ByBatch[bid], op.Value)
+			}
+		}
+	}
+}
+
+func toConflictRow(op wire.RowOp, batchIds []uint16) ConflictRow {
+	var local, remote, ancestor *Value
+	if op.LocalValue != nil {
+		local = &Value{
+			val:       op.LocalValue.Bytes,
+			WriteTs:   toTime(op.LocalValue.WriteTs),
+			selection: wire.ValueSelectionLocal,
+		}
+	}
+	if op.RemoteValue != nil {
+		remote = &Value{
+			val:       op.RemoteValue.Bytes,
+			WriteTs:   toTime(op.RemoteValue.WriteTs),
+			selection: wire.ValueSelectionRemote,
+		}
+	}
+	if op.AncestorValue != nil {
+		ancestor = &Value{
+			val:       op.AncestorValue.Bytes,
+			WriteTs:   toTime(op.AncestorValue.WriteTs),
+			selection: wire.ValueSelectionOther,
+		}
+	}
+	return ConflictRow{
+		Key:           op.Key,
+		LocalValue:    local,
+		RemoteValue:   remote,
+		AncestorValue: ancestor,
+		BatchIds:      batchIds,
+	}
+}
+
+// TODO(jlodhia): remove this method once time is stored as time.Time instead
+// of int64
+func toTime(unixNanos int64) time.Time {
+	return time.Unix(
+		unixNanos / 1e9,  // seconds
+		unixNanos % 1e9)  // nanoseconds
+}
+
+func toResolutionInfo(r ResolvedRow, lastRow bool) wire.ResolutionInfo {
+	sel := wire.ValueSelectionOther
+	resVal := (*wire.Value)(nil)
+	if r.Result != nil {
+		sel = r.Result.selection
+		resVal = &wire.Value{
+			Bytes:   r.Result.val,
+			WriteTs: r.Result.WriteTs.UnixNano(),  // this timestamp is ignored by syncbase
+		}
+	}
+	return wire.ResolutionInfo{
+		Key:       r.Key,
+		Selection: sel,
+		Result:    resVal,
+		Continued: lastRow,
+	}
+}
+
+func needsResolver(metadata wire.SchemaMetadata) bool {
+	for _, rule := range metadata.Policy.Rules {
+		if rule.Resolver == wire.ResolverTypeAppResolves {
+			return true
+		}
+	}
+	return false
+}
+
 func (d *database) getSchemaManager() schemaManagerImpl {
-	return newSchemaManager(d.fullName)
+	return newSchemaManager(d.c)
 }
 
 func (d *database) schemaVersion() int32 {
diff --git a/v23/syncbase/nosql/model.go b/v23/syncbase/nosql/model.go
index 622f82d..d9ae93f 100644
--- a/v23/syncbase/nosql/model.go
+++ b/v23/syncbase/nosql/model.go
@@ -6,12 +6,15 @@
 package nosql
 
 import (
+	"time"
+
 	wire "v.io/syncbase/v23/services/syncbase/nosql"
 	"v.io/syncbase/v23/syncbase/util"
 	"v.io/v23/context"
 	"v.io/v23/security/access"
 	"v.io/v23/services/watch"
 	"v.io/v23/vdl"
+	"v.io/v23/verror"
 	"v.io/v23/vom"
 )
 
@@ -45,6 +48,10 @@
 	// time of the RPC, and will not reflect subsequent writes to keys not yet
 	// reached by the stream.
 	Exec(ctx *context.T, query string) ([]string, ResultStream, error)
+
+	// Close cleans up any state associated with this client handle including
+	// shutting down any open conflict resolution stream.
+	Close()
 }
 
 // Database represents a collection of Tables. Batches, queries, sync, watch,
@@ -131,14 +138,15 @@
 	// Blob returns a handle to the blob with the given BlobRef.
 	Blob(br wire.BlobRef) Blob
 
-	// This method compares the current schema version of the database with the
-	// schema version provided while creating this database handle. If the
-	// current database schema version is lower, then the SchemaUpdater is
+	// EnforceSchema compares the current schema version of the database
+	// with the schema version provided while creating this database handle. If
+	// the current database schema version is lower, then the SchemaUpdater is
 	// called. If SchemaUpdater is successful this method stores the new schema
 	// metadata in database.
-	// Note: schema can be nil, in which case this method skips schema check
-	// and the caller is responsible for maintaining schema sanity.
-	UpgradeIfOutdated(ctx *context.T) (bool, error)
+	// This method also registers a conflict resolver with syncbase to receive
+	// conflicts. Note: schema can be nil, in which case this method skips
+	// schema check and the caller is responsible for maintaining schema sanity.
+	EnforceSchema(ctx *context.T) error
 }
 
 // BatchDatabase is a handle to a set of reads and writes to the database that
@@ -550,4 +558,114 @@
 type Schema struct {
 	Metadata wire.SchemaMetadata
 	Upgrader SchemaUpgrader
+	Resolver ConflictResolver
+}
+
+// ConflictResolver interface allows the App to define resolution of conflicts
+// that it requested to handle.
+type ConflictResolver interface {
+	OnConflict(ctx *context.T, conflict *Conflict) Resolution
+}
+
+// Conflict contains information to fully specify a conflict. Since syncbase
+// supports batches there can be one or more rows within the batch that has a
+// conflict. Each of these rows will be sent together as part of a single
+// conflict. Each row contains an Id of the batch to which it belongs,
+// enabling the client to group together rows that are part of a batch. Note
+// that a single row can be part of more than one batch.
+//
+// WriteSet contains rows that were written.
+// ReadSet contains rows that were read within a batch corresponding to a row
+// within the write set.
+// ScanSet contains scans performed within a batch corresponding to a row
+// within the write set.
+// Batches is a map of unique ids to BatchInfo objects. The id is unique only in
+// the context of a given conflict and is otherwise meaningless.
+type Conflict struct {
+	ReadSet       *ConflictRowSet
+	WriteSet      *ConflictRowSet
+	ScanSet       *ConflictScanSet
+	Batches       map[uint16]wire.BatchInfo
+}
+
+// ConflictRowSet contains a set of rows under conflict. It provides two different
+// ways to access the same set.
+// ByKey is a map of ConflictRows keyed by the row key.
+// ByBatch is a map of []ConflictRows keyed by batch id. This map lets the client
+// access all ConflictRows within this set that contain a given hint.
+type ConflictRowSet struct {
+	ByKey  map[string]ConflictRow
+	ByBatch map[uint16][]ConflictRow
+}
+
+// ConflictScanSet contains a set of scans under conflict.
+// ByBatch is a map of array of ScanOps keyed by batch id.
+type ConflictScanSet struct {
+	ByBatch map[uint16][]wire.ScanOp
+}
+
+// ConflictRow represents a row under conflict.
+// Key is the key for the row.
+// LocalValue is the value present in the local db.
+// RemoteValue is the value received via sync.
+// AncestorValue is the value for the key which is the lowest common
+// ancestor of the two values represented by LocalValue and RemoteValue.
+// AncestorValue is nil if the ConflictRow is a part of the read set.
+// BatchIds is a list of ids of all the batches that this row belongs to.
+type ConflictRow struct {
+	Key           string
+	LocalValue    *Value
+	RemoteValue   *Value
+	AncestorValue *Value
+	BatchIds      []uint16
+}
+
+// Resolution contains the application’s reply to a conflict. It must contain a
+// resolved value for each conflict row within the WriteSet of the given
+// conflict.
+// ResultSet is a map of row key to ResolvedRow.
+type Resolution struct {
+	ResultSet map[string]ResolvedRow
+	// TODO(jlodhia): Hint []string
+}
+
+// ResolvedRow represents a result of resolution of a row under conflict.
+// Key is the key for the row.
+// Selection represents the value that was selected for resolution.
+// Value is the resolved value for the key. This field should be used only
+// if value of Selection field is 'Other'.
+type ResolvedRow struct {
+	Key    string
+	Result *Value
+}
+
+// Value contains a specific version of data for the row under conflict along
+// with the write timestamp and hints associated with the version.
+// WriteTs is the write timestamp for this value.
+type Value struct {
+	val       []byte
+	WriteTs   time.Time
+	selection wire.ValueSelection
+}
+
+// Get takes a reference to an instance of a type that is expected to be
+// represented by Value.
+func (v *Value) Get(value interface{}) error {
+	return vom.Decode(v.val, value)
+}
+
+// NewValue creates a new Value to be added to Resolution.
+func NewValue(ctx *context.T, data interface{}) (*Value, error) {
+	if data == nil {
+		return nil, verror.New(verror.ErrBadArg, ctx, "data cannot be nil")
+	}
+	bytes, err := vom.Encode(data)
+	if err != nil {
+		return nil, err
+	}
+	return &Value{
+		val:       bytes,
+		WriteTs:   time.Now(),  // ignored by syncbase
+		selection: wire.ValueSelectionOther,
+	}, nil
 }
diff --git a/v23/syncbase/nosql/schema.go b/v23/syncbase/nosql/schema.go
index 9f52a1d..8dd480d 100644
--- a/v23/syncbase/nosql/schema.go
+++ b/v23/syncbase/nosql/schema.go
@@ -21,14 +21,12 @@
 // Implementation of SchemaManager (Not part of public client API)
 
 type schemaManagerImpl struct {
-	dbName string
-	c      wire.DatabaseClientMethods
+	c wire.DatabaseClientMethods
 }
 
-func newSchemaManager(dbName string) schemaManagerImpl {
+func newSchemaManager(client wire.DatabaseClientMethods) schemaManagerImpl {
 	return schemaManagerImpl{
-		dbName: dbName,
-		c:      wire.DatabaseClient(dbName),
+		c: client,
 	}
 }
 
diff --git a/v23/syncbase/nosql/schema_test.go b/v23/syncbase/nosql/schema_test.go
index 6d7ed0d..78ebd95 100644
--- a/v23/syncbase/nosql/schema_test.go
+++ b/v23/syncbase/nosql/schema_test.go
@@ -19,10 +19,10 @@
 // This test as following steps:
 // 1) Call NoSQLDatabase() for a non existent db.
 // 2) Create the database, and verify if Schema got stored properly.
-// 3) Call UpgradeIfOutdated() to make sure that the method is no-op and is
+// 3) Call EnforceSchema() to make sure that the method is no-op and is
 //    able to read the schema from db.
 // 4) Call NoSQLDatabase() on the same db to create a new handle with an
-//    upgraded schema, call UpgradeIfOutdated() and check if SchemaUpgrader
+//    upgraded schema, call EnforceSchema() and check if SchemaUpgrader
 //    is called and if the new schema is stored appropriately.
 func TestSchemaCheck(t *testing.T) {
 	ctx, sName, cleanup := tu.SetupOrDie(nil)
@@ -35,9 +35,9 @@
 
 	// Verify that calling Upgrade on a non existing database does not throw
 	// errors.
-	_, err := db1.UpgradeIfOutdated(ctx)
+	err := db1.EnforceSchema(ctx)
 	if err != nil {
-		t.Fatalf("db1.UpgradeIfOutdated() failed: %v", err)
+		t.Fatalf("db1.EnforceSchema() failed: %v", err)
 	}
 	if mockUpgrader.CallCount > 0 {
 		t.Fatal("Call to upgrader was not expected.")
@@ -53,12 +53,8 @@
 	}
 
 	// Make redundant call to Upgrade to verify that it is a no-op
-	result, err1 := db1.UpgradeIfOutdated(ctx)
-	if result {
-		t.Fatalf("db1.UpgradeIfOutdated() should not return true")
-	}
-	if err1 != nil {
-		t.Fatalf("db1.UpgradeIfOutdated() failed: %v", err1)
+	if err := db1.EnforceSchema(ctx); err != nil {
+		t.Fatalf("db1.EnforceSchema() failed: %v", err)
 	}
 	if mockUpgrader.CallCount > 0 {
 		t.Fatal("Call to upgrader was not expected.")
@@ -73,13 +69,8 @@
 	}
 	schema.Metadata.Policy = policy
 	otherdb1 := a.NoSQLDatabase("db1", schema)
-	otherresult, othererr := otherdb1.UpgradeIfOutdated(ctx)
-
-	if !otherresult {
-		t.Fatalf("otherdb1.UpgradeIfOutdated() expected to return true")
-	}
-	if othererr != nil {
-		t.Fatalf("otherdb1.UpgradeIfOutdated() failed: %v", othererr)
+	if err := otherdb1.EnforceSchema(ctx); err != nil {
+		t.Fatalf("otherdb1.EnforceSchema() failed: %v", err)
 	}
 	if mockUpgrader.CallCount != 1 {
 		t.Fatalf("Unexpected number of calls to upgrader. Expected: %d, Actual: %d.", 1, mockUpgrader.CallCount)
@@ -226,7 +217,7 @@
 	// Upgrade schema version for underlying db using a different handle
 	schema2 := tu.DefaultSchema(1)
 	dbHandle2 := a.NoSQLDatabase("db1", schema2)
-	dbHandle2.UpgradeIfOutdated(ctx)
+	dbHandle2.EnforceSchema(ctx)
 
 	// Commit batch1, abort batch2, attempt writing a row using batch3.
 	// Each of these operations should fail.
diff --git a/v23/syncbase/util/util.go b/v23/syncbase/util/util.go
index 6eeaaaa..dda4c6d 100644
--- a/v23/syncbase/util/util.go
+++ b/v23/syncbase/util/util.go
@@ -64,6 +64,12 @@
 	return string(x)
 }
 
+// IsPrefix returns true if start and limit strings together represent a prefix
+// range. If true, start represents the prefix.
+func IsPrefix(start string, limit string) bool {
+	return PrefixRangeLimit(start) == limit
+}
+
 // AccessController provides access control for various syncbase objects.
 type AccessController interface {
 	// SetPermissions replaces the current Permissions for an object.
diff --git a/v23/syncbase/util/util_test.go b/v23/syncbase/util/util_test.go
index 321691a..2d6384a 100644
--- a/v23/syncbase/util/util_test.go
+++ b/v23/syncbase/util/util_test.go
@@ -61,3 +61,42 @@
 		}
 	}
 }
+
+func TestIsPrefix(t *testing.T) {
+	tests := []struct {
+		isPrefix bool
+		start  string
+		limit  string
+	}{
+		{true, "", ""},
+		{true, "a", "b"},
+		{true, "aa", "ab"},
+		{true, "\xfe", "\xff"},
+		{true, "a\xfe", "a\xff"},
+		{true, "aa\xfe", "aa\xff"},
+		{true, "a\xff", "b"},
+		{true, "aa\xff", "ab"},
+		{true, "a\xff\xff", "b"},
+		{true, "aa\xff\xff", "ab"},
+		{true, "\xff", ""},
+		{true, "\xff\xff", ""},
+
+		{false, "", "\x00"},
+		{false, "a", "aa"},
+		{false, "aa", "aa"},
+		{false, "\xfe", "\x00"},
+		{false, "a\xfe", "b\xfe"},
+		{false, "aa\xfe", "aa\x00"},
+		{false, "a\xff", "b\x00"},
+		{false, "aa\xff", "ab\x00"},
+		{false, "a\xff\xff", "a\xff\xff\xff"},
+		{false, "aa\xff\xff", "a"},
+		{false, "\xff", "\x00"},
+	}
+	for _, test := range tests {
+		result := util.IsPrefix(test.start, test.limit)
+		if result != test.isPrefix {
+			t.Errorf("%q, %q: got %v, want %v", test.start, test.limit, result, test.isPrefix)
+		}
+	}
+}
diff --git a/x/ref/services/syncbase/server/nosql/database.go b/x/ref/services/syncbase/server/nosql/database.go
index 2818a8c..a9f503f 100644
--- a/x/ref/services/syncbase/server/nosql/database.go
+++ b/x/ref/services/syncbase/server/nosql/database.go
@@ -49,6 +49,11 @@
 	mu  sync.Mutex // protects the fields below
 	sns map[uint64]store.Snapshot
 	txs map[uint64]store.Transaction
+
+	// Active ConflictResolver connection from the app to this database.
+	// NOTE: For now, we assume there's only one open conflict resolution stream
+	// per database (typically, from the app that owns the database).
+	resolver wire.ConflictManagerStartConflictResolverServerCall
 }
 
 // databaseReq is a per-request object that handles Database RPCs.
diff --git a/x/ref/services/syncbase/server/nosql/database_crm.go b/x/ref/services/syncbase/server/nosql/database_crm.go
new file mode 100644
index 0000000..e1135a9
--- /dev/null
+++ b/x/ref/services/syncbase/server/nosql/database_crm.go
@@ -0,0 +1,20 @@
+// Copyright 2015 The Vanadium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package nosql
+
+import (
+	wire "v.io/syncbase/v23/services/syncbase/nosql"
+	"v.io/v23/context"
+)
+
+////////////////////////////////////////
+// ConflictManager RPC methods
+
+func (d *databaseReq) StartConflictResolver(ctx *context.T, call wire.ConflictManagerStartConflictResolverServerCall) error {
+	// Store the conflict resolver connection in the per-app, per-database
+	// singleton so that sync can access it.
+	d.database.resolver = call
+	return nil
+}